294 lines
10 KiB
Python
294 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
import tempfile
|
|
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.core.access import PrincipalRef
|
|
from govoplan_core.core.lifecycle import require_module_active
|
|
from govoplan_core.core.module_entitlements import (
|
|
ModuleEntitlementConflict,
|
|
ModuleEntitlementError,
|
|
tenant_module_entitlement_state,
|
|
update_system_tenant_module_policy,
|
|
update_tenant_module_selection,
|
|
)
|
|
from govoplan_core.core.modules import ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.db.session import configure_database, get_database
|
|
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)
|
|
|
|
|
|
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)
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|