512 lines
18 KiB
Python
512 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import APIRouter, Depends, FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.admin.models import SystemSettings
|
|
from govoplan_core.celery_app import _run_tenant_worker_batches
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.lifecycle import require_module_active
|
|
from govoplan_core.core.module_entitlements import (
|
|
ModuleEntitlementConflict,
|
|
ModuleEntitlementError,
|
|
TenantModuleEntitlementResolver,
|
|
TenantModuleOperatorActionRequired,
|
|
TenantModuleUnavailable,
|
|
tenant_module_entitlement_state,
|
|
update_system_tenant_module_policy,
|
|
update_tenant_module_selection,
|
|
)
|
|
from govoplan_core.core.modules import ModuleContext, ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.db.session import configure_database, get_database
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.server.platform import create_platform_router
|
|
from govoplan_core.tenancy.scope import Tenant, create_scope_tables
|
|
|
|
|
|
class TenantModuleEntitlementTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.manifests = {
|
|
"access": ModuleManifest(id="access", name="Access", version="test"),
|
|
"admin": ModuleManifest(
|
|
id="admin",
|
|
name="Admin",
|
|
version="test",
|
|
dependencies=("access",),
|
|
),
|
|
"files": ModuleManifest(
|
|
id="files",
|
|
name="Files",
|
|
version="test",
|
|
dependencies=("access",),
|
|
),
|
|
"campaigns": ModuleManifest(
|
|
id="campaigns",
|
|
name="Campaigns",
|
|
version="test",
|
|
dependencies=("access", "files"),
|
|
),
|
|
"encryption": ModuleManifest(
|
|
id="encryption",
|
|
name="Encryption",
|
|
version="test",
|
|
),
|
|
}
|
|
|
|
def test_unconfigured_tenant_preserves_current_module_visibility(self) -> None:
|
|
state = tenant_module_entitlement_state({}, self.manifests)
|
|
|
|
self.assertFalse(state.configured)
|
|
self.assertEqual(set(self.manifests), set(state.effective_modules))
|
|
self.assertEqual({"access", "admin"}, set(state.forced_modules))
|
|
|
|
def test_system_policy_closes_dependencies_and_tenant_selection(self) -> None:
|
|
settings, state = update_system_tenant_module_policy(
|
|
{},
|
|
self.manifests,
|
|
available_modules=("campaigns", "encryption"),
|
|
forced_modules=("campaigns",),
|
|
enabled_modules=("encryption",),
|
|
expected_revision=0,
|
|
)
|
|
|
|
self.assertEqual(1, state.revision)
|
|
self.assertEqual(
|
|
{"access", "admin", "files", "campaigns", "encryption"},
|
|
set(state.available_modules),
|
|
)
|
|
self.assertEqual(
|
|
{"access", "admin", "files", "campaigns"},
|
|
set(state.forced_modules),
|
|
)
|
|
self.assertEqual({"encryption"}, set(state.selected_modules))
|
|
self.assertEqual(set(self.manifests), set(state.effective_modules))
|
|
self.assertIn("module_entitlements", settings)
|
|
|
|
def test_tenant_cannot_enable_system_unavailable_module(self) -> None:
|
|
settings, _state = update_system_tenant_module_policy(
|
|
{},
|
|
self.manifests,
|
|
available_modules=("files",),
|
|
forced_modules=(),
|
|
enabled_modules=(),
|
|
expected_revision=0,
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
ModuleEntitlementError,
|
|
"unavailable by system policy: encryption",
|
|
):
|
|
update_tenant_module_selection(
|
|
settings,
|
|
self.manifests,
|
|
enabled_modules=("encryption",),
|
|
expected_revision=1,
|
|
)
|
|
|
|
def test_forced_modules_remain_effective_when_tenant_selection_is_empty(self) -> None:
|
|
settings, _state = update_system_tenant_module_policy(
|
|
{},
|
|
self.manifests,
|
|
available_modules=("campaigns",),
|
|
forced_modules=("campaigns",),
|
|
enabled_modules=(),
|
|
expected_revision=0,
|
|
)
|
|
_settings, state = update_tenant_module_selection(
|
|
settings,
|
|
self.manifests,
|
|
enabled_modules=(),
|
|
expected_revision=1,
|
|
)
|
|
|
|
self.assertEqual(
|
|
{"access", "admin", "files", "campaigns"},
|
|
set(state.effective_modules),
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
not item.tenant_can_toggle
|
|
for item in state.modules
|
|
if item.id in state.forced_modules
|
|
)
|
|
)
|
|
|
|
def test_inactive_runtime_module_is_selected_but_not_effective(self) -> None:
|
|
settings, state = update_system_tenant_module_policy(
|
|
{},
|
|
self.manifests,
|
|
available_modules=("files", "encryption"),
|
|
forced_modules=(),
|
|
enabled_modules=("encryption",),
|
|
expected_revision=0,
|
|
runtime_active_modules=("access", "admin", "files"),
|
|
)
|
|
|
|
self.assertIn("encryption", state.selected_modules)
|
|
self.assertNotIn("encryption", state.effective_modules)
|
|
encryption = next(item for item in state.modules if item.id == "encryption")
|
|
self.assertIn("not active in the deployment", encryption.reason or "")
|
|
self.assertIn("module_entitlements", settings)
|
|
|
|
def test_stale_revision_is_rejected(self) -> None:
|
|
settings, _state = update_system_tenant_module_policy(
|
|
{},
|
|
self.manifests,
|
|
available_modules=("files",),
|
|
forced_modules=(),
|
|
enabled_modules=("files",),
|
|
expected_revision=0,
|
|
)
|
|
|
|
with self.assertRaises(ModuleEntitlementConflict):
|
|
update_tenant_module_selection(
|
|
settings,
|
|
self.manifests,
|
|
enabled_modules=(),
|
|
expected_revision=0,
|
|
)
|
|
|
|
def test_malformed_document_fails_closed_to_protected_modules(self) -> None:
|
|
state = tenant_module_entitlement_state(
|
|
{"module_entitlements": {"revision": "invalid"}},
|
|
self.manifests,
|
|
)
|
|
|
|
self.assertEqual({"access", "admin"}, set(state.effective_modules))
|
|
self.assertTrue(state.diagnostics)
|
|
|
|
def test_resolver_caches_and_invalidates_tenant_state(self) -> None:
|
|
registry = PlatformRegistry()
|
|
for manifest in self.manifests.values():
|
|
registry.register(manifest)
|
|
tenant = SimpleNamespace(id="tenant-1", is_active=True, settings={})
|
|
|
|
class CountingSession:
|
|
calls = 0
|
|
|
|
def get(self, _model, _tenant_id):
|
|
self.calls += 1
|
|
return tenant
|
|
|
|
session = CountingSession()
|
|
resolver = TenantModuleEntitlementResolver(
|
|
registry,
|
|
ttl_seconds=60,
|
|
max_entries=2,
|
|
)
|
|
|
|
resolver.resolve(session, "tenant-1")
|
|
resolver.resolve(session, "tenant-1")
|
|
self.assertEqual(1, session.calls)
|
|
|
|
resolver.invalidate("tenant-1")
|
|
resolver.resolve(session, "tenant-1")
|
|
self.assertEqual(2, session.calls)
|
|
|
|
def test_new_and_accepted_work_have_distinct_disable_semantics(self) -> None:
|
|
registry = PlatformRegistry()
|
|
for manifest in self.manifests.values():
|
|
registry.register(manifest)
|
|
settings, _state = update_system_tenant_module_policy(
|
|
{},
|
|
self.manifests,
|
|
available_modules=(),
|
|
forced_modules=(),
|
|
enabled_modules=(),
|
|
expected_revision=0,
|
|
)
|
|
tenant = SimpleNamespace(
|
|
id="tenant-1",
|
|
is_active=True,
|
|
settings=settings,
|
|
)
|
|
session = SimpleNamespace(get=lambda _model, _tenant_id: tenant)
|
|
resolver = TenantModuleEntitlementResolver(registry, ttl_seconds=0)
|
|
|
|
with self.assertRaises(TenantModuleUnavailable) as rejected:
|
|
resolver.require(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
work_state="new",
|
|
)
|
|
self.assertEqual("rejected", rejected.exception.admission.disposition)
|
|
|
|
with self.assertRaises(TenantModuleOperatorActionRequired) as preserved:
|
|
resolver.require(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
module_id="files",
|
|
work_state="accepted",
|
|
)
|
|
self.assertEqual(
|
|
"operator_action_required",
|
|
preserved.exception.admission.disposition,
|
|
)
|
|
|
|
|
|
class TenantModuleEntitlementRouteTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-entitlement-test-"))
|
|
configure_database(f"sqlite:///{root / 'test.db'}")
|
|
create_scope_tables(get_database().engine)
|
|
Base.metadata.create_all(
|
|
bind=get_database().engine,
|
|
tables=[SystemSettings.__table__],
|
|
)
|
|
self.manifests = (
|
|
ModuleManifest(id="access", name="Access", version="test"),
|
|
ModuleManifest(
|
|
id="admin",
|
|
name="Admin",
|
|
version="test",
|
|
dependencies=("access",),
|
|
),
|
|
ModuleManifest(
|
|
id="files",
|
|
name="Files",
|
|
version="test",
|
|
dependencies=("access",),
|
|
),
|
|
)
|
|
self.registry = PlatformRegistry()
|
|
for manifest in self.manifests:
|
|
self.registry.register(manifest)
|
|
settings, _state = update_system_tenant_module_policy(
|
|
{},
|
|
{manifest.id: manifest for manifest in self.manifests},
|
|
available_modules=(),
|
|
forced_modules=(),
|
|
enabled_modules=(),
|
|
expected_revision=0,
|
|
)
|
|
with get_database().session() as session:
|
|
session.add(
|
|
Tenant(
|
|
id="tenant-1",
|
|
slug="tenant-1",
|
|
name="Tenant 1",
|
|
settings=settings,
|
|
)
|
|
)
|
|
session.commit()
|
|
self.principal = ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
),
|
|
account=object(),
|
|
user=object(),
|
|
)
|
|
|
|
def test_platform_metadata_excludes_tenant_unavailable_module(self) -> None:
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = self.registry
|
|
app.include_router(create_platform_router(), prefix="/api/v1")
|
|
app.dependency_overrides[get_api_principal] = lambda: self.principal
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/v1/platform/modules")
|
|
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertEqual(
|
|
{"access", "admin"},
|
|
{item["id"] for item in response.json()["modules"]},
|
|
)
|
|
self.assertNotIn(
|
|
"files",
|
|
response.json()["module_entitlement"]["effective_modules"],
|
|
)
|
|
|
|
def test_authenticated_module_route_is_hidden_when_tenant_unavailable(self) -> None:
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = self.registry
|
|
guarded = APIRouter(dependencies=[Depends(require_module_active("files"))])
|
|
|
|
@guarded.get("/files")
|
|
def files_route():
|
|
return {"ok": True}
|
|
|
|
app.include_router(guarded)
|
|
with patch(
|
|
"govoplan_core.core.lifecycle.get_api_principal",
|
|
return_value=self.principal,
|
|
), TestClient(app) as client:
|
|
response = client.get(
|
|
"/files",
|
|
headers={"Authorization": "Bearer test"},
|
|
)
|
|
|
|
self.assertEqual(404, response.status_code, response.text)
|
|
self.assertEqual(
|
|
"Module is unavailable in the active tenant: files",
|
|
response.json()["detail"],
|
|
)
|
|
|
|
def test_unauthenticated_public_route_is_not_turned_into_login(self) -> None:
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = self.registry
|
|
guarded = APIRouter(dependencies=[Depends(require_module_active("files"))])
|
|
|
|
@guarded.get("/public-files")
|
|
def public_files_route():
|
|
return {"ok": True}
|
|
|
|
app.include_router(guarded)
|
|
with TestClient(app) as client:
|
|
response = client.get("/public-files")
|
|
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
|
|
def test_public_tenant_route_enforces_module_entitlement(self) -> None:
|
|
registry = PlatformRegistry()
|
|
for manifest in self.manifests:
|
|
registry.register(
|
|
ModuleManifest(
|
|
id=manifest.id,
|
|
name=manifest.name,
|
|
version=manifest.version,
|
|
dependencies=manifest.dependencies,
|
|
public_tenant_resolver=(
|
|
(lambda _request, _session: "tenant-1")
|
|
if manifest.id == "files"
|
|
else None
|
|
),
|
|
)
|
|
)
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = registry
|
|
guarded = APIRouter(dependencies=[Depends(require_module_active("files"))])
|
|
|
|
@guarded.get("/public-files/{token}")
|
|
def public_files_route(token: str):
|
|
return {"token": token}
|
|
|
|
app.include_router(guarded)
|
|
with TestClient(app) as client:
|
|
response = client.get("/public-files/example")
|
|
|
|
self.assertEqual(404, response.status_code, response.text)
|
|
self.assertEqual(
|
|
"Module is unavailable in the active tenant: files",
|
|
response.json()["detail"],
|
|
)
|
|
|
|
def test_tenant_capability_rejects_unavailable_provider(self) -> None:
|
|
registry = PlatformRegistry()
|
|
for manifest in self.manifests:
|
|
registry.register(
|
|
ModuleManifest(
|
|
id=manifest.id,
|
|
name=manifest.name,
|
|
version=manifest.version,
|
|
dependencies=manifest.dependencies,
|
|
capability_factories=(
|
|
{"files.example": lambda _context: object()}
|
|
if manifest.id == "files"
|
|
else {}
|
|
),
|
|
)
|
|
)
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=SimpleNamespace())
|
|
)
|
|
with get_database().session() as session:
|
|
with self.assertRaises(TenantModuleUnavailable):
|
|
registry.require_tenant_capability(
|
|
"files.example",
|
|
session,
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
def test_request_context_treats_unavailable_optional_capability_as_absent(self) -> None:
|
|
registry = PlatformRegistry()
|
|
for manifest in self.manifests:
|
|
registry.register(
|
|
ModuleManifest(
|
|
id=manifest.id,
|
|
name=manifest.name,
|
|
version=manifest.version,
|
|
dependencies=manifest.dependencies,
|
|
capability_factories=(
|
|
{"files.example": lambda _context: object()}
|
|
if manifest.id == "files"
|
|
else {}
|
|
),
|
|
)
|
|
)
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=SimpleNamespace())
|
|
)
|
|
app = FastAPI()
|
|
app.state.govoplan_registry = registry
|
|
guarded = APIRouter(dependencies=[Depends(require_module_active("admin"))])
|
|
|
|
@guarded.get("/admin-capability")
|
|
def admin_capability_route():
|
|
return {"files_available": registry.capability("files.example") is not None}
|
|
|
|
app.include_router(guarded)
|
|
with patch(
|
|
"govoplan_core.core.lifecycle.get_api_principal",
|
|
return_value=self.principal,
|
|
), TestClient(app) as client:
|
|
response = client.get(
|
|
"/admin-capability",
|
|
headers={"Authorization": "Bearer test"},
|
|
)
|
|
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertFalse(response.json()["files_available"])
|
|
|
|
def test_worker_preserves_accepted_work_for_operator_when_disabled(self) -> None:
|
|
registry = PlatformRegistry()
|
|
for manifest in self.manifests:
|
|
registry.register(
|
|
ModuleManifest(
|
|
id=manifest.id,
|
|
name=manifest.name,
|
|
version=manifest.version,
|
|
dependencies=manifest.dependencies,
|
|
capability_factories=(
|
|
{"files.worker": lambda _context: object()}
|
|
if manifest.id == "files"
|
|
else {}
|
|
),
|
|
)
|
|
)
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=SimpleNamespace())
|
|
)
|
|
invoked: list[str] = []
|
|
with get_database().session() as session:
|
|
result = _run_tenant_worker_batches(
|
|
registry,
|
|
session,
|
|
capability_name="files.worker",
|
|
tenant_id="tenant-1",
|
|
operation=lambda tenant_id: invoked.append(tenant_id) or {},
|
|
defaults={"processed": 0},
|
|
)
|
|
|
|
self.assertEqual([], invoked)
|
|
self.assertEqual(1, result["operator_action_required"])
|
|
self.assertEqual(
|
|
"operator_action_required",
|
|
result["operator_actions"][0]["disposition"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|