from __future__ import annotations from uuid import uuid4 import pytest from fastapi import HTTPException, status from dependencies import ( require_any_role, require_authenticated_user, require_current_user, require_project_resource, require_project_scenario_resource, require_roles, require_scenario_resource, ) from models import Project, Scenario, User from services.session import AuthSession, SessionTokens from tests.utils.security import random_password, random_token @pytest.fixture() def uow(unit_of_work_factory): with unit_of_work_factory() as uow: assert uow.users and uow.roles and uow.projects and uow.scenarios uow.ensure_default_roles() yield uow def _unique(prefix: str) -> str: return f"{prefix}-{uuid4().hex}" def _create_user( uow, *, roles: tuple[str, ...] = (), is_active: bool = True, is_superuser: bool = False, ) -> User: user = User( email=f"{_unique('user')}@example.com", username=_unique('user'), password_hash=User.hash_password(random_password()), is_active=is_active, is_superuser=is_superuser, ) assert uow.users and uow.roles uow.users.create(user) uow.flush() for role_name in roles: role = uow.roles.get_by_name(role_name) if role is None: # pragma: no cover - defensive for missing roles raise AssertionError(f"Role {role_name} expected in test database") uow.users.assign_role(user_id=user.id, role_id=role.id) return uow.users.get(user.id, with_roles=True) def _create_project(uow) -> Project: assert uow.projects project = Project(name=_unique('project'), location="Test Site") uow.projects.create(project) uow.flush() return project def _create_scenario(uow, project: Project) -> Scenario: assert uow.scenarios scenario = Scenario(project_id=project.id, name=_unique('scenario')) uow.scenarios.create(scenario) uow.flush() return scenario def test_require_current_user_returns_authenticated_user(): user = User( email="user@example.com", username="user", password_hash=User.hash_password(random_password()), is_active=True, ) session = AuthSession( tokens=SessionTokens(access_token=random_token(), refresh_token=None) ) session.user = user result = require_current_user(session=session) assert result is user def test_require_current_user_raises_when_session_missing(): anonymous_session = AuthSession.anonymous() with pytest.raises(HTTPException) as exc: require_current_user(session=anonymous_session) assert exc.value.status_code == status.HTTP_401_UNAUTHORIZED assert exc.value.detail == "Authentication required." def test_require_authenticated_user_blocks_inactive_users(): inactive_password = random_password() user = User( email="user@example.com", username="user", password_hash=User.hash_password(inactive_password), is_active=False, ) with pytest.raises(HTTPException) as exc: require_authenticated_user(user=user) assert exc.value.status_code == status.HTTP_403_FORBIDDEN assert exc.value.detail == "User account is disabled." def test_require_roles_accepts_user_with_role(uow): user = _create_user(uow, roles=("viewer",)) dependency = require_roles("viewer") result = dependency(user=user) assert result is user def test_require_roles_allows_superuser_without_matching_role(uow): user = _create_user(uow, roles=(), is_superuser=True) dependency = require_roles("project_manager") result = dependency(user=user) assert result is user def test_require_roles_rejects_user_missing_required_role(uow): user = _create_user(uow, roles=("viewer",)) dependency = require_any_role("project_manager", "admin") with pytest.raises(HTTPException) as exc: dependency(user=user) assert exc.value.status_code == status.HTTP_403_FORBIDDEN assert exc.value.detail == "Insufficient permissions for this action." def test_require_roles_raises_value_error_when_no_roles_provided(): with pytest.raises(ValueError) as exc: require_roles() assert str(exc.value) == "require_roles requires at least one role name" def test_require_project_resource_returns_project_for_authorized_user(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) dependency = require_project_resource() result = dependency(project.id, user=user, uow=uow) assert result.id == project.id def test_require_project_resource_enforces_manage_requirement(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) dependency = require_project_resource(require_manage=True) with pytest.raises(HTTPException) as exc: dependency(project.id, user=user, uow=uow) assert exc.value.status_code == status.HTTP_403_FORBIDDEN assert exc.value.detail == "Insufficient role permissions for this action." def test_require_project_resource_raises_not_found_for_missing_project(uow): user = _create_user(uow, roles=("viewer",)) dependency = require_project_resource() with pytest.raises(HTTPException) as exc: dependency(9999, user=user, uow=uow) assert exc.value.status_code == status.HTTP_404_NOT_FOUND assert exc.value.detail == "Project 9999 not found" def test_require_project_resource_rejects_inactive_user(uow): user = _create_user(uow, roles=("viewer",), is_active=False) project = _create_project(uow) dependency = require_project_resource() with pytest.raises(HTTPException) as exc: dependency(project.id, user=user, uow=uow) assert exc.value.status_code == status.HTTP_403_FORBIDDEN assert exc.value.detail == "User account is disabled." def test_require_scenario_resource_returns_scenario_for_authorized_user(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) scenario = _create_scenario(uow, project) dependency = require_scenario_resource() result = dependency(scenario.id, user=user, uow=uow) assert result.id == scenario.id def test_require_scenario_resource_requires_manage_role(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) scenario = _create_scenario(uow, project) dependency = require_scenario_resource(require_manage=True) with pytest.raises(HTTPException) as exc: dependency(scenario.id, user=user, uow=uow) assert exc.value.status_code == status.HTTP_403_FORBIDDEN assert exc.value.detail == "Insufficient role permissions for this action." def test_require_scenario_resource_raises_not_found(uow): user = _create_user(uow, roles=("viewer",)) dependency = require_scenario_resource() with pytest.raises(HTTPException) as exc: dependency(12345, user=user, uow=uow) assert exc.value.status_code == status.HTTP_404_NOT_FOUND assert exc.value.detail == "Scenario 12345 not found" def test_require_project_scenario_resource_returns_scenario_when_linked(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) scenario = _create_scenario(uow, project) dependency = require_project_scenario_resource() result = dependency(project.id, scenario.id, user=user, uow=uow) assert result.id == scenario.id def test_require_project_scenario_resource_raises_when_scenario_not_in_project(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) other_project = _create_project(uow) scenario = _create_scenario(uow, other_project) dependency = require_project_scenario_resource() with pytest.raises(HTTPException) as exc: dependency(project.id, scenario.id, user=user, uow=uow) assert exc.value.status_code == status.HTTP_404_NOT_FOUND assert "does not belong" in exc.value.detail def test_require_project_scenario_resource_requires_manage_role(uow): user = _create_user(uow, roles=("viewer",)) project = _create_project(uow) scenario = _create_scenario(uow, project) dependency = require_project_scenario_resource(require_manage=True) with pytest.raises(HTTPException) as exc: dependency(project.id, scenario.id, user=user, uow=uow) assert exc.value.status_code == status.HTTP_403_FORBIDDEN assert exc.value.detail == "Insufficient role permissions for this action."