Files
TJWaterServerBinary/tests/auth/test_permissions.py
T

160 lines
4.0 KiB
Python

from uuid import uuid4
import pytest
from fastapi import HTTPException
from app.auth.permissions import (
AUDIT_VIEW,
ENVIRONMENT_MANAGE,
MODEL_IMPORT,
OPTIMIZATION_RUN,
SCADA_CLEAN,
SIMULATION_RUN,
SIMULATION_VIEW,
WEBGIS_EDIT,
WEBGIS_VIEW,
permissions_for_context,
require_method_permission,
require_permission,
resolve_permissions,
)
from app.auth.project_dependencies import ProjectContext
@pytest.fixture
def anyio_backend():
return "asyncio"
def _context(project_role: str) -> ProjectContext:
return ProjectContext(
project_id=uuid4(),
project_code="demo",
user_id=uuid4(),
project_role=project_role,
)
def test_project_role_permission_matrix():
member = resolve_permissions(
project_role="member",
system_role="user",
is_superuser=False,
)
viewer = resolve_permissions(
project_role="viewer",
system_role="user",
is_superuser=False,
)
assert WEBGIS_EDIT in member
assert SCADA_CLEAN in member
assert SIMULATION_RUN in member
assert OPTIMIZATION_RUN in member
assert MODEL_IMPORT not in member
assert WEBGIS_VIEW in viewer
assert SIMULATION_VIEW in viewer
assert WEBGIS_EDIT not in viewer
assert SCADA_CLEAN not in viewer
assert SIMULATION_RUN not in viewer
def test_system_admin_permissions_do_not_grant_project_business_access():
permissions = resolve_permissions(
project_role=None,
system_role="admin",
is_superuser=False,
)
assert ENVIRONMENT_MANAGE in permissions
assert AUDIT_VIEW in permissions
assert MODEL_IMPORT in permissions
assert WEBGIS_VIEW not in permissions
@pytest.mark.anyio
async def test_permission_dependency_returns_context_when_allowed():
ctx = _context("member")
dependency = require_permission(WEBGIS_EDIT)
request = type(
"Request",
(),
{
"path_params": {},
"query_params": {},
"headers": {},
},
)()
assert await dependency(request, ctx) is ctx
@pytest.mark.anyio
async def test_permission_dependency_returns_structured_403_when_denied():
ctx = _context("viewer")
dependency = require_permission(WEBGIS_EDIT)
with pytest.raises(HTTPException) as exc:
await dependency(None, ctx)
assert exc.value.status_code == 403
assert exc.value.detail == {
"code": "permission_denied",
"permission": WEBGIS_EDIT,
}
@pytest.mark.anyio
async def test_permission_dependency_rejects_cross_project_network():
ctx = _context("member")
dependency = require_permission(WEBGIS_VIEW)
request = type(
"Request",
(),
{
"path_params": {},
"query_params": {"network": "other-project"},
"headers": {},
},
)()
with pytest.raises(HTTPException) as exc:
await dependency(request, ctx)
assert exc.value.status_code == 403
assert exc.value.detail["code"] == "project_scope_denied"
def test_member_keeps_full_web_business_access():
permissions = permissions_for_context(_context("member"))
assert SCADA_CLEAN in permissions
assert SIMULATION_RUN in permissions
@pytest.mark.anyio
async def test_viewer_can_read_but_cannot_write_or_run():
ctx = _context("viewer")
dependency = require_method_permission(
read_permission=SIMULATION_VIEW,
write_permission=SIMULATION_RUN,
)
read_request = type(
"Request",
(),
{"method": "GET", "path_params": {}, "query_params": {}, "headers": {}},
)()
write_request = type(
"Request",
(),
{"method": "POST", "path_params": {}, "query_params": {}, "headers": {}},
)()
assert await dependency(read_request, ctx) is ctx
with pytest.raises(HTTPException) as exc:
await dependency(write_request, ctx)
assert exc.value.status_code == 403
assert exc.value.detail["permission"] == SIMULATION_RUN