From 2511fbb5a864819589974399a3bce0ff4512193d Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 14 Jul 2026 13:22:12 +0200 Subject: [PATCH] intermittent commit --- src/govoplan_policy/backend/api/v1/routes.py | 2 +- src/govoplan_policy/backend/api/v1/schemas.py | 73 +----- src/govoplan_policy/backend/retention.py | 233 ++++++++++-------- tests/test_policy_hierarchy.py | 38 +++ tests/test_policy_module_contract.py | 2 +- 5 files changed, 177 insertions(+), 171 deletions(-) diff --git a/src/govoplan_policy/backend/api/v1/routes.py b/src/govoplan_policy/backend/api/v1/routes.py index 21d4ff0..496a758 100644 --- a/src/govoplan_policy/backend/api/v1/routes.py +++ b/src/govoplan_policy/backend/api/v1/routes.py @@ -14,9 +14,9 @@ from govoplan_core.core.configuration_control import ( ) from govoplan_core.core.policy import PolicyDecision, PolicySourceStep from govoplan_core.db.session import get_session +from govoplan_core.privacy.schemas import RETENTION_POLICY_FIELD_KEYS from govoplan_policy.backend.retention import ( PrivacyPolicyError, - RETENTION_POLICY_FIELD_KEYS, apply_retention_policy, effective_privacy_policy, effective_privacy_policy_sources, diff --git a/src/govoplan_policy/backend/api/v1/schemas.py b/src/govoplan_policy/backend/api/v1/schemas.py index 1f4834b..d7aec46 100644 --- a/src/govoplan_policy/backend/api/v1/schemas.py +++ b/src/govoplan_policy/backend/api/v1/schemas.py @@ -1,43 +1,10 @@ from __future__ import annotations -from datetime import datetime from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import BaseModel, ConfigDict, Field - -RETENTION_DAY_KEYS = ( - "raw_campaign_json_retention_days", - "generated_eml_retention_days", - "stored_report_detail_retention_days", - "mock_mailbox_retention_days", - "audit_detail_retention_days", -) - - -RETENTION_POLICY_FIELD_KEYS = ( - "store_raw_campaign_json", - *RETENTION_DAY_KEYS, - "audit_detail_level", -) - - -def default_allow_lower_level_limits() -> dict[str, bool]: - return {key: True for key in RETENTION_POLICY_FIELD_KEYS} - - -def normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None: - if value in (None, ""): - return default_allow_lower_level_limits() if fill_defaults else None - if not isinstance(value, dict): - raise ValueError("allow_lower_level_limits must be an object") - normalized = default_allow_lower_level_limits() if fill_defaults else {} - for key, allowed in value.items(): - clean_key = str(key) - if clean_key not in RETENTION_POLICY_FIELD_KEYS: - raise ValueError(f"Unknown retention policy field: {clean_key}") - normalized[clean_key] = bool(allowed) - return normalized +from govoplan_core.privacy.schemas import PrivacyRetentionPolicyItem, PrivacyRetentionPolicyPatchItem class PolicySourceStepItem(BaseModel): @@ -49,42 +16,6 @@ class PolicySourceStepItem(BaseModel): policy: dict[str, Any] = Field(default_factory=dict) -class PrivacyRetentionPolicyItem(BaseModel): - model_config = ConfigDict(extra="forbid") - - store_raw_campaign_json: bool = True - raw_campaign_json_retention_days: int | None = Field(default=None, ge=0) - generated_eml_retention_days: int | None = Field(default=None, ge=0) - stored_report_detail_retention_days: int | None = Field(default=None, ge=0) - mock_mailbox_retention_days: int | None = Field(default=None, ge=0) - audit_detail_retention_days: int | None = Field(default=None, ge=0) - audit_detail_level: Literal["full", "redacted", "minimal"] = "full" - allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits) - - @field_validator("allow_lower_level_limits", mode="before") - @classmethod - def _normalize_allow_lower_level_limits(cls, value: Any) -> Any: - return normalize_allow_lower_level_limits(value, fill_defaults=True) - - -class PrivacyRetentionPolicyPatchItem(BaseModel): - model_config = ConfigDict(extra="forbid") - - store_raw_campaign_json: bool | None = None - raw_campaign_json_retention_days: int | None = Field(default=None, ge=0) - generated_eml_retention_days: int | None = Field(default=None, ge=0) - stored_report_detail_retention_days: int | None = Field(default=None, ge=0) - mock_mailbox_retention_days: int | None = Field(default=None, ge=0) - audit_detail_retention_days: int | None = Field(default=None, ge=0) - audit_detail_level: Literal["full", "redacted", "minimal"] | None = None - allow_lower_level_limits: dict[str, bool] | None = None - - @field_validator("allow_lower_level_limits", mode="before") - @classmethod - def _normalize_allow_lower_level_limits(cls, value: Any) -> Any: - return normalize_allow_lower_level_limits(value, fill_defaults=False) - - class PrivacyRetentionPolicyScopeRequest(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/src/govoplan_policy/backend/retention.py b/src/govoplan_policy/backend/retention.py index d493584..b1aabcd 100644 --- a/src/govoplan_policy/backend/retention.py +++ b/src/govoplan_policy/backend/retention.py @@ -3,9 +3,9 @@ from __future__ import annotations import json from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Literal +from typing import Any -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import field_validator from sqlalchemy.orm import Session from govoplan_core.admin.settings import get_system_settings @@ -26,6 +26,14 @@ from govoplan_core.core.access import ( from govoplan_core.core.policy import PolicySourceStep, policy_source_step from govoplan_core.core.runtime import get_registry from govoplan_core.admin.models import SystemSettings +from govoplan_core.privacy.schemas import ( + RETENTION_DAY_KEYS, + RETENTION_POLICY_FIELD_KEYS, + PrivacyRetentionPolicyItem, + PrivacyRetentionPolicyPatchItem, + default_allow_lower_level_limits, + normalize_allow_lower_level_limits, +) from govoplan_core.settings import settings from govoplan_core.tenancy.scope import Tenant from govoplan_policy.backend.hierarchy import ( @@ -35,38 +43,8 @@ from govoplan_policy.backend.hierarchy import ( ) PRIVACY_POLICY_SETTINGS_KEY = "privacy_retention_policy" -RETENTION_DAY_KEYS = ( - "raw_campaign_json_retention_days", - "generated_eml_retention_days", - "stored_report_detail_retention_days", - "mock_mailbox_retention_days", - "audit_detail_retention_days", -) -RETENTION_POLICY_FIELD_KEYS = ( - "store_raw_campaign_json", - *RETENTION_DAY_KEYS, - "audit_detail_level", -) AUDIT_DETAIL_LEVEL_ORDER = {"full": 0, "redacted": 1, "minimal": 2} - -def default_allow_lower_level_limits() -> dict[str, bool]: - return {key: True for key in RETENTION_POLICY_FIELD_KEYS} - - -def _normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None: - if value in (None, ""): - return default_allow_lower_level_limits() if fill_defaults else None - if not isinstance(value, dict): - raise ValueError("allow_lower_level_limits must be an object") - normalized = default_allow_lower_level_limits() if fill_defaults else {} - for key, allowed in value.items(): - clean_key = str(key) - if clean_key not in RETENTION_POLICY_FIELD_KEYS: - raise ValueError(f"Unknown retention policy field: {clean_key}") - normalized[clean_key] = bool(allowed) - return normalized - AUDIT_DETAIL_REDACTION_KEYS = { "attachments", "bcc", @@ -91,18 +69,7 @@ AUDIT_DETAIL_REDACTION_KEYS = { } -class PrivacyRetentionPolicy(BaseModel): - model_config = ConfigDict(extra="forbid") - - store_raw_campaign_json: bool = True - raw_campaign_json_retention_days: int | None = Field(default=None, ge=0) - generated_eml_retention_days: int | None = Field(default=None, ge=0) - stored_report_detail_retention_days: int | None = Field(default=None, ge=0) - mock_mailbox_retention_days: int | None = Field(default=None, ge=0) - audit_detail_retention_days: int | None = Field(default=None, ge=0) - audit_detail_level: Literal["full", "redacted", "minimal"] = "full" - allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits) - +class PrivacyRetentionPolicy(PrivacyRetentionPolicyItem): @field_validator( "raw_campaign_json_retention_days", "generated_eml_retention_days", @@ -120,21 +87,10 @@ class PrivacyRetentionPolicy(BaseModel): @field_validator("allow_lower_level_limits", mode="before") @classmethod def _normalize_allow_lower_level_limits(cls, value: Any) -> Any: - return _normalize_allow_lower_level_limits(value, fill_defaults=True) + return normalize_allow_lower_level_limits(value, fill_defaults=True) -class PrivacyRetentionPolicyPatch(BaseModel): - model_config = ConfigDict(extra="forbid") - - store_raw_campaign_json: bool | None = None - raw_campaign_json_retention_days: int | None = Field(default=None, ge=0) - generated_eml_retention_days: int | None = Field(default=None, ge=0) - stored_report_detail_retention_days: int | None = Field(default=None, ge=0) - mock_mailbox_retention_days: int | None = Field(default=None, ge=0) - audit_detail_retention_days: int | None = Field(default=None, ge=0) - audit_detail_level: Literal["full", "redacted", "minimal"] | None = None - allow_lower_level_limits: dict[str, bool] | None = None - +class PrivacyRetentionPolicyPatch(PrivacyRetentionPolicyPatchItem): @field_validator(*RETENTION_DAY_KEYS, mode="before") @classmethod def _blank_string_is_none(cls, value: Any) -> Any: @@ -145,7 +101,7 @@ class PrivacyRetentionPolicyPatch(BaseModel): @field_validator("allow_lower_level_limits", mode="before") @classmethod def _normalize_allow_lower_level_limits(cls, value: Any) -> Any: - return _normalize_allow_lower_level_limits(value, fill_defaults=False) + return normalize_allow_lower_level_limits(value, fill_defaults=False) class PrivacyPolicyError(RuntimeError): @@ -524,49 +480,130 @@ def set_privacy_policy_for_scope( ) -> dict[str, Any]: clean_scope = scope_type.strip().casefold() if clean_scope == "system": - item = get_system_settings(session) - validated = set_privacy_policy(item, PrivacyRetentionPolicy.model_validate(policy or {})) - session.add(item) - return validated.model_dump(mode="json") - patch = PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True) - _validate_privacy_patch_against_parent(parent_privacy_policy(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id or (tenant_id if clean_scope == "tenant" else None)), patch) - if clean_scope == "tenant": - tenant = session.get(Tenant, scope_id or tenant_id) - if tenant is None or tenant.id != tenant_id: - raise PrivacyPolicyError("Tenant privacy policy not found") - tenant.settings = _set_settings_privacy_policy(tenant.settings, patch) - session.add(tenant) - return patch - if not scope_id: - raise PrivacyPolicyError(f"{clean_scope.capitalize()} privacy policy requires scope_id") - if clean_scope == "user": - current_settings = _user_settings(session, user_id=scope_id, tenant_id=tenant_id) - if current_settings is None: - raise PrivacyPolicyError("User privacy policy not found") - if _set_user_settings(session, user_id=scope_id, tenant_id=tenant_id, settings_payload=_set_settings_privacy_policy(current_settings, patch)) is None: - raise PrivacyPolicyError("User privacy policy not found") - return patch - if clean_scope == "group": - current_settings = _group_settings(session, group_id=scope_id, tenant_id=tenant_id) - if current_settings is None: - raise PrivacyPolicyError("Group privacy policy not found") - if _set_group_settings(session, group_id=scope_id, tenant_id=tenant_id, settings_payload=_set_settings_privacy_policy(current_settings, patch)) is None: - raise PrivacyPolicyError("Group privacy policy not found") - return patch - if clean_scope == "campaign": - provider = _campaign_policy_provider() - if provider is None: - raise PrivacyPolicyError("Campaign module is not installed") - campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) - settings_payload = _set_settings_privacy_policy(dict(campaign.settings or {}), patch) - try: - provider.set_campaign_settings(session, tenant_id=tenant_id, campaign_id=scope_id, settings=settings_payload) - except ValueError as exc: - raise PrivacyPolicyError("Campaign privacy policy not found") from exc - return patch + return _set_system_privacy_policy(session, policy) + patch = _privacy_policy_patch_payload(policy) + _validate_privacy_policy_patch_for_scope( + session, + tenant_id=tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + patch=patch, + ) + return _set_scoped_privacy_policy( + session, + tenant_id=tenant_id, + scope_type=clean_scope, + scope_id=scope_id, + patch=patch, + ) + + +def _set_system_privacy_policy(session: Session, policy: dict[str, Any] | None) -> dict[str, Any]: + item = get_system_settings(session) + validated = set_privacy_policy(item, PrivacyRetentionPolicy.model_validate(policy or {})) + session.add(item) + return validated.model_dump(mode="json") + + +def _privacy_policy_patch_payload(policy: dict[str, Any] | None) -> dict[str, Any]: + return PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True) + + +def _validate_privacy_policy_patch_for_scope( + session: Session, + *, + tenant_id: str, + scope_type: str, + scope_id: str | None, + patch: dict[str, Any], +) -> None: + parent_scope_id = _parent_privacy_policy_scope_id(tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id) + parent = parent_privacy_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=parent_scope_id) + _validate_privacy_patch_against_parent(parent, patch) + + +def _parent_privacy_policy_scope_id(*, tenant_id: str, scope_type: str, scope_id: str | None) -> str | None: + if scope_id: + return scope_id + if scope_type == "tenant": + return tenant_id + return None + + +def _set_scoped_privacy_policy( + session: Session, + *, + tenant_id: str, + scope_type: str, + scope_id: str | None, + patch: dict[str, Any], +) -> dict[str, Any]: + if scope_type == "tenant": + return _set_tenant_privacy_policy(session, tenant_id=tenant_id, scope_id=scope_id, patch=patch) + clean_scope_id = _required_privacy_policy_scope_id(scope_type, scope_id) + if scope_type == "user": + return _set_user_privacy_policy(session, tenant_id=tenant_id, user_id=clean_scope_id, patch=patch) + if scope_type == "group": + return _set_group_privacy_policy(session, tenant_id=tenant_id, group_id=clean_scope_id, patch=patch) + if scope_type == "campaign": + return _set_campaign_privacy_policy(session, tenant_id=tenant_id, campaign_id=clean_scope_id, patch=patch) raise PrivacyPolicyError("Privacy policy scope must be system, tenant, user, group or campaign") +def _required_privacy_policy_scope_id(scope_type: str, scope_id: str | None) -> str: + if scope_id: + return scope_id + raise PrivacyPolicyError(f"{scope_type.capitalize()} privacy policy requires scope_id") + + +def _set_tenant_privacy_policy( + session: Session, + *, + tenant_id: str, + scope_id: str | None, + patch: dict[str, Any], +) -> dict[str, Any]: + tenant = session.get(Tenant, scope_id or tenant_id) + if tenant is None or tenant.id != tenant_id: + raise PrivacyPolicyError("Tenant privacy policy not found") + tenant.settings = _set_settings_privacy_policy(tenant.settings, patch) + session.add(tenant) + return patch + + +def _set_user_privacy_policy(session: Session, *, tenant_id: str, user_id: str, patch: dict[str, Any]) -> dict[str, Any]: + current_settings = _user_settings(session, user_id=user_id, tenant_id=tenant_id) + if current_settings is None: + raise PrivacyPolicyError("User privacy policy not found") + settings_payload = _set_settings_privacy_policy(current_settings, patch) + if _set_user_settings(session, user_id=user_id, tenant_id=tenant_id, settings_payload=settings_payload) is None: + raise PrivacyPolicyError("User privacy policy not found") + return patch + + +def _set_group_privacy_policy(session: Session, *, tenant_id: str, group_id: str, patch: dict[str, Any]) -> dict[str, Any]: + current_settings = _group_settings(session, group_id=group_id, tenant_id=tenant_id) + if current_settings is None: + raise PrivacyPolicyError("Group privacy policy not found") + settings_payload = _set_settings_privacy_policy(current_settings, patch) + if _set_group_settings(session, group_id=group_id, tenant_id=tenant_id, settings_payload=settings_payload) is None: + raise PrivacyPolicyError("Group privacy policy not found") + return patch + + +def _set_campaign_privacy_policy(session: Session, *, tenant_id: str, campaign_id: str, patch: dict[str, Any]) -> dict[str, Any]: + provider = _campaign_policy_provider() + if provider is None: + raise PrivacyPolicyError("Campaign module is not installed") + campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) + settings_payload = _set_settings_privacy_policy(dict(campaign.settings or {}), patch) + try: + provider.set_campaign_settings(session, tenant_id=tenant_id, campaign_id=campaign_id, settings=settings_payload) + except ValueError as exc: + raise PrivacyPolicyError("Campaign privacy policy not found") from exc + return patch + + def simulate_privacy_policy_change( session: Session, *, diff --git a/tests/test_policy_hierarchy.py b/tests/test_policy_hierarchy.py index 49b7030..150aa1b 100644 --- a/tests/test_policy_hierarchy.py +++ b/tests/test_policy_hierarchy.py @@ -2,6 +2,14 @@ from __future__ import annotations import unittest +from govoplan_policy.backend.retention import ( + PrivacyRetentionPolicy, + PrivacyRetentionPolicyPatch, + PrivacyPolicyError, + _parent_privacy_policy_scope_id, + _privacy_policy_patch_payload, + _required_privacy_policy_scope_id, +) from govoplan_policy.backend.hierarchy import ( PolicyRestrictionRule, simulate_hierarchical_policy_change, @@ -10,6 +18,36 @@ from govoplan_policy.backend.hierarchy import ( class PolicyHierarchyTests(unittest.TestCase): + def test_privacy_policy_parent_scope_ids_follow_scope_precedence(self) -> None: + self.assertEqual( + "tenant-1", + _parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="tenant", scope_id=None), + ) + self.assertIsNone(_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="user", scope_id=None)) + self.assertIsNone(_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="group", scope_id=None)) + self.assertEqual( + "campaign-1", + _parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="campaign", scope_id="campaign-1"), + ) + + def test_privacy_policy_patch_payload_validates_and_removes_unset_fields(self) -> None: + payload = _privacy_policy_patch_payload({"audit_detail_level": "minimal", "generated_eml_retention_days": None}) + + self.assertEqual({"audit_detail_level": "minimal"}, payload) + + def test_privacy_policy_models_extend_shared_schema_contract(self) -> None: + policy = PrivacyRetentionPolicy.model_validate({"generated_eml_retention_days": ""}) + patch = PrivacyRetentionPolicyPatch.model_validate({"allow_lower_level_limits": {"audit_detail_level": False}}) + + self.assertIsNone(policy.generated_eml_retention_days) + self.assertEqual({"audit_detail_level": False}, patch.allow_lower_level_limits) + + def test_required_privacy_policy_scope_id_reports_missing_scope(self) -> None: + with self.assertRaises(PrivacyPolicyError) as captured: + _required_privacy_policy_scope_id("group", None) + + self.assertEqual("Group privacy policy requires scope_id", str(captured.exception)) + def test_parent_locks_block_lower_level_field_changes_and_limit_reenable(self) -> None: issues = validate_hierarchical_policy_patch( parent_policy={"retention_days": 30}, diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py index d15e711..1780da4 100644 --- a/tests/test_policy_module_contract.py +++ b/tests/test_policy_module_contract.py @@ -13,7 +13,7 @@ class PolicyModuleContractTests(unittest.TestCase): project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"] dependencies = tuple(project["dependencies"]) - self.assertIn("govoplan-core>=0.1.6", dependencies) + self.assertTrue(any(item.startswith("govoplan-core>=") for item in dependencies)) self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies)) def test_policy_source_does_not_import_access_implementation(self) -> None: