"""Integration tests for the department admin routes (U2). Uses FastAPI TestClient with a test app that mounts only the new ``admin_router`` from ``routes.admin``. The ``_require_admin`` dependency is overridden via ``app.dependency_overrides`` so the tests don't need real JWTs — they can simulate admin and non-admin callers directly. """ from __future__ import annotations import sqlite3 import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any import pytest from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from agentkit.server.auth.models import init_auth_db from agentkit.server.routes import admin as admin_routes_module # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture async def tmp_auth_db(tmp_path: Path) -> Path: db_path = tmp_path / "admin_departments.db" await init_auth_db(db_path) return db_path def _make_admin_user() -> dict[str, Any]: return {"user_id": "admin-1", "username": "admin", "role": "admin"} @pytest.fixture def admin_app(tmp_auth_db: Path) -> FastAPI: """A minimal FastAPI app with only the department admin router mounted. The ``_require_admin`` dependency is overridden to return a fake admin user. Individual tests can re-override it to simulate a non-admin. """ app = FastAPI() app.state.auth_db_path = str(tmp_auth_db) app.include_router(admin_routes_module.admin_router, prefix="/api/v1") # Default: allow admin access. app.dependency_overrides[admin_routes_module._require_admin] = lambda: _make_admin_user() return app @pytest.fixture def admin_client(admin_app: FastAPI) -> TestClient: return TestClient(admin_app) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _insert_user(db_path: Path, *, user_id: str | None = None) -> str: """Insert a minimal user row synchronously (for test setup).""" user_id = user_id or str(uuid.uuid4()) now_iso = datetime.now(timezone.utc).isoformat() with sqlite3.connect(str(db_path)) as db: db.execute( "INSERT INTO users " "(id, username, email, password_hash, role, is_active, " " is_terminal_authorized, is_server_terminal_authorized, " " created_at, updated_at) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( user_id, f"user-{user_id[:8]}", f"{user_id[:8]}@example.com", "$2b$12$placeholder.hash.placeholder.hash.placeholder.hash", "member", 1, 0, 0, now_iso, now_iso, ), ) db.commit() return user_id def _assign_user_to_department(db_path: Path, user_id: str, department_id: str) -> None: """Insert a user_departments row synchronously (for test setup).""" with sqlite3.connect(str(db_path)) as db: db.execute( "INSERT INTO user_departments (user_id, department_id, created_at) VALUES (?, ?, ?)", (user_id, department_id, datetime.now(timezone.utc).isoformat()), ) db.commit() def _create_department(client: TestClient, name: str, description: str = "") -> dict: resp = client.post( "/api/v1/admin/departments", json={"name": name, "description": description}, ) assert resp.status_code == 201, resp.text return resp.json() # --------------------------------------------------------------------------- # Department CRUD # --------------------------------------------------------------------------- class TestCreateDepartment: def test_create_returns_201_with_department_dict(self, admin_client: TestClient): resp = admin_client.post( "/api/v1/admin/departments", json={"name": "Engineering", "description": "Eng team"}, ) assert resp.status_code == 201 body = resp.json() assert body["id"] assert body["name"] == "Engineering" assert body["description"] == "Eng team" assert body["is_active"] is True def test_create_duplicate_name_returns_409(self, admin_client: TestClient): _create_department(admin_client, "Engineering") resp = admin_client.post( "/api/v1/admin/departments", json={"name": "Engineering"}, ) assert resp.status_code == 409 def test_non_admin_returns_403(self, admin_app: FastAPI): """When _require_admin raises 403, the endpoint returns 403.""" admin_app.dependency_overrides[admin_routes_module._require_admin] = _raise_forbidden client = TestClient(admin_app) resp = client.post( "/api/v1/admin/departments", json={"name": "Engineering"}, ) assert resp.status_code == 403 class TestListDepartments: def test_list_returns_all_departments(self, admin_client: TestClient): _create_department(admin_client, "Engineering") _create_department(admin_client, "HR") resp = admin_client.get("/api/v1/admin/departments") assert resp.status_code == 200 names = {d["name"] for d in resp.json()} assert names == {"Engineering", "HR"} def test_list_exclude_inactive(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/disable") resp = admin_client.get("/api/v1/admin/departments", params={"include_inactive": False}) assert resp.status_code == 200 assert resp.json() == [] class TestGetDepartment: def test_get_returns_department_by_id(self, admin_client: TestClient): created = _create_department(admin_client, "Engineering") resp = admin_client.get(f"/api/v1/admin/departments/{created['id']}") assert resp.status_code == 200 assert resp.json()["name"] == "Engineering" def test_get_unknown_id_returns_404(self, admin_client: TestClient): resp = admin_client.get(f"/api/v1/admin/departments/{uuid.uuid4()}") assert resp.status_code == 404 class TestUpdateDepartment: def test_update_name_and_description(self, admin_client: TestClient): created = _create_department(admin_client, "Engineering", "Old") resp = admin_client.patch( f"/api/v1/admin/departments/{created['id']}", json={"name": "Eng", "description": "New"}, ) assert resp.status_code == 200 body = resp.json() assert body["name"] == "Eng" assert body["description"] == "New" def test_update_duplicate_name_returns_409(self, admin_client: TestClient): _create_department(admin_client, "Engineering") hr = _create_department(admin_client, "HR") resp = admin_client.patch( f"/api/v1/admin/departments/{hr['id']}", json={"name": "Engineering"}, ) assert resp.status_code == 409 def test_update_unknown_id_returns_404(self, admin_client: TestClient): resp = admin_client.patch( f"/api/v1/admin/departments/{uuid.uuid4()}", json={"name": "X"}, ) assert resp.status_code == 404 class TestDisableEnableDepartment: def test_disable_sets_is_active_false(self, admin_client: TestClient): created = _create_department(admin_client, "Engineering") resp = admin_client.post(f"/api/v1/admin/departments/{created['id']}/disable") assert resp.status_code == 200 assert resp.json()["is_active"] is False def test_enable_sets_is_active_true(self, admin_client: TestClient): created = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{created['id']}/disable") resp = admin_client.post(f"/api/v1/admin/departments/{created['id']}/enable") assert resp.status_code == 200 assert resp.json()["is_active"] is True def test_disable_unknown_id_returns_404(self, admin_client: TestClient): resp = admin_client.post(f"/api/v1/admin/departments/{uuid.uuid4()}/disable") assert resp.status_code == 404 class TestDeleteDepartment: def test_delete_removes_department(self, admin_client: TestClient): created = _create_department(admin_client, "Engineering") resp = admin_client.delete(f"/api/v1/admin/departments/{created['id']}") assert resp.status_code == 200 assert resp.json() == {"deleted": True} # Confirm it's gone. assert admin_client.get(f"/api/v1/admin/departments/{created['id']}").status_code == 404 def test_delete_unknown_id_returns_404(self, admin_client: TestClient): resp = admin_client.delete(f"/api/v1/admin/departments/{uuid.uuid4()}") assert resp.status_code == 404 def test_delete_department_with_users_returns_400( self, admin_client: TestClient, tmp_auth_db: Path ): created = _create_department(admin_client, "Engineering") # Insert a user and assign to the department directly in the DB # (synchronous sqlite3 — no event-loop mixing with TestClient). user_id = _insert_user(tmp_auth_db) _assign_user_to_department(tmp_auth_db, user_id, created["id"]) resp = admin_client.delete(f"/api/v1/admin/departments/{created['id']}") assert resp.status_code == 400 assert "users" in resp.json()["detail"].lower() # --------------------------------------------------------------------------- # Skill bindings # --------------------------------------------------------------------------- class TestSkillBindings: def test_bind_skill_returns_201(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") resp = admin_client.post(f"/api/v1/admin/departments/{dept['id']}/skills/code_review") assert resp.status_code == 201 body = resp.json() assert body["skill_name"] == "code_review" assert body["department_id"] == dept["id"] def test_bind_duplicate_skill_returns_409(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/skills/code_review") resp = admin_client.post(f"/api/v1/admin/departments/{dept['id']}/skills/code_review") assert resp.status_code == 409 def test_list_department_skills(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/skills/code_review") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/skills/web_search") resp = admin_client.get(f"/api/v1/admin/departments/{dept['id']}/skills") assert resp.status_code == 200 assert resp.json() == ["code_review", "web_search"] def test_unbind_skill(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/skills/code_review") resp = admin_client.delete(f"/api/v1/admin/departments/{dept['id']}/skills/code_review") assert resp.status_code == 200 assert resp.json() == {"unbound": True} # Confirm it's gone. assert admin_client.get(f"/api/v1/admin/departments/{dept['id']}/skills").json() == [] def test_unbind_skill_idempotent(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") resp = admin_client.delete(f"/api/v1/admin/departments/{dept['id']}/skills/never_bound") assert resp.status_code == 200 assert resp.json() == {"unbound": True} # --------------------------------------------------------------------------- # KB bindings # --------------------------------------------------------------------------- class TestKbBindings: def test_bind_kb_returns_201(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") resp = admin_client.post(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-1") assert resp.status_code == 201 body = resp.json() assert body["kb_source_id"] == "kb-source-1" assert body["department_id"] == dept["id"] def test_bind_duplicate_kb_returns_409(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-1") resp = admin_client.post(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-1") assert resp.status_code == 409 def test_list_department_kbs(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-2") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-1") resp = admin_client.get(f"/api/v1/admin/departments/{dept['id']}/kb") assert resp.status_code == 200 assert resp.json() == ["kb-source-1", "kb-source-2"] def test_unbind_kb(self, admin_client: TestClient): dept = _create_department(admin_client, "Engineering") admin_client.post(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-1") resp = admin_client.delete(f"/api/v1/admin/departments/{dept['id']}/kb/kb-source-1") assert resp.status_code == 200 assert resp.json() == {"unbound": True} assert admin_client.get(f"/api/v1/admin/departments/{dept['id']}/kb").json() == [] # --------------------------------------------------------------------------- # Non-admin access # --------------------------------------------------------------------------- class TestNonAdminAccess: """All department endpoints must return 403 for non-admin users.""" def test_non_admin_cannot_create(self, admin_app: FastAPI): admin_app.dependency_overrides[admin_routes_module._require_admin] = _raise_forbidden client = TestClient(admin_app) resp = client.post( "/api/v1/admin/departments", json={"name": "Engineering"}, ) assert resp.status_code == 403 def test_non_admin_cannot_list(self, admin_app: FastAPI): admin_app.dependency_overrides[admin_routes_module._require_admin] = _raise_forbidden client = TestClient(admin_app) resp = client.get("/api/v1/admin/departments") assert resp.status_code == 403 def test_non_admin_cannot_delete(self, admin_app: FastAPI): admin_app.dependency_overrides[admin_routes_module._require_admin] = _raise_forbidden client = TestClient(admin_app) resp = client.delete(f"/api/v1/admin/departments/{uuid.uuid4()}") assert resp.status_code == 403 # --------------------------------------------------------------------------- # Helpers for non-admin simulation # --------------------------------------------------------------------------- def _raise_forbidden() -> dict[str, Any]: """Dependency override that simulates a non-admin (403) response.""" raise HTTPException(status_code=403, detail="Admin permission required")