import ast import os import unittest from pathlib import Path from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool os.environ.setdefault("JWT_SECRET", "test-jwt-secret-value-with-32-chars") from app.api.auth import router as auth_router from app.core.database import Base, get_db from app.core.models import Tenant, TenantStatus, User, UserRole, UserStatus from app.core.security import hash_password, verify_password ROOT = Path(__file__).resolve().parents[2] class UsersAPIContractTests(unittest.TestCase): def setUp(self): self.engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) self.SessionLocal = sessionmaker(bind=self.engine) Base.metadata.create_all(self.engine, tables=[Tenant.__table__, User.__table__]) self.addCleanup(self.engine.dispose) self.app = FastAPI() self.app.include_router(auth_router) # Seed a default tenant with Session(self.engine) as s: t = Tenant(company_code="test", name="Test Co", status=TenantStatus.active) s.add(t) s.commit() s.refresh(t) self._tenant_id = t.id def override_get_db(): db = self.SessionLocal() try: yield db finally: db.close() self.app.dependency_overrides[get_db] = override_get_db self.addCleanup(self.app.dependency_overrides.clear) def seed_user(self, **overrides) -> User: defaults = { "tenant_id": self._tenant_id, "username": "admin_user", "password_hash": hash_password("admin123"), "display_name": "Admin User", "role": UserRole.admin, "status": UserStatus.active, "allowed_permissions_json": None, "denied_permissions_json": None, "allowed_routes_json": None, } defaults.update(overrides) with Session(self.engine) as session: user = User(**defaults) session.add(user) session.commit() session.refresh(user) return user def make_client(self) -> TestClient: try: from app.api.users import router as users_router except ModuleNotFoundError as exc: self.fail(f"Expected app.api.users router to exist: {exc}") self.app.include_router(users_router) return TestClient(self.app) def auth_headers(self, client: TestClient, username: str, password: str) -> dict[str, str]: response = client.post( "/api/auth/login", json={"company_code": "test", "username": username, "password": password}, ) self.assertEqual(200, response.status_code, response.text) token = response.json()["access_token"] return {"Authorization": f"Bearer {token}"} def test_users_api_supports_crud_reset_and_permission_fields(self): from app.api.users import router as users_router route_methods = {} for route in users_router.routes: route_methods.setdefault(route.path, set()).update(route.methods or set()) self.assertIn("GET", route_methods["/api/users"]) self.assertIn("POST", route_methods["/api/users"]) self.assertIn("PUT", route_methods["/api/users/{user_id}"]) self.assertIn("POST", route_methods["/api/users/{user_id}/reset-password"]) main_tree = ast.parse((ROOT / "backend/app/main.py").read_text(encoding="utf-8")) include_users_router = any( isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "include_router" and node.args and isinstance(node.args[0], ast.Name) and node.args[0].id == "users_router" for node in ast.walk(main_tree) ) self.assertTrue(include_users_router) def test_users_api_enforces_admin_and_supports_create_update_and_reset(self): self.seed_user() self.seed_user( username="staff_user", password_hash=hash_password("staff123"), role=UserRole.staff, display_name="Staff User", ) client = self.make_client() staff_headers = self.auth_headers(client, "staff_user", "staff123") admin_headers = self.auth_headers(client, "admin_user", "admin123") forbidden = client.get("/api/users", headers=staff_headers) self.assertEqual(403, forbidden.status_code) blank_username = client.post( "/api/users", headers=admin_headers, json={ "username": " ", "password": "viewer123", "display_name": "Viewer User", "role": "viewer", }, ) self.assertEqual(422, blank_username.status_code) duplicate_username = client.post( "/api/users", headers=admin_headers, json={ "username": "admin_user", "password": "viewer123", "display_name": "Viewer User", "role": "viewer", }, ) self.assertEqual(400, duplicate_username.status_code) self.assertEqual("Username already exists", duplicate_username.json()["detail"]) create_response = client.post( "/api/users", headers=admin_headers, json={ "username": "viewer_user", "password": "viewer123", "display_name": "Viewer User", "role": "viewer", "status": "active", "allowed_permissions": ["invoices.send"], "denied_permissions": ["finance.read"], "allowed_routes": ["Dashboard", "Invoices"], }, ) self.assertEqual(201, create_response.status_code, create_response.text) created = create_response.json() self.assertEqual("viewer_user", created["username"]) self.assertEqual("viewer", created["role"]) self.assertEqual(["invoices.send"], created["allowed_permissions"]) self.assertEqual(["finance.read"], created["denied_permissions"]) self.assertEqual(["Dashboard", "Invoices"], created["allowed_routes"]) self.assertIn("customers.read", created["permissions"]) self.assertIn("invoices.send", created["permissions"]) self.assertNotIn("finance.read", created["permissions"]) list_response = client.get("/api/users", headers=admin_headers) self.assertEqual(200, list_response.status_code, list_response.text) usernames = [item["username"] for item in list_response.json()["items"]] self.assertIn("viewer_user", usernames) updated_disabled = client.put( f"/api/users/{created['id']}", headers=admin_headers, json={ "display_name": "Viewer User Updated", "role": "staff", "status": "disabled", "allowed_permissions": ["settings.read"], "denied_permissions": ["expenses.write"], "allowed_routes": ["Dashboard", "Customers"], }, ) self.assertEqual(200, updated_disabled.status_code, updated_disabled.text) disabled_payload = updated_disabled.json() self.assertEqual("Viewer User Updated", disabled_payload["display_name"]) self.assertEqual("staff", disabled_payload["role"]) self.assertEqual("disabled", disabled_payload["status"]) self.assertIn("settings.read", disabled_payload["permissions"]) self.assertNotIn("expenses.write", disabled_payload["permissions"]) disabled_login = client.post( "/api/auth/login", json={"company_code": "test", "username": "viewer_user", "password": "viewer123"}, ) self.assertEqual(403, disabled_login.status_code) updated_active = client.put( f"/api/users/{created['id']}", headers=admin_headers, json={ "status": "active", }, ) self.assertEqual(200, updated_active.status_code, updated_active.text) self.assertEqual("active", updated_active.json()["status"]) with Session(self.engine) as session: db_user = session.get(User, created["id"]) self.assertIsNotNone(db_user) self.assertTrue(verify_password("viewer123", db_user.password_hash)) reset_response = client.post( f"/api/users/{created['id']}/reset-password", headers=admin_headers, json={"password": "newpass456"}, ) self.assertEqual(200, reset_response.status_code, reset_response.text) self.assertEqual({"message": "Password reset successfully"}, reset_response.json()) old_login = client.post( "/api/auth/login", json={"company_code": "test", "username": "viewer_user", "password": "viewer123"}, ) self.assertEqual(401, old_login.status_code) new_login = client.post( "/api/auth/login", json={"company_code": "test", "username": "viewer_user", "password": "newpass456"}, ) self.assertEqual(200, new_login.status_code, new_login.text) with Session(self.engine) as session: db_user = session.get(User, created["id"]) self.assertIsNotNone(db_user) self.assertTrue(verify_password("newpass456", db_user.password_hash)) if __name__ == "__main__": unittest.main()