373 lines
15 KiB
Python
373 lines
15 KiB
Python
"""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")
|