diff --git a/src/govoplan_core/core/policy.py b/src/govoplan_core/core/policy.py index d23e97b..3c97701 100644 --- a/src/govoplan_core/core/policy.py +++ b/src/govoplan_core/core/policy.py @@ -18,12 +18,27 @@ DefinitionGovernanceAction = Literal[ "derive", "automate", ] +ViewGovernanceAction = Literal[ + "view", + "select", + "assign", + "edit", + "derive", + "workflow_activate", +] CAPABILITY_POLICY_PRIVACY_RETENTION = "policy.privacyRetention" CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY = "policy.schedulingParticipantPrivacy" CAPABILITY_POLICY_DEFINITION_GOVERNANCE = "policy.definitionGovernance" +CAPABILITY_POLICY_VIEW_GOVERNANCE = "policy.viewGovernance" -POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ("system", "tenant", "user", "group", "campaign") +POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ( + "system", + "tenant", + "user", + "group", + "campaign", +) def normalize_policy_scope_type(scope_type: str) -> PolicyScopeType: @@ -43,7 +58,11 @@ class PolicySourceRef: return policy_source_path(self.scope_type, self.scope_id) def to_dict(self) -> dict[str, Any]: - return {"scope_type": self.scope_type, "scope_id": self.scope_id, "path": self.path} + return { + "scope_type": self.scope_type, + "scope_id": self.scope_id, + "path": self.path, + } def policy_source_path(scope_type: str, scope_id: str | None = None) -> str: @@ -53,7 +72,9 @@ def policy_source_path(scope_type: str, scope_id: str | None = None) -> str: raise ValueError("System policy sources do not carry a scope_id") return "system" if not scope_id: - raise ValueError(f"{clean_scope.capitalize()} policy sources require a scope_id") + raise ValueError( + f"{clean_scope.capitalize()} policy sources require a scope_id" + ) return f"{clean_scope}:{quote(str(scope_id), safe='')}" @@ -63,7 +84,9 @@ def parse_policy_source_path(path: str) -> PolicySourceRef: return PolicySourceRef(scope_type="system") scope_type, separator, encoded_scope_id = clean_path.partition(":") if not separator: - raise ValueError("Policy source path must be system or :") + raise ValueError( + "Policy source path must be system or :" + ) clean_scope = normalize_policy_scope_type(scope_type) if clean_scope == "system": raise ValueError("System policy source path must be exactly system") @@ -100,9 +123,13 @@ class PolicySourceStep: policy_value = value.get("policy") return cls( scope_type=normalize_policy_scope_type(str(value.get("scope_type", ""))), - scope_id=str(value["scope_id"]) if value.get("scope_id") is not None else None, + scope_id=str(value["scope_id"]) + if value.get("scope_id") is not None + else None, label=str(value.get("label") or ""), - applied_fields=tuple(str(field) for field in (value.get("applied_fields") or ())), + applied_fields=tuple( + str(field) for field in (value.get("applied_fields") or ()) + ), policy=policy_value if isinstance(policy_value, Mapping) else {}, ) @@ -176,8 +203,7 @@ class DefinitionGovernancePolicy(Protocol): session: object | None = None, *, request: DefinitionGovernanceRequest, - ) -> PolicyDecision: - ... + ) -> PolicyDecision: ... def definition_governance_policy( @@ -186,17 +212,79 @@ def definition_governance_policy( if ( registry is None or not hasattr(registry, "has_capability") - or not registry.has_capability( - CAPABILITY_POLICY_DEFINITION_GOVERNANCE - ) + or not registry.has_capability(CAPABILITY_POLICY_DEFINITION_GOVERNANCE) ): return None capability = registry.capability(CAPABILITY_POLICY_DEFINITION_GOVERNANCE) - return ( - capability - if isinstance(capability, DefinitionGovernancePolicy) - else None - ) + return capability if isinstance(capability, DefinitionGovernancePolicy) else None + + +@dataclass(frozen=True, slots=True) +class ViewGovernanceRequest: + tenant_id: str + action: ViewGovernanceAction + actor: PrincipalRef + target_scope: DefinitionScopeRef + view_id: str | None = None + candidate_view_ids: tuple[str, ...] = () + candidate_surface_ids: tuple[str, ...] = () + requested_surface_ids: tuple[str, ...] = () + context: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ViewGovernanceDecision: + allowed: bool + reason: str | None = None + allowed_view_ids: frozenset[str] | None = None + visible_surface_ids: frozenset[str] | None = None + source_path: tuple[PolicySourceStep, ...] = () + requirements: tuple[str, ...] = () + diagnostics: tuple[Mapping[str, Any], ...] = () + details: Mapping[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "allowed": self.allowed, + "reason": self.reason, + "allowed_view_ids": ( + sorted(self.allowed_view_ids) + if self.allowed_view_ids is not None + else None + ), + "visible_surface_ids": ( + sorted(self.visible_surface_ids) + if self.visible_surface_ids is not None + else None + ), + "source_path": [step.to_dict() for step in self.source_path], + "requirements": list(self.requirements), + "diagnostics": [dict(item) for item in self.diagnostics], + "details": dict(self.details), + } + + +@runtime_checkable +class ViewGovernancePolicy(Protocol): + def resolve_view_action( + self, + session: object | None = None, + *, + request: ViewGovernanceRequest, + ) -> ViewGovernanceDecision: ... + + +def view_governance_policy( + registry: object | None, +) -> ViewGovernancePolicy | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_POLICY_VIEW_GOVERNANCE) + ): + return None + capability = registry.capability(CAPABILITY_POLICY_VIEW_GOVERNANCE) + return capability if isinstance(capability, ViewGovernancePolicy) else None @dataclass(frozen=True, slots=True) @@ -244,44 +332,32 @@ class SchedulingParticipantPrivacyPolicy(Protocol): session: object, *, request: SchedulingParticipantPrivacyRequest, - ) -> SchedulingParticipantPrivacyDecision: - ... + ) -> SchedulingParticipantPrivacyDecision: ... @runtime_checkable class PrivacyRetentionService(Protocol): - def privacy_policy_from_settings(self, *args: Any, **kwargs: Any) -> Any: - ... + def privacy_policy_from_settings(self, *args: Any, **kwargs: Any) -> Any: ... - def privacy_policy_from_session(self, *args: Any, **kwargs: Any) -> Any: - ... + def privacy_policy_from_session(self, *args: Any, **kwargs: Any) -> Any: ... - def set_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: - ... + def set_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: ... - def effective_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: - ... + def effective_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: ... - def parent_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: - ... + def parent_privacy_policy(self, *args: Any, **kwargs: Any) -> Any: ... - def effective_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: - ... + def effective_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: ... - def parent_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: - ... + def parent_privacy_policy_sources(self, *args: Any, **kwargs: Any) -> Any: ... - def get_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: - ... + def get_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: ... - def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: - ... + def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any: ... - def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any: - ... + def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any: ... - def apply_retention_policy(self, *args: Any, **kwargs: Any) -> Any: - ... + def apply_retention_policy(self, *args: Any, **kwargs: Any) -> Any: ... def scheduling_participant_privacy_policy( @@ -294,4 +370,8 @@ def scheduling_participant_privacy_policy( if not registry.has_capability(CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY): return None capability = registry.capability(CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY) - return capability if isinstance(capability, SchedulingParticipantPrivacyPolicy) else None + return ( + capability + if isinstance(capability, SchedulingParticipantPrivacyPolicy) + else None + ) diff --git a/src/govoplan_core/core/views.py b/src/govoplan_core/core/views.py index fbb0cdb..e7601c1 100644 --- a/src/govoplan_core/core/views.py +++ b/src/govoplan_core/core/views.py @@ -43,6 +43,7 @@ class EffectiveView: name: str | None visible_surface_ids: frozenset[str] locked: bool = False + projection_active: bool = False provenance: tuple[dict[str, object], ...] = () diff --git a/tests/test_view_governance_contract.py b/tests/test_view_governance_contract.py new file mode 100644 index 0000000..0d5856c --- /dev/null +++ b/tests/test_view_governance_contract.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_VIEW_GOVERNANCE, + DefinitionScopeRef, + ViewGovernanceDecision, + ViewGovernancePolicy, + ViewGovernanceRequest, + view_governance_policy, +) +from govoplan_core.core.registry import PlatformRegistry + + +class _ViewPolicy: + def resolve_view_action(self, session=None, *, request): + del session + return ViewGovernanceDecision( + allowed=request.action != "assign", + allowed_view_ids=frozenset(request.candidate_view_ids[:1]), + ) + + +class ViewGovernanceContractTests(unittest.TestCase): + def test_policy_is_runtime_resolved_through_core_contract(self) -> None: + provider = _ViewPolicy() + self.assertIsInstance(provider, ViewGovernancePolicy) + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="view_governance_test", + name="View governance test", + version="test", + capability_factories={ + CAPABILITY_POLICY_VIEW_GOVERNANCE: lambda context: provider, + }, + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + + resolved = view_governance_policy(registry) + self.assertIs(provider, resolved) + decision = resolved.resolve_view_action( + request=ViewGovernanceRequest( + tenant_id="tenant-1", + action="view", + actor=PrincipalRef( + account_id="account-1", + membership_id=None, + tenant_id="tenant-1", + ), + target_scope=DefinitionScopeRef("user", "account-1"), + candidate_view_ids=("view-1", "view-2"), + ) + ) + + self.assertTrue(decision.allowed) + self.assertEqual(frozenset({"view-1"}), decision.allowed_view_ids) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/platform/views.ts b/webui/src/platform/views.ts index bf11851..a25c2db 100644 --- a/webui/src/platform/views.ts +++ b/webui/src/platform/views.ts @@ -84,11 +84,13 @@ export function isViewSurfaceVisible( surfaceId: string | null | undefined, catalogue: PlatformViewSurface[] ): boolean { - if (!surfaceId || !projection?.activeViewId) return true; + const projectionActive = + projection?.projectionActive ?? Boolean(projection?.activeViewId); + if (!surfaceId || !projectionActive) return true; const byId = new Map(catalogue.map((surface) => [surface.id, surface])); const surface = byId.get(surfaceId); if (!surface) return true; - const visible = new Set(projection.visibleSurfaceIds); + const visible = new Set(projection?.visibleSurfaceIds ?? []); let current: PlatformViewSurface | undefined = surface; const visited = new Set(); while (current) { diff --git a/webui/src/types.ts b/webui/src/types.ts index 3c1442c..b203ec6 100644 --- a/webui/src/types.ts +++ b/webui/src/types.ts @@ -410,6 +410,7 @@ export type EffectiveViewProjection = { activeRevisionId: string | null; activeViewName: string | null; visibleSurfaceIds: string[]; + projectionActive?: boolean; locked: boolean; availableViews: EffectiveViewOption[]; provenance: Array<{