"""Unit tests for DepartmentService (U2 — department CRUD + bindings). Covers: - Happy path: create → list → get → update → disable → enable → delete - Happy path: bind_skill → list_skills → unbind_skill - Happy path: bind_kb → list_kbs → unbind_kb - Edge case: create duplicate name → ValueError - Edge case: delete department with users → ValueError - Edge case: update to duplicate name → ValueError - Edge case: get/update/delete non-existent department → None/ValueError/False - Edge case: bind duplicate skill → ValueError - Edge case: count_department_users returns correct count """ from __future__ import annotations import uuid from datetime import datetime, timezone from pathlib import Path import aiosqlite import pytest from agentkit.server.admin.department_service import DepartmentService from agentkit.server.auth.models import init_auth_db # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture async def fresh_db(tmp_path: Path) -> Path: """A brand-new auth DB on a fresh path (no data).""" db_path = tmp_path / "auth.db" await init_auth_db(db_path) return db_path @pytest.fixture def service() -> DepartmentService: return DepartmentService() def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() async def _insert_user(db_path: Path, *, user_id: str | None = None) -> str: """Insert a minimal user row and return its id.""" user_id = user_id or str(uuid.uuid4()) now_iso = _now_iso() async with aiosqlite.connect(str(db_path)) as db: await 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, ), ) await db.commit() return user_id async def _assign_user_to_department(db_path: Path, user_id: str, department_id: str) -> None: """Insert a user_departments row.""" async with aiosqlite.connect(str(db_path)) as db: await db.execute( "INSERT INTO user_departments (user_id, department_id, created_at) VALUES (?, ?, ?)", (user_id, department_id, _now_iso()), ) await db.commit() # --------------------------------------------------------------------------- # Department CRUD happy path # --------------------------------------------------------------------------- class TestDepartmentCrudHappyPath: async def test_create_returns_department_dict(self, service: DepartmentService, fresh_db: Path): dept = await service.create_department(fresh_db, "Engineering", "Eng team") assert dept["id"] assert dept["name"] == "Engineering" assert dept["description"] == "Eng team" assert dept["is_active"] is True assert "created_at" in dept async def test_create_with_empty_description_stores_none( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "HR") assert dept["description"] is None async def test_list_returns_created_departments( self, service: DepartmentService, fresh_db: Path ): await service.create_department(fresh_db, "Engineering") await service.create_department(fresh_db, "HR") depts = await service.list_departments(fresh_db) assert len(depts) == 2 names = {d["name"] for d in depts} assert names == {"Engineering", "HR"} async def test_list_excludes_inactive_when_asked( self, service: DepartmentService, fresh_db: Path ): eng = await service.create_department(fresh_db, "Engineering") await service.set_department_active(fresh_db, eng["id"], is_active=False) active_only = await service.list_departments(fresh_db, include_inactive=False) assert len(active_only) == 0 all_depts = await service.list_departments(fresh_db, include_inactive=True) assert len(all_depts) == 1 async def test_get_returns_department_by_id(self, service: DepartmentService, fresh_db: Path): created = await service.create_department(fresh_db, "Engineering") fetched = await service.get_department(fresh_db, created["id"]) assert fetched is not None assert fetched["id"] == created["id"] assert fetched["name"] == "Engineering" async def test_get_returns_none_for_unknown_id( self, service: DepartmentService, fresh_db: Path ): result = await service.get_department(fresh_db, str(uuid.uuid4())) assert result is None async def test_update_name_and_description(self, service: DepartmentService, fresh_db: Path): created = await service.create_department(fresh_db, "Engineering", "Old desc") updated = await service.update_department( fresh_db, created["id"], name="Eng", description="New desc" ) assert updated["name"] == "Eng" assert updated["description"] == "New desc" async def test_update_partial_only_changes_provided_fields( self, service: DepartmentService, fresh_db: Path ): created = await service.create_department(fresh_db, "Engineering", "Old desc") # Only update name, description should be preserved. updated = await service.update_department(fresh_db, created["id"], name="Eng") assert updated["name"] == "Eng" assert updated["description"] == "Old desc" async def test_disable_sets_is_active_false(self, service: DepartmentService, fresh_db: Path): created = await service.create_department(fresh_db, "Engineering") disabled = await service.set_department_active(fresh_db, created["id"], is_active=False) assert disabled["is_active"] is False async def test_enable_sets_is_active_true(self, service: DepartmentService, fresh_db: Path): created = await service.create_department(fresh_db, "Engineering") await service.set_department_active(fresh_db, created["id"], is_active=False) enabled = await service.set_department_active(fresh_db, created["id"], is_active=True) assert enabled["is_active"] is True async def test_delete_removes_department(self, service: DepartmentService, fresh_db: Path): created = await service.create_department(fresh_db, "Engineering") deleted = await service.delete_department(fresh_db, created["id"]) assert deleted is True # Confirm it's gone. assert await service.get_department(fresh_db, created["id"]) is None async def test_delete_returns_false_for_unknown_id( self, service: DepartmentService, fresh_db: Path ): deleted = await service.delete_department(fresh_db, str(uuid.uuid4())) assert deleted is False # --------------------------------------------------------------------------- # Department CRUD edge cases # --------------------------------------------------------------------------- class TestDepartmentCrudEdgeCases: async def test_create_duplicate_name_raises_value_error( self, service: DepartmentService, fresh_db: Path ): await service.create_department(fresh_db, "Engineering") with pytest.raises(ValueError, match="already exists"): await service.create_department(fresh_db, "Engineering") async def test_update_to_duplicate_name_raises_value_error( self, service: DepartmentService, fresh_db: Path ): await service.create_department(fresh_db, "Engineering") hr = await service.create_department(fresh_db, "HR") with pytest.raises(ValueError, match="already exists"): await service.update_department(fresh_db, hr["id"], name="Engineering") async def test_update_nonexistent_raises_value_error( self, service: DepartmentService, fresh_db: Path ): with pytest.raises(ValueError, match="not found"): await service.update_department(fresh_db, str(uuid.uuid4()), name="X") async def test_set_active_nonexistent_raises_value_error( self, service: DepartmentService, fresh_db: Path ): with pytest.raises(ValueError, match="not found"): await service.set_department_active(fresh_db, str(uuid.uuid4()), is_active=False) async def test_delete_department_with_users_raises_value_error( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") user_id = await _insert_user(fresh_db) await _assign_user_to_department(fresh_db, user_id, dept["id"]) with pytest.raises(ValueError, match="Department has users"): await service.delete_department(fresh_db, dept["id"]) async def test_delete_department_after_removing_users_succeeds( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") user_id = await _insert_user(fresh_db) await _assign_user_to_department(fresh_db, user_id, dept["id"]) # Remove the user-department association directly. async with aiosqlite.connect(str(fresh_db)) as db: await db.execute( "DELETE FROM user_departments WHERE department_id = ?", (dept["id"],), ) await db.commit() # Now deletion should succeed. deleted = await service.delete_department(fresh_db, dept["id"]) assert deleted is True async def test_delete_cascades_bindings(self, service: DepartmentService, fresh_db: Path): """Deleting a department should also clear its skill/KB bindings.""" dept = await service.create_department(fresh_db, "Engineering") await service.bind_skill(fresh_db, dept["id"], "code_review") await service.bind_kb(fresh_db, dept["id"], "kb-source-1") await service.delete_department(fresh_db, dept["id"]) # Bindings should be gone. assert await service.list_department_skills(fresh_db, dept["id"]) == [] assert await service.list_department_kbs(fresh_db, dept["id"]) == [] # --------------------------------------------------------------------------- # Skill bindings # --------------------------------------------------------------------------- class TestSkillBindings: async def test_bind_skill_returns_binding_dict( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") binding = await service.bind_skill(fresh_db, dept["id"], "code_review") assert binding["department_id"] == dept["id"] assert binding["skill_name"] == "code_review" assert binding["id"] assert "created_at" in binding async def test_list_department_skills_returns_bound_names( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") await service.bind_skill(fresh_db, dept["id"], "code_review") await service.bind_skill(fresh_db, dept["id"], "web_search") skills = await service.list_department_skills(fresh_db, dept["id"]) assert skills == ["code_review", "web_search"] # sorted alphabetically async def test_unbind_skill_removes_binding(self, service: DepartmentService, fresh_db: Path): dept = await service.create_department(fresh_db, "Engineering") await service.bind_skill(fresh_db, dept["id"], "code_review") unbound = await service.unbind_skill(fresh_db, dept["id"], "code_review") assert unbound is True assert await service.list_department_skills(fresh_db, dept["id"]) == [] async def test_unbind_skill_returns_false_for_nonexistent( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") unbound = await service.unbind_skill(fresh_db, dept["id"], "never_bound") assert unbound is False async def test_bind_duplicate_skill_raises_value_error( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") await service.bind_skill(fresh_db, dept["id"], "code_review") with pytest.raises(ValueError, match="already bound"): await service.bind_skill(fresh_db, dept["id"], "code_review") async def test_bind_skill_to_nonexistent_department_raises( self, service: DepartmentService, fresh_db: Path ): with pytest.raises(ValueError, match="not found"): await service.bind_skill(fresh_db, str(uuid.uuid4()), "code_review") async def test_same_skill_name_in_different_departments( self, service: DepartmentService, fresh_db: Path ): dept_a = await service.create_department(fresh_db, "Engineering") dept_b = await service.create_department(fresh_db, "HR") await service.bind_skill(fresh_db, dept_a["id"], "shared_skill") await service.bind_skill(fresh_db, dept_b["id"], "shared_skill") assert len(await service.list_department_skills(fresh_db, dept_a["id"])) == 1 assert len(await service.list_department_skills(fresh_db, dept_b["id"])) == 1 # --------------------------------------------------------------------------- # KB bindings # --------------------------------------------------------------------------- class TestKbBindings: async def test_bind_kb_returns_binding_dict(self, service: DepartmentService, fresh_db: Path): dept = await service.create_department(fresh_db, "Engineering") binding = await service.bind_kb(fresh_db, dept["id"], "kb-source-1") assert binding["department_id"] == dept["id"] assert binding["kb_source_id"] == "kb-source-1" assert binding["id"] assert "created_at" in binding async def test_list_department_kbs_returns_bound_ids( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") await service.bind_kb(fresh_db, dept["id"], "kb-source-2") await service.bind_kb(fresh_db, dept["id"], "kb-source-1") kbs = await service.list_department_kbs(fresh_db, dept["id"]) assert kbs == ["kb-source-1", "kb-source-2"] # sorted alphabetically async def test_unbind_kb_removes_binding(self, service: DepartmentService, fresh_db: Path): dept = await service.create_department(fresh_db, "Engineering") await service.bind_kb(fresh_db, dept["id"], "kb-source-1") unbound = await service.unbind_kb(fresh_db, dept["id"], "kb-source-1") assert unbound is True assert await service.list_department_kbs(fresh_db, dept["id"]) == [] async def test_unbind_kb_returns_false_for_nonexistent( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") unbound = await service.unbind_kb(fresh_db, dept["id"], "never_bound") assert unbound is False async def test_bind_duplicate_kb_raises_value_error( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") await service.bind_kb(fresh_db, dept["id"], "kb-source-1") with pytest.raises(ValueError, match="already bound"): await service.bind_kb(fresh_db, dept["id"], "kb-source-1") async def test_bind_kb_to_nonexistent_department_raises( self, service: DepartmentService, fresh_db: Path ): with pytest.raises(ValueError, match="not found"): await service.bind_kb(fresh_db, str(uuid.uuid4()), "kb-source-1") # --------------------------------------------------------------------------- # count_department_users # --------------------------------------------------------------------------- class TestCountDepartmentUsers: async def test_count_returns_zero_for_empty_department( self, service: DepartmentService, fresh_db: Path ): dept = await service.create_department(fresh_db, "Engineering") count = await service.count_department_users(fresh_db, dept["id"]) assert count == 0 async def test_count_returns_correct_count(self, service: DepartmentService, fresh_db: Path): dept = await service.create_department(fresh_db, "Engineering") user_a = await _insert_user(fresh_db) user_b = await _insert_user(fresh_db) await _assign_user_to_department(fresh_db, user_a, dept["id"]) await _assign_user_to_department(fresh_db, user_b, dept["id"]) count = await service.count_department_users(fresh_db, dept["id"]) assert count == 2 async def test_count_returns_zero_for_unknown_department( self, service: DepartmentService, fresh_db: Path ): count = await service.count_department_users(fresh_db, str(uuid.uuid4())) assert count == 0 # --------------------------------------------------------------------------- # Singleton helpers # --------------------------------------------------------------------------- class TestSingletonHelpers: def test_get_department_service_returns_singleton(self): from agentkit.server.admin.department_service import ( get_department_service, set_department_service, ) # Save the original singleton so we don't disturb other tests. original = get_department_service() try: custom = DepartmentService() set_department_service(custom) assert get_department_service() is custom # Clearing falls back to a new lazy instance. set_department_service(None) new_one = get_department_service() assert new_one is not custom finally: set_department_service(original)