intermittent commit

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,

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")
 

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,
     *,

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},

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: