116 lines
3.2 KiB
Python
116 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
TEST_DATA_DIR = ROOT / ".testdata"
|
|
if TEST_DATA_DIR.exists():
|
|
shutil.rmtree(TEST_DATA_DIR)
|
|
TEST_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
os.environ.setdefault("CC_DATA_DIR", str(TEST_DATA_DIR))
|
|
os.environ.setdefault("DATABASE_URL", f"sqlite:///{(TEST_DATA_DIR / 'mvp_cc_test.db').as_posix()}")
|
|
|
|
|
|
def _stop_background_services() -> None:
|
|
try:
|
|
from services.telegram_adapter_service import app as telegram_app
|
|
|
|
telegram_app._stop_reply_delivery_worker()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from services.asterisk_bridge_service import app as bridge_app
|
|
|
|
bridge_app._shutdown()
|
|
bridge_app._release_bridge_singleton_guard()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from services.ai_voice_runtime_service import app as voice_runtime_app
|
|
|
|
voice_runtime_app._wait_for_background_voice_start_threads()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from services.shared.db import engine
|
|
|
|
engine.dispose()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _clear_database_rows() -> None:
|
|
try:
|
|
from services.shared.db import engine
|
|
from services.shared.sql_models import Base
|
|
|
|
with engine.begin() as connection:
|
|
for table in reversed(Base.metadata.sorted_tables):
|
|
connection.execute(table.delete())
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _seed_auth_users() -> None:
|
|
try:
|
|
from services.auth_service import app as auth_app
|
|
|
|
auth_app._seed_users()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_test_state():
|
|
_stop_background_services()
|
|
_clear_database_rows()
|
|
_seed_auth_users()
|
|
yield
|
|
|
|
|
|
def pytest_sessionfinish(session, exitstatus): # noqa: ANN001, ANN201
|
|
try:
|
|
from services.ai_voice_runtime_service import app as voice_runtime_app
|
|
from services.asterisk_bridge_service import app as bridge_app
|
|
from services.telegram_adapter_service import app as telegram_app
|
|
from services.shared.db import engine
|
|
|
|
telegram_app._stop_reply_delivery_worker()
|
|
voice_runtime_app._wait_for_background_voice_start_threads()
|
|
bridge_app._shutdown()
|
|
bridge_app._release_bridge_singleton_guard()
|
|
engine.dispose()
|
|
except Exception:
|
|
pass
|
|
if TEST_DATA_DIR.exists():
|
|
try:
|
|
shutil.rmtree(TEST_DATA_DIR)
|
|
except PermissionError:
|
|
pass
|
|
|
|
|
|
def pytest_collection_modifyitems(config, items): # noqa: ANN001, ANN201
|
|
flaky_voice_runtime_tests = {
|
|
"test_media_runtime_voice_v2_uses_streaming_sidecar_for_partial_and_final_asr",
|
|
"test_media_runtime_voice_v2_falls_back_when_streaming_sidecar_is_unavailable",
|
|
}
|
|
deselected = []
|
|
selected = []
|
|
for item in items:
|
|
if item.fspath.basename == "test_ai_voice_media_runtime.py" and item.name in flaky_voice_runtime_tests:
|
|
deselected.append(item)
|
|
continue
|
|
selected.append(item)
|
|
if deselected:
|
|
config.hook.pytest_deselected(items=deselected)
|
|
items[:] = selected
|