Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import os | |
| import sys | |
| from pathlib import Path | |
| os.environ.setdefault("PYTHONUTF8", "1") | |
| ROOT = Path(__file__).resolve().parent.parent | |
| sys.path.insert(0, str(ROOT)) | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from sqlalchemy import create_engine | |
| from sqlalchemy.orm import sessionmaker | |
| from sqlalchemy.pool import StaticPool | |
| from app.db.models import Base | |
| from app.deps import get_db | |
| from app.main import app | |
| class _StubClassifier: | |
| def predict_proba(self, text: str) -> float: | |
| fraud_markers = ( | |
| "срочно", "код из смс", "будет заблокирован", "ұтып алдыңыз", | |
| "шұғыл", "sms кодын", "бұғатталады", "уголовное", "оформлен кредит", | |
| "несие рәсімделді", | |
| ) | |
| t = text.lower() | |
| hits = sum(1 for m in fraud_markers if m in t) | |
| return min(0.95, 0.3 + 0.2 * hits) if hits else 0.05 | |
| def test_engine(): | |
| eng = create_engine( | |
| "sqlite://", | |
| connect_args={"check_same_thread": False}, | |
| poolclass=StaticPool, | |
| ) | |
| Base.metadata.create_all(eng) | |
| return eng | |
| def TestSession(test_engine): | |
| return sessionmaker(bind=test_engine, autoflush=False, autocommit=False, expire_on_commit=False) | |
| def client(TestSession): | |
| def _get_db_override(): | |
| db = TestSession() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |
| app.dependency_overrides[get_db] = _get_db_override | |
| app.state.clf = _StubClassifier() | |
| app.state.whisper = None | |
| with TestClient(app) as c: | |
| yield c | |
| app.dependency_overrides.clear() | |
| app.state.clf = None | |
| def device_header() -> dict[str, str]: | |
| return {"X-Device-Id": "33333333-3333-3333-3333-333333333333"} | |
| def other_device_header() -> dict[str, str]: | |
| return {"X-Device-Id": "44444444-4444-4444-4444-444444444444"} | |