miem_workers/tests/test_api.py

from datetime import datetime, timezone
from types import SimpleNamespace

from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.config import Settings, get_settings
from app.db import Base, get_db
from app.main import app
from app.models import CrawlRun, CrawlRunEmployeeChange, Employee
from app.security import SESSION_COOKIE, sign_session


def test_health_returns_versions():
    response = TestClient(app).get("/api/health")

    assert response.status_code == 200
    assert response.json()["backend_version"] == "0.8.7"


def test_api_employees_and_stats_require_admin_session():
    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine)
    session_factory = sessionmaker(bind=engine)
    db = session_factory()
    employee = Employee(
        profile_key="staff:alpha",
        profile_type="staff",
        profile_id="alpha",
        canonical_url="https://www.hse.ru/staff/alpha",
        full_name="Alpha Person",
        status="active",
        first_seen_at=datetime.now(timezone.utc),
        last_seen_at=datetime.now(timezone.utc),
        current_data={"contacts": {"emails": ["alpha@hse.ru"]}, "sections": []},
    )
    run = CrawlRun(source_url="https://miem.hse.ru/persons", status="completed", new_count=1)
    db.add_all([employee, run])
    db.commit()
    db.add(CrawlRunEmployeeChange(
        crawl_run_id=run.id,
        employee_id=employee.id,
        profile_key=employee.profile_key,
        profile_url=employee.canonical_url,
        full_name=employee.full_name,
        change_type="new",
        profile_available=True,
        message="added",
    ))
    db.commit()
    run_id = run.id
    db.close()

    settings = Settings(admin_username="admin", admin_password="password", session_secret="session-secret")

    def override_db():
        session = session_factory()
        try:
            yield session
        finally:
            session.close()

    app.dependency_overrides[get_db] = override_db
    app.dependency_overrides[get_settings] = lambda: settings
    client = TestClient(app)
    client.cookies.set(SESSION_COOKIE, sign_session("admin", settings))

    employees = client.get("/api/employees", params={"q": "Alpha", "has_email": True})
    stats = client.get("/api/stats")
    run_details = client.get(f"/api/crawl-runs/{run_id}")

    assert employees.status_code == 200
    assert employees.json()["total"] == 1
    assert stats.status_code == 200
    assert stats.json()["new_in_last_run"] == 1
    assert run_details.status_code == 200
    assert run_details.json()["changes"]["new"][0]["full_name"] == "Alpha Person"
    directory = client.get("/admin/directory?q=Alpha&offset=100")
    assert directory.status_code == 200
    assert "Показаны 1–1 из 1" in directory.text
    assert 'name="offset"' not in directory.text
    assert "Alpha Person" in directory.text
    empty = client.get("/admin/directory?q=missing&offset=100")
    assert "Показаны 0–0 из 0" in empty.text
    assert "Сбросить фильтры" in empty.text
    app.dependency_overrides.clear()


def test_admin_refresh_employee_route_updates_only_requested_employee(monkeypatch):
    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine)
    session_factory = sessionmaker(bind=engine)
    db = session_factory()
    db.add(Employee(
        profile_key="org_person:133709486",
        profile_type="org_person",
        profile_id="133709486",
        canonical_url="https://www.hse.ru/org/persons/133709486",
        full_name="Будков Юрий Алексеевич",
        status="active",
    ))
    db.commit()
    employee_id = db.scalar(select(Employee.id))
    db.close()

    settings = Settings(admin_username="admin", admin_password="password", session_secret="session-secret")

    def override_db():
        session = session_factory()
        try:
            yield session
        finally:
            session.close()

    calls = []

    def fake_refresh_employee(db, refreshed_employee, route_settings):
        calls.append((refreshed_employee.id, route_settings))
        return SimpleNamespace(status="completed")

    app.dependency_overrides[get_db] = override_db
    app.dependency_overrides[get_settings] = lambda: settings
    monkeypatch.setattr("app.admin.refresh_employee", fake_refresh_employee)
    client = TestClient(app)
    client.cookies.set(SESSION_COOKIE, sign_session("admin", settings))

    response = client.post(f"/admin/employees/{employee_id}/refresh", follow_redirects=False)

    assert response.status_code == 303
    assert response.headers["location"] == f"/admin/employees/{employee_id}?refresh_status=success"
    assert calls == [(employee_id, settings)]
    app.dependency_overrides.clear()


def test_directory_database_error_has_retry_and_preserves_filters(db_session, monkeypatch):
    settings = Settings(_env_file=None, session_secret="test-session-secret")
    app.dependency_overrides[get_db] = lambda: db_session
    app.dependency_overrides[get_settings] = lambda: settings
    def unavailable(*args, **kwargs):
        raise OperationalError("SELECT", {}, Exception("private database details"))
    monkeypatch.setattr("app.admin.list_employees_page", unavailable)
    try:
        client = TestClient(app)
        client.cookies.set(SESSION_COOKIE, sign_session(settings.admin_username, settings))
        response = client.get("/admin/directory?q=Alice")
        assert response.status_code == 503
        assert 'href="/admin/directory?q=Alice"' in response.text
        assert "Повторить загрузку" in response.text
        assert "private database details" not in response.text
    finally:
        app.dependency_overrides.clear()