GovOPlaN/govoplan-views
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(views): add governed DSAR coverage

Browse Source
This commit is contained in:
zemion
2026-08-21 04:02:21 +02:00
parent 24dca43e2e
commit 266f5dae9b
3 changed files with 1444 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_views/backend/dsar_provider.py
+834
View File
@@ -0,0 +1,834 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy import or_
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarExecutionResultRef,
DsarRecordRef,
DsarSubjectRef,
dsar_capability_name,
)
from govoplan_views.backend.db.models import (
ViewAssignment,
ViewDefinition,
ViewPreference,
ViewRevision,
)
VIEWS_DSAR_CAPABILITY = dsar_capability_name("views")
_MAX_RECORDS = 5_000
_MAX_REVISIONS_PER_DEFINITION = 1_000
_MAX_PERSONAL_REVISIONS = 5_000
_CONFLICT = object()
_ATTRIBUTION_TYPES = frozenset(
{
"view_definition_attribution",
"view_revision_attribution",
"view_assignment_attribution",
}
)
_PERSONAL_TYPES = frozenset(
{
"personal_view_assignment",
"personal_view_preference",
"personal_view_definition",
}
)
@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
account_id: str
definition_id: str | None
assignment_id: str | None
preference_id: str | None
class ViewsDsarProvider:
provider_id = "views"
module_id = "views"
def search_subject(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
) -> Sequence[DsarRecordRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
return ()
records: list[DsarRecordRef] = []
personal_revision_count = 0
if selectors.preference_id is None and selectors.assignment_id is None:
definitions = db.query(ViewDefinition).filter(
ViewDefinition.tenant_id == tenant_id,
ViewDefinition.scope_type == "user",
ViewDefinition.scope_id == selectors.account_id,
)
if selectors.definition_id:
definitions = definitions.filter(
ViewDefinition.id == selectors.definition_id
)
for definition in _limited(definitions, label="personal definitions"):
record = _personal_definition_record(db, definition)
revisions = record.data["revisions"]
personal_revision_count += len(revisions) # type: ignore[arg-type]
if personal_revision_count > _MAX_PERSONAL_REVISIONS:
raise ValueError(
"Views DSAR personal revision limit exceeded; narrow the selectors."
)
records.append(record)
if selectors.preference_id is None:
assignments = db.query(ViewAssignment).filter(
ViewAssignment.tenant_id == tenant_id,
ViewAssignment.scope_type == "user",
ViewAssignment.scope_id == selectors.account_id,
)
if selectors.assignment_id:
assignments = assignments.filter(
ViewAssignment.id == selectors.assignment_id
)
if selectors.definition_id:
assignments = assignments.filter(
ViewAssignment.definition_id == selectors.definition_id
)
for assignment in _limited(assignments, label="personal assignments"):
records.append(_personal_assignment_record(assignment))
if selectors.assignment_id is None:
preferences = db.query(ViewPreference).filter(
ViewPreference.tenant_id == tenant_id,
ViewPreference.account_id == selectors.account_id,
)
if selectors.preference_id:
preferences = preferences.filter(
ViewPreference.id == selectors.preference_id
)
if selectors.definition_id:
preferences = preferences.filter(
ViewPreference.view_id == selectors.definition_id
)
for preference in _limited(preferences, label="personal preferences"):
records.append(_personal_preference_record(preference))
if selectors.preference_id is None and selectors.assignment_id is None:
definitions = db.query(ViewDefinition).filter(
ViewDefinition.tenant_id == tenant_id,
or_(
ViewDefinition.scope_type != "user",
ViewDefinition.scope_id != selectors.account_id,
),
or_(
ViewDefinition.created_by == selectors.account_id,
ViewDefinition.updated_by == selectors.account_id,
),
)
if selectors.definition_id:
definitions = definitions.filter(
ViewDefinition.id == selectors.definition_id
)
for definition in _limited(
definitions,
label="definition attribution",
):
records.append(
_definition_attribution_record(definition, selectors.account_id)
)
revisions = (
db.query(ViewRevision)
.join(ViewDefinition, ViewRevision.definition_id == ViewDefinition.id)
.filter(
ViewRevision.tenant_id == tenant_id,
ViewRevision.created_by == selectors.account_id,
or_(
ViewDefinition.scope_type != "user",
ViewDefinition.scope_id != selectors.account_id,
),
)
)
if selectors.definition_id:
revisions = revisions.filter(
ViewRevision.definition_id == selectors.definition_id
)
for revision in _limited(revisions, label="revision attribution"):
records.append(_revision_attribution_record(revision))
if selectors.preference_id is None:
assignments = db.query(ViewAssignment).filter(
ViewAssignment.tenant_id == tenant_id,
or_(
ViewAssignment.scope_type != "user",
ViewAssignment.scope_id != selectors.account_id,
),
or_(
ViewAssignment.created_by == selectors.account_id,
ViewAssignment.updated_by == selectors.account_id,
),
)
if selectors.assignment_id:
assignments = assignments.filter(
ViewAssignment.id == selectors.assignment_id
)
if selectors.definition_id:
assignments = assignments.filter(
ViewAssignment.definition_id == selectors.definition_id
)
for assignment in _limited(
assignments,
label="assignment attribution",
):
records.append(
_assignment_attribution_record(assignment, selectors.account_id)
)
if len(records) > _MAX_RECORDS:
raise ValueError("Views DSAR result limit exceeded; narrow the selectors.")
order = {
"personal_view_assignment": 10,
"personal_view_preference": 20,
"personal_view_definition": 30,
"view_assignment_attribution": 40,
"view_definition_attribution": 50,
"view_revision_attribution": 60,
}
return tuple(
sorted(
records,
key=lambda item: (order[item.resource_type], item.resource_id),
)
)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
raise ValueError("Views DSAR requires one corroborated account.")
actions: list[DsarErasureActionRef] = []
for record in records:
_validate_record(record)
if record.resource_type in _ATTRIBUTION_TYPES:
actions.append(_retain_action(record))
continue
row = _personal_row(
db,
tenant_id=tenant_id,
account_id=selectors.account_id,
resource_type=record.resource_type,
resource_id=record.resource_id,
)
if row is None or not _row_matches_selectors(row, selectors):
actions.append(_manual_action(record, "ownership or selector mismatch"))
continue
if isinstance(row, ViewDefinition) and _has_foreign_dependents(
db,
row,
account_id=selectors.account_id,
):
actions.append(
_manual_action(
record,
"the personal definition has assignments or selections owned by another account",
)
)
continue
version = _row_version(row)
actions.append(
DsarErasureActionRef(
action_id=(
f"views:delete:{record.resource_type}:{record.resource_id}:"
f"{version['token']}"
),
provider_id=self.provider_id,
module_id=self.module_id,
kind="delete",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Delete {record.title}",
rationale=(
"The record is an exact account-owned presentation "
"preference; deleting it does not change domain data."
),
executable=True,
irreversible=True,
metadata={"account_id": selectors.account_id, **version},
)
)
return tuple(actions)
def execute_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
actions: Sequence[DsarErasureActionRef],
request_id: str,
) -> Sequence[DsarExecutionResultRef]:
db = _session(session)
selectors = _subject_selectors(subject)
if selectors is None:
raise ValueError("Views DSAR requires one corroborated account.")
results: list[DsarExecutionResultRef] = []
for action in actions:
_validate_action(action)
if not action.executable or action.kind != "delete":
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"Institutional View configuration attribution remains "
"accountability evidence."
),
)
)
continue
if action.resource_type not in _PERSONAL_TYPES:
raise ValueError("Unsupported executable Views DSAR action.")
row = _personal_row(
db,
tenant_id=tenant_id,
account_id=selectors.account_id,
resource_type=action.resource_type,
resource_id=action.resource_id,
)
if row is None:
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="unchanged",
summary="The personal View record was already absent.",
evidence={"request_id": request_id},
)
)
continue
if str(
action.metadata.get("account_id") or ""
) != selectors.account_id or not _row_matches_selectors(row, selectors):
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"The View record does not match the corroborated "
"account and resource selectors."
),
)
)
continue
if not _version_matches(row, action.metadata):
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"The View record changed after the erasure plan; "
"create a new plan before deleting it."
),
)
)
continue
if isinstance(row, ViewDefinition) and _has_foreign_dependents(
db,
row,
account_id=selectors.account_id,
):
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"The personal View now affects another account; "
"review its assignments and selections manually."
),
)
)
continue
resource_id = str(row.id)
db.delete(row)
db.flush()
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="executed",
summary=(
"Deleted the personal View record without changing any "
"institutional View or domain data."
),
evidence={
"request_id": request_id,
"resource_type": action.resource_type,
"resource_id": resource_id,
},
)
)
return tuple(results)
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
references = subject.external_references
values = {
"account_id": _coalesce(
subject.account_id,
references.get("views.account"),
references.get("access.account"),
),
"definition_id": _coalesce(
references.get("views.definition"),
references.get("views.view"),
),
"assignment_id": _coalesce(
references.get("views.assignment"),
references.get("views.assignment_id"),
),
"preference_id": _coalesce(
references.get("views.preference"),
references.get("views.preference_id"),
),
}
if any(value is _CONFLICT for value in values.values()):
return None
account_id = _optional_string(values["account_id"])
if account_id is None:
return None
return _SubjectSelectors(
account_id=account_id,
definition_id=_optional_string(values["definition_id"]),
assignment_id=_optional_string(values["assignment_id"]),
preference_id=_optional_string(values["preference_id"]),
)
def _coalesce(*values: str | None) -> str | None | object:
normalized = {str(value).strip() for value in values if str(value or "").strip()}
if len(normalized) > 1:
return _CONFLICT
return next(iter(normalized), None)
def _optional_string(value: object) -> str | None:
return value if isinstance(value, str) and value else None
def _limited(query, *, label: str):
rows = (
query.order_by(query.column_descriptions[0]["entity"].id)
.limit(_MAX_RECORDS + 1)
.all()
)
if len(rows) > _MAX_RECORDS:
raise ValueError(f"Views DSAR {label} limit exceeded; narrow the selectors.")
return rows
def _personal_definition_record(
session: Session,
definition: ViewDefinition,
) -> DsarRecordRef:
revisions = (
session.query(ViewRevision)
.filter(ViewRevision.definition_id == definition.id)
.order_by(ViewRevision.revision, ViewRevision.id)
.limit(_MAX_REVISIONS_PER_DEFINITION + 1)
.all()
)
if len(revisions) > _MAX_REVISIONS_PER_DEFINITION:
raise ValueError(
"Views DSAR personal definition revision limit exceeded; narrow the selectors."
)
return DsarRecordRef(
provider_id="views",
module_id="views",
resource_type="personal_view_definition",
resource_id=definition.id,
category="personal_interface_preference",
title=f"Personal View: {definition.name[:200]}",
data={
"definition_key": definition.definition_key,
"name": definition.name[:200],
"description": (definition.description or "")[:2_000] or None,
"status": definition.status,
"current_revision": definition.current_revision,
"published_revision_id": definition.published_revision_id,
"deleted_at": _iso(definition.deleted_at),
"revisions": [
{
"id": revision.id,
"revision": revision.revision,
"surface_contract_version": revision.surface_contract_version,
"visible_surface_ids": _string_list(
revision.visible_surface_ids,
limit=1_000,
),
"presentation": _presentation_projection(revision.presentation),
"created_at": _iso(revision.created_at),
}
for revision in revisions
],
},
observed_at=_aware(definition.updated_at),
)
def _personal_assignment_record(assignment: ViewAssignment) -> DsarRecordRef:
return DsarRecordRef(
provider_id="views",
module_id="views",
resource_type="personal_view_assignment",
resource_id=assignment.id,
category="personal_interface_preference",
title="Personal View assignment",
data={
"definition_id": assignment.definition_id,
"revision_id": assignment.revision_id,
"mode": assignment.mode,
"priority": assignment.priority,
"is_active": assignment.is_active,
},
observed_at=_aware(assignment.updated_at),
)
def _personal_preference_record(preference: ViewPreference) -> DsarRecordRef:
return DsarRecordRef(
provider_id="views",
module_id="views",
resource_type="personal_view_preference",
resource_id=preference.id,
category="personal_interface_preference",
title="Personal active-View selection",
data={
"selection_kind": preference.selection_kind,
"view_id": preference.view_id,
},
observed_at=_aware(preference.updated_at),
)
def _definition_attribution_record(
definition: ViewDefinition,
account_id: str,
) -> DsarRecordRef:
activities = []
if definition.created_by == account_id:
activities.append("created_view_definition")
if definition.updated_by == account_id:
activities.append("updated_view_definition")
return _attribution_record(
"view_definition_attribution",
definition.id,
"View-definition attribution",
{
"activities": activities,
"scope_type": definition.scope_type,
"current_revision": definition.current_revision,
"status": definition.status,
},
observed_at=definition.updated_at,
)
def _revision_attribution_record(revision: ViewRevision) -> DsarRecordRef:
return _attribution_record(
"view_revision_attribution",
revision.id,
"View-revision attribution",
{
"activity": "created_view_revision",
"definition_id": revision.definition_id,
"revision": revision.revision,
"surface_contract_version": revision.surface_contract_version,
},
observed_at=revision.created_at,
)
def _assignment_attribution_record(
assignment: ViewAssignment,
account_id: str,
) -> DsarRecordRef:
activities = []
if assignment.created_by == account_id:
activities.append("created_view_assignment")
if assignment.updated_by == account_id:
activities.append("updated_view_assignment")
return _attribution_record(
"view_assignment_attribution",
assignment.id,
"View-assignment attribution",
{
"activities": activities,
"scope_type": assignment.scope_type,
"definition_id": assignment.definition_id,
"mode": assignment.mode,
"is_active": assignment.is_active,
},
observed_at=assignment.updated_at,
)
def _attribution_record(
resource_type: str,
resource_id: str,
title: str,
data: Mapping[str, object],
*,
observed_at: datetime | None,
) -> DsarRecordRef:
return DsarRecordRef(
provider_id="views",
module_id="views",
resource_type=resource_type,
resource_id=resource_id,
category="operator_accountability_evidence",
title=title,
data=data,
observed_at=_aware(observed_at),
immutable_evidence=True,
retention_reason=(
"Institutional View configuration attribution is retained for "
"accountability; definition, presentation, and metadata payloads are excluded."
),
)
def _personal_row(
session: Session,
*,
tenant_id: str,
account_id: str,
resource_type: str,
resource_id: str,
) -> ViewAssignment | ViewPreference | ViewDefinition | None:
if resource_type == "personal_view_assignment":
return (
session.query(ViewAssignment)
.filter(
ViewAssignment.id == resource_id,
ViewAssignment.tenant_id == tenant_id,
ViewAssignment.scope_type == "user",
ViewAssignment.scope_id == account_id,
)
.one_or_none()
)
if resource_type == "personal_view_preference":
return (
session.query(ViewPreference)
.filter(
ViewPreference.id == resource_id,
ViewPreference.tenant_id == tenant_id,
ViewPreference.account_id == account_id,
)
.one_or_none()
)
if resource_type == "personal_view_definition":
return (
session.query(ViewDefinition)
.filter(
ViewDefinition.id == resource_id,
ViewDefinition.tenant_id == tenant_id,
ViewDefinition.scope_type == "user",
ViewDefinition.scope_id == account_id,
)
.one_or_none()
)
raise ValueError("Unsupported personal Views DSAR record type.")
def _row_matches_selectors(
row: ViewAssignment | ViewPreference | ViewDefinition,
selectors: _SubjectSelectors,
) -> bool:
if selectors.assignment_id and (
not isinstance(row, ViewAssignment) or row.id != selectors.assignment_id
):
return False
if selectors.preference_id and (
not isinstance(row, ViewPreference) or row.id != selectors.preference_id
):
return False
if selectors.definition_id:
if isinstance(row, ViewDefinition):
return row.id == selectors.definition_id
return (
row.view_id == selectors.definition_id
if isinstance(row, ViewPreference)
else row.definition_id == selectors.definition_id
)
return True
def _has_foreign_dependents(
session: Session,
definition: ViewDefinition,
*,
account_id: str,
) -> bool:
foreign_assignment = (
session.query(ViewAssignment.id)
.filter(
ViewAssignment.definition_id == definition.id,
or_(
ViewAssignment.tenant_id != definition.tenant_id,
ViewAssignment.scope_type != "user",
ViewAssignment.scope_id != account_id,
),
)
.first()
)
foreign_preference = (
session.query(ViewPreference.id)
.filter(
ViewPreference.view_id == definition.id,
or_(
ViewPreference.tenant_id != definition.tenant_id,
ViewPreference.account_id != account_id,
),
)
.first()
)
return foreign_assignment is not None or foreign_preference is not None
def _row_version(
row: ViewAssignment | ViewPreference | ViewDefinition,
) -> dict[str, object]:
updated_at = _iso(row.updated_at)
if isinstance(row, ViewDefinition):
return {
"token": f"r{row.current_revision}",
"current_revision": row.current_revision,
"updated_at": updated_at,
}
return {"token": "timestamp", "updated_at": updated_at}
def _version_matches(
row: ViewAssignment | ViewPreference | ViewDefinition,
metadata: Mapping[str, object],
) -> bool:
if metadata.get("updated_at") != _iso(row.updated_at):
return False
if isinstance(row, ViewDefinition):
return metadata.get("current_revision") == row.current_revision
return True
def _presentation_projection(value: object) -> dict[str, object]:
if not isinstance(value, Mapping):
raise ValueError("View presentation must be an object.")
projected: dict[str, object] = {}
mode = value.get("navigation_mode")
if isinstance(mode, str):
projected["navigation_mode"] = mode[:20]
for key in (
"product_area_order",
"quick_access_recommended_tool_ids",
"quick_access_focused_tool_ids",
):
raw = value.get(key)
if raw is not None:
projected[key] = _string_list(raw, limit=100)
labels = value.get("product_area_labels")
if labels is not None:
if not isinstance(labels, Mapping) or len(labels) > 100:
raise ValueError("View presentation labels exceed the DSAR bound.")
projected["product_area_labels"] = {
str(key)[:80]: str(label)[:200] for key, label in labels.items()
}
return projected
def _string_list(value: object, *, limit: int) -> list[str]:
if not isinstance(value, list) or len(value) > limit:
raise ValueError("View list payload exceeds the DSAR bound.")
return [str(item)[:255] for item in value]
def _retain_action(record: DsarRecordRef) -> DsarErasureActionRef:
return DsarErasureActionRef(
action_id=f"views:retain:{record.resource_type}:{record.resource_id}",
provider_id="views",
module_id="views",
kind="retain",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Retain {record.title}",
rationale=record.retention_reason
or "Institutional View attribution is accountability evidence.",
executable=False,
)
def _manual_action(record: DsarRecordRef, reason: str) -> DsarErasureActionRef:
return DsarErasureActionRef(
action_id=f"views:manual_review:{record.resource_type}:{record.resource_id}",
provider_id="views",
module_id="views",
kind="manual_review",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Review {record.title}",
rationale=f"Automatic deletion stopped because of {reason}.",
executable=False,
)
def _iso(value: datetime | None) -> str | None:
aware = _aware(value)
return aware.isoformat() if aware else None
def _aware(value: datetime | None) -> datetime | None:
if value is None or value.tzinfo is not None:
return value
return value.replace(tzinfo=timezone.utc)
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Views DSAR requires a SQLAlchemy Session.")
return value
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "views" or record.module_id != "views":
raise ValueError("Views DSAR cannot plan a foreign provider record.")
if record.resource_type not in _PERSONAL_TYPES | _ATTRIBUTION_TYPES:
raise ValueError("Views DSAR record type is invalid.")
if not record.resource_id:
raise ValueError("Views DSAR record identity is incomplete.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "views" or action.module_id != "views":
raise ValueError("Views DSAR cannot execute a foreign provider action.")
if not action.action_id.startswith("views:"):
raise ValueError("Views DSAR action identity is invalid.")
__all__ = ["VIEWS_DSAR_CAPABILITY", "ViewsDsarProvider"]
src/govoplan_views/backend/manifest.py
+88 -3
View File
@@ -11,6 +11,7 @@ from govoplan_core.core.module_guards import (
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationLink,
DocumentationTopic,
FrontendModule,
@@ -26,6 +27,10 @@ from govoplan_core.core.policy import CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX
from govoplan_core.core.views import CAPABILITY_VIEWS_RESOLVER, ViewSurface
from govoplan_core.db.base import Base
from govoplan_views.backend.db import models as view_models
from govoplan_views.backend.dsar_provider import (
VIEWS_DSAR_CAPABILITY,
ViewsDsarProvider,
)
MODULE_ID = "views"
@@ -216,6 +221,10 @@ def _policy_impact_subjects(context: ModuleContext):
return ViewsPolicyImpactSubjectProvider(context.registry)
def _dsar_provider(_context: ModuleContext) -> ViewsDsarProvider:
return ViewsDsarProvider()
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
@@ -228,6 +237,7 @@ manifest = ModuleManifest(
provides_interfaces=(
ModuleInterfaceProvider(name="views.surface_contract", version="1.0.0"),
ModuleInterfaceProvider(name="views.resolver", version="0.1.0"),
ModuleInterfaceProvider(name=VIEWS_DSAR_CAPABILITY, version="0.1.0"),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
@@ -298,8 +308,71 @@ manifest = ModuleManifest(
capability_factories={
CAPABILITY_VIEWS_RESOLVER: _resolver,
f"{CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX}views": _policy_impact_subjects,
VIEWS_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={
VIEWS_DSAR_CAPABILITY: CapabilityDocumentation(
label="Views data-subject request provider",
summary=(
"Exports and deletes account-owned View preferences while "
"retaining minimized institutional configuration attribution."
),
contract_version="0.1.0",
),
},
documentation=(
DocumentationTopic(
id="views.data-subject-requests",
title="Views data-subject requests",
summary=(
"Export or delete personal Views, assignments, and selections "
"without changing institutional projections or domain data."
),
body=(
"Views correlates one exact account in the active tenant and can "
"narrow the request to a definition, assignment, or preference. "
"The access package includes bounded personal definition and "
"revision presentation, user assignments, and the active-View "
"selection. Arbitrary assignment metadata is excluded and View "
"surfaces are identifiers only; the provider never traverses them "
"into feature data. Tenant and group configuration authored or "
"updated by the subject contributes minimized attribution only and "
"is retained as institutional accountability evidence. System-wide "
"Views are outside tenant-scoped requests. Erasure can delete an "
"exact personal selection, user assignment, or user-owned definition "
"after timestamp, revision, ownership, and dependent-record checks. "
"A personal definition that affects another account is sent to "
"manual review. Deleting a selection or assignment never deletes "
"the referenced institutional View. Repeated execution is unchanged."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "tenant_admin", "operator", "auditor"),
related_modules=("core", "access", "quick_access"),
metadata={
"help_contexts": [
"views.selector",
"views.settings.personal",
"views.admin.tenant",
"privacy.data-subject-requests",
],
"consequence_classes": {
"delete_selection": (
"Removes the account selection; effective defaults apply again."
),
"delete_assignment": (
"Removes only the personal assignment, not its View definition."
),
"delete_personal_definition": (
"Removes the personal definition and revisions only when no "
"other account depends on it."
),
"retain_attribution": (
"Preserves minimized tenant/group configuration accountability."
),
},
},
),
DocumentationTopic(
id="views.interface-projections",
title="Task-focused Views",
@@ -443,9 +516,21 @@ manifest = ModuleManifest(
maturity="vertical_slice",
documentation_ref="README.md",
test_ref="tests/test_views.py",
known_limits=("A View filters presentation only; modules still vary in the granularity of announced surfaces.",),
owned_concepts=("view definition", "view revision", "view assignment", "view selection"),
non_owned_concepts=("authorization", "module navigation", "workflow definition", "dashboard layout"),
known_limits=(
"A View filters presentation only; modules still vary in the granularity of announced surfaces.",
),
owned_concepts=(
"view definition",
"view revision",
"view assignment",
"view selection",
),
non_owned_concepts=(
"authorization",
"module navigation",
"workflow definition",
"dashboard layout",
),
recovery_docs=("README.md",),
security_docs=("README.md",),
),
tests/test_dsar_provider.py
+522
View File
@@ -0,0 +1,522 @@
from __future__ import annotations
import json
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarProvider,
DsarRecordRef,
DsarSubjectRef,
)
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
create_data_subject_request,
search_data_subject_request,
)
from govoplan_views.backend.db.models import (
ViewAssignment,
ViewDefinition,
ViewPreference,
ViewRevision,
)
from govoplan_views.backend.dsar_provider import (
VIEWS_DSAR_CAPABILITY,
ViewsDsarProvider,
)
from govoplan_views.backend.manifest import manifest
class _Registry:
def __init__(self, provider: ViewsDsarProvider, *, active: bool = True) -> None:
self.provider = provider
self.active = active
def capability_names(self):
return (VIEWS_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "views"
def tenant_entitlement_resolver(self):
active = self.active
class _Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("views",) if active else ()},
)()
return _Resolver()
def require_tenant_capability(self, name, session, **kwargs):
del session, kwargs
self._assert_capability(name)
return self.provider
def manifests(self):
return (type("Manifest", (), {"id": "views"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != VIEWS_DSAR_CAPABILITY:
raise KeyError(name)
class ViewsDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session = Session(self.engine)
self.provider = ViewsDsarProvider()
self.assertIsInstance(self.provider, DsarProvider)
self._seed()
self.session.commit()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def _definition(
self,
definition_id: str,
*,
tenant_id: str,
scope_type: str,
scope_id: str,
actor_id: str,
name: str,
) -> ViewDefinition:
definition = ViewDefinition(
id=definition_id,
tenant_id=tenant_id,
scope_type=scope_type,
scope_id=scope_id,
scope_key=f"{scope_type}:{tenant_id}:{scope_id}",
definition_key=definition_id,
name=name,
description=f"Description for {name}",
status="published",
current_revision=1,
published_revision_id=f"revision-{definition_id}",
created_by=actor_id,
updated_by=actor_id,
)
definition.revisions.append(
ViewRevision(
id=f"revision-{definition_id}",
tenant_id=tenant_id,
revision=1,
surface_contract_version="1.0.0",
visible_surface_ids=["files.route.files"],
presentation={
"navigation_mode": "flat",
"quick_access_focused_tool_ids": ["files.recent"],
"private_payload_do_not_export": f"private-{definition_id}",
},
content_hash=("a" * 64),
created_by=actor_id,
)
)
return definition
def _seed(self) -> None:
personal = self._definition(
"definition-personal",
tenant_id="tenant-1",
scope_type="user",
scope_id="account-1",
actor_id="account-1",
name="My personal work",
)
tenant = self._definition(
"definition-tenant",
tenant_id="tenant-1",
scope_type="tenant",
scope_id="tenant-1",
actor_id="account-1",
name="private-tenant-definition-do-not-export",
)
other_account = self._definition(
"definition-other-account",
tenant_id="tenant-1",
scope_type="user",
scope_id="account-other",
actor_id="account-other",
name="Other account",
)
other_tenant = self._definition(
"definition-other-tenant",
tenant_id="tenant-2",
scope_type="user",
scope_id="account-1",
actor_id="account-1",
name="Other tenant",
)
self.session.add_all((personal, tenant, other_account, other_tenant))
self.session.flush()
self.session.add_all(
(
ViewAssignment(
id="assignment-personal",
tenant_id="tenant-1",
scope_type="user",
scope_id="account-1",
target_key="user:tenant-1:account-1",
definition_id=personal.id,
revision_id=None,
mode="available",
priority=1,
is_active=True,
metadata_={"private": "personal-metadata-do-not-export"},
created_by="account-1",
updated_by="account-1",
),
ViewAssignment(
id="assignment-tenant",
tenant_id="tenant-1",
scope_type="tenant",
scope_id="tenant-1",
target_key="tenant:tenant-1",
definition_id=tenant.id,
revision_id=None,
mode="default",
priority=2,
is_active=True,
metadata_={"private": "tenant-metadata-do-not-export"},
created_by="account-1",
updated_by="account-other",
),
ViewPreference(
id="preference-personal",
tenant_id="tenant-1",
account_id="account-1",
selection_kind="selected",
view_id=personal.id,
),
ViewPreference(
id="preference-other-account",
tenant_id="tenant-1",
account_id="account-other",
selection_kind="auto",
view_id=None,
),
ViewPreference(
id="preference-other-tenant",
tenant_id="tenant-2",
account_id="account-1",
selection_kind="auto",
view_id=None,
),
)
)
def test_search_exports_personal_records_and_minimized_attribution(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(account_id="account-1"),
)
self.assertEqual(
[
"personal_view_assignment",
"personal_view_preference",
"personal_view_definition",
"view_assignment_attribution",
"view_definition_attribution",
"view_revision_attribution",
],
[record.resource_type for record in records],
)
exported = json.dumps([record.to_dict() for record in records])
self.assertIn("My personal work", exported)
self.assertIn("files.recent", exported)
self.assertNotIn("personal-metadata-do-not-export", exported)
self.assertNotIn("tenant-metadata-do-not-export", exported)
self.assertNotIn("private-tenant-definition-do-not-export", exported)
self.assertNotIn("definition-other-account", exported)
self.assertNotIn("definition-other-tenant", exported)
def test_resource_references_narrow_and_conflicts_fail_closed(self) -> None:
definition = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"views.definition": "definition-personal"},
),
)
assignment = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"views.assignment": "assignment-personal"},
),
)
conflict = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"views.account": "account-other"},
),
)
no_account = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={"views.preference": "preference-personal"}
),
)
self.assertEqual(
{
"personal_view_assignment",
"personal_view_preference",
"personal_view_definition",
},
{
item.resource_type
for item in definition
if item.resource_type.startswith("personal_")
},
)
self.assertEqual(
["personal_view_assignment"],
[item.resource_type for item in assignment],
)
self.assertEqual((), conflict)
self.assertEqual((), no_account)
def test_erasure_deletes_personal_records_and_retains_institutional_ones(
self,
) -> None:
subject = DsarSubjectRef(account_id="account-1")
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=records,
)
self.assertEqual(
["delete", "delete", "delete", "retain", "retain", "retain"],
[action.kind for action in actions],
)
first = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
second = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
self.assertEqual(
["executed", "executed", "executed", "blocked", "blocked", "blocked"],
[result.status for result in first],
)
self.assertEqual(
["unchanged", "unchanged", "unchanged", "blocked", "blocked", "blocked"],
[result.status for result in second],
)
self.assertIsNone(self.session.get(ViewAssignment, "assignment-personal"))
self.assertIsNone(self.session.get(ViewPreference, "preference-personal"))
self.assertIsNone(self.session.get(ViewDefinition, "definition-personal"))
self.assertIsNone(
self.session.get(ViewRevision, "revision-definition-personal")
)
self.assertIsNotNone(self.session.get(ViewDefinition, "definition-tenant"))
self.assertIsNotNone(self.session.get(ViewAssignment, "assignment-tenant"))
def test_foreign_dependency_and_changed_definition_block_deletion(self) -> None:
subject = DsarSubjectRef(
account_id="account-1",
external_references={"views.definition": "definition-personal"},
)
definition_record = next(
item
for item in self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=subject,
)
if item.resource_type == "personal_view_definition"
)
definition = self.session.get(ViewDefinition, "definition-personal")
definition.current_revision += 1
stale_action = DsarErasureActionRef(
action_id="views:delete:personal_view_definition:definition-personal:r1",
provider_id="views",
module_id="views",
kind="delete",
resource_type="personal_view_definition",
resource_id="definition-personal",
title="Delete personal View",
rationale="Personal preference",
executable=True,
metadata={
"account_id": "account-1",
"current_revision": 1,
"updated_at": definition_record.observed_at.isoformat(),
},
)
changed = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(stale_action,),
request_id="dsar-1",
)
self.assertEqual("blocked", changed[0].status)
definition.current_revision = 1
self.session.add(
ViewAssignment(
id="assignment-foreign-dependent",
tenant_id="tenant-1",
scope_type="tenant",
scope_id="tenant-1",
target_key="tenant:tenant-1:foreign-dependent",
definition_id=definition.id,
revision_id=None,
mode="available",
priority=0,
is_active=True,
metadata_={},
created_by="account-other",
updated_by="account-other",
)
)
self.session.flush()
manual = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(definition_record,),
)
self.assertEqual("manual_review", manual[0].kind)
self.assertFalse(manual[0].executable)
def test_foreign_records_and_actions_are_rejected(self) -> None:
subject = DsarSubjectRef(account_id="account-1")
with self.assertRaisesRegex(ValueError, "foreign provider record"):
self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(
DsarRecordRef(
provider_id="dashboard",
module_id="dashboard",
resource_type="personal_view_preference",
resource_id="preference-personal",
category="preference",
title="Foreign preference",
),
),
)
with self.assertRaisesRegex(ValueError, "foreign provider action"):
self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(
DsarErasureActionRef(
action_id="dashboard:delete:view:preference-personal",
provider_id="dashboard",
module_id="dashboard",
kind="delete",
resource_type="personal_view_preference",
resource_id="preference-personal",
title="Delete preference",
rationale="Foreign action",
executable=True,
),
),
request_id="dsar-1",
)
def test_core_workflow_and_manifest_register_provider(self) -> None:
row = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-VIEWS-1",
request_kind="access",
subject=DsarSubjectRef(account_id="account-1"),
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider),
row=row,
expected_revision=1,
)
self.assertEqual([VIEWS_DSAR_CAPABILITY], row.coverage["provider_capabilities"])
self.assertEqual(6, row.search_result["record_count"])
inactive = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-VIEWS-2",
request_kind="access",
subject=DsarSubjectRef(account_id="account-1"),
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider, active=False),
row=inactive,
expected_revision=1,
)
self.assertEqual(
[VIEWS_DSAR_CAPABILITY],
inactive.coverage["inactive_provider_capabilities"],
)
self.assertIn(VIEWS_DSAR_CAPABILITY, manifest.capability_factories)
self.assertIn(VIEWS_DSAR_CAPABILITY, manifest.capability_documentation)
self.assertIn(
VIEWS_DSAR_CAPABILITY,
{item.name for item in manifest.provides_interfaces},
)
self.assertTrue(
any(
topic.id == "views.data-subject-requests"
and {"admin", "user"}.issubset(topic.documentation_types)
for topic in manifest.documentation
)
)
if __name__ == "__main__":
unittest.main()