miem_workers/tests/test_api_mcp.py

from datetime import datetime, timezone

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
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():
    client = TestClient(app)

    response = client.get("/api/health")

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


def test_mcp_lists_tools_without_auth_and_ignores_auth_header():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)

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

    app.dependency_overrides[get_db] = override_db
    client = TestClient(app)

    without_auth = client.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}})
    with_auth = client.post(
        "/mcp",
        headers={"Authorization": "Bearer anything"},
        json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
    )

    assert without_auth.status_code == 200
    assert with_auth.status_code == 200
    assert without_auth.json()["result"]["tools"][0]["name"] == "search_employees"
    assert any(tool["name"] == "get_crawl_run_details" for tool in without_auth.json()["result"]["tools"])
    assert with_auth.json()["result"]["tools"] == without_auth.json()["result"]["tools"]

    app.dependency_overrides.clear()


def test_mcp_search_employees_returns_matching_employee():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    session.add(
        Employee(
            profile_key="staff:avsergeev",
            profile_type="staff",
            profile_id="avsergeev",
            canonical_url="https://www.hse.ru/staff/avsergeev",
            full_name="Сергеев Алексей Викторович",
            status="active",
            first_seen_at=datetime.now(timezone.utc),
            last_seen_at=datetime.now(timezone.utc),
            current_data={"sections": []},
        )
    )
    session.commit()
    session.close()

    def override_db():
        db = Session()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_db
    client = TestClient(app)

    response = client.post(
        "/mcp",
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "search_employees", "arguments": {"query": "Сергеев"}},
        },
    )

    assert response.status_code == 200
    assert "Сергеев Алексей Викторович" in response.json()["result"]["content"][0]["text"]

    app.dependency_overrides.clear()


def test_mcp_get_crawl_run_details_returns_changes():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    run = CrawlRun(source_url="https://miem.hse.ru/persons", status="completed", new_count=1)
    employee = Employee(
        profile_key="staff:new",
        profile_type="staff",
        profile_id="new",
        canonical_url="https://www.hse.ru/staff/new",
        full_name="New Person",
        status="active",
        first_seen_at=datetime.now(timezone.utc),
        last_seen_at=datetime.now(timezone.utc),
    )
    session.add_all([run, employee])
    session.commit()
    session.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",
        )
    )
    session.commit()
    run_id = run.id
    session.close()

    def override_db():
        db = Session()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_db
    client = TestClient(app)

    response = client.post(
        "/mcp",
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": "get_crawl_run_details", "arguments": {"run_id": run_id}},
        },
    )

    assert response.status_code == 200
    text = response.json()["result"]["content"][0]["text"]
    assert "New Person" in text
    assert "changes_detail_available" in text

    app.dependency_overrides.clear()


def test_mcp_protected_resource_metadata_route_is_removed():
    client = TestClient(app)

    response = client.get("/.well-known/oauth-protected-resource")

    assert response.status_code == 404


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 = sessionmaker(bind=engine)
    db = Session()
    db.add(
        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(run)
    db.commit()
    db.add(
        CrawlRunEmployeeChange(
            crawl_run_id=run.id,
            employee_id=1,
            profile_key="staff:alpha",
            profile_url="https://www.hse.ru/staff/alpha",
            full_name="Alpha Person",
            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()
        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"

    app.dependency_overrides.clear()