131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, select
|
|
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.7.9"
|
|
|
|
|
|
def test_mcp_endpoint_is_removed():
|
|
assert TestClient(app).get("/mcp").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_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"
|
|
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()
|