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

feat(tasks): add governed DSAR coverage

Browse Source
This commit is contained in:
zemion
2026-08-21 04:21:01 +02:00
parent 39bb6c0d18
commit 88bd0e6aae
3 changed files with 856 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_tasks/backend/dsar_provider.py
+410
View File
@@ -0,0 +1,410 @@
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_tasks.backend.db.models import TaskAssignment, TaskItem
TASKS_DSAR_CAPABILITY = dsar_capability_name("tasks")
_MAX_RECORDS = 5_000
_MAX_SOURCES = 100
_CONFLICT = object()
@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
account_id: str | None
actor_ids: tuple[str, ...]
task_id: str | None
class TasksDsarProvider:
provider_id = "tasks"
module_id = "tasks"
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 ()
assigned_ids: set[str] = set()
if selectors.account_id:
assigned = (
db.query(TaskAssignment.task_id)
.join(TaskItem, TaskAssignment.task_id == TaskItem.id)
.filter(
TaskItem.tenant_id == tenant_id,
TaskAssignment.tenant_id == tenant_id,
TaskAssignment.assignment_kind == "account",
TaskAssignment.assignment_id == selectors.account_id,
)
)
if selectors.task_id:
assigned = assigned.filter(TaskItem.id == selectors.task_id)
rows = assigned.limit(_MAX_RECORDS + 1).all()
if len(rows) > _MAX_RECORDS:
raise ValueError(
"Tasks DSAR assignment result limit exceeded; narrow the selectors."
)
assigned_ids = {str(task_id) for (task_id,) in rows}
actor_query = db.query(TaskItem.id).filter(
TaskItem.tenant_id == tenant_id,
or_(
TaskItem.created_by.in_(selectors.actor_ids),
TaskItem.updated_by.in_(selectors.actor_ids),
TaskItem.completed_by.in_(selectors.actor_ids),
),
)
if selectors.task_id:
actor_query = actor_query.filter(TaskItem.id == selectors.task_id)
actor_rows = actor_query.limit(_MAX_RECORDS + 1).all()
if len(actor_rows) > _MAX_RECORDS:
raise ValueError(
"Tasks DSAR actor result limit exceeded; narrow the selectors."
)
actor_ids = {str(task_id) for (task_id,) in actor_rows}
task_ids = assigned_ids | actor_ids
if len(task_ids) > _MAX_RECORDS:
raise ValueError("Tasks DSAR result limit exceeded; narrow the selectors.")
if not task_ids:
return ()
tasks = (
db.query(TaskItem)
.filter(
TaskItem.tenant_id == tenant_id,
TaskItem.id.in_(task_ids),
)
.order_by(TaskItem.created_at, TaskItem.id)
.all()
)
return tuple(
_assigned_task_record(
task,
account_id=selectors.account_id,
actor_ids=selectors.actor_ids,
)
if task.id in assigned_ids
else _actor_attribution_record(task, selectors.actor_ids)
for task in tasks
)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
del tenant_id
_session(session)
if _subject_selectors(subject) is None:
raise ValueError("Tasks DSAR subject selectors conflict.")
actions: list[DsarErasureActionRef] = []
for record in records:
_validate_record(record)
if record.resource_type == "assigned_task":
actions.append(
DsarErasureActionRef(
action_id=f"tasks:manual_review:assigned_task:{record.resource_id}",
provider_id=self.provider_id,
module_id=self.module_id,
kind="manual_review",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Review {record.title}",
rationale=(
"The account assignment and task content may be shared "
"institutional work. Its source owner and retention state "
"must be reviewed before detachment or minimization."
),
executable=False,
)
)
else:
actions.append(
DsarErasureActionRef(
action_id=(
f"tasks:retain:task_actor_attribution:{record.resource_id}"
),
provider_id=self.provider_id,
module_id=self.module_id,
kind="retain",
resource_type=record.resource_type,
resource_id=record.resource_id,
title=f"Retain {record.title}",
rationale=record.retention_reason
or "Task lifecycle attribution is accountability evidence.",
executable=False,
)
)
return tuple(actions)
def execute_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
actions: Sequence[DsarErasureActionRef],
request_id: str,
) -> Sequence[DsarExecutionResultRef]:
del tenant_id
_session(session)
if _subject_selectors(subject) is None:
raise ValueError("Tasks DSAR subject selectors conflict.")
results: list[DsarExecutionResultRef] = []
for action in actions:
_validate_action(action)
if action.executable or action.kind not in {"retain", "manual_review"}:
raise ValueError("Tasks DSAR publishes non-executable actions only.")
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"The Task remains unchanged pending its institutional "
"retention and source-owner review."
if action.kind == "manual_review"
else "Task lifecycle attribution remains immutable evidence."
),
evidence={"request_id": request_id},
)
)
return tuple(results)
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
references = subject.external_references
values = {
"account_id": _coalesce(
subject.account_id,
references.get("tasks.account"),
references.get("access.account"),
),
"membership_id": _coalesce(
subject.membership_id,
references.get("tasks.membership"),
references.get("tenancy.membership"),
),
"identity_id": _coalesce(
subject.identity_id,
references.get("tasks.identity"),
references.get("identity.id"),
),
"task_id": _coalesce(
references.get("tasks.task"),
references.get("tasks.item"),
),
}
if any(value is _CONFLICT for value in values.values()):
return None
actor_ids = tuple(
dict.fromkeys(
value
for key in ("account_id", "membership_id", "identity_id")
if (value := _optional_string(values[key]))
)
)
if not actor_ids:
return None
return _SubjectSelectors(
account_id=_optional_string(values["account_id"]),
actor_ids=actor_ids,
task_id=_optional_string(values["task_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 _assigned_task_record(
task: TaskItem,
*,
account_id: str | None,
actor_ids: Sequence[str],
) -> DsarRecordRef:
assignments = [
{
"id": assignment.id,
"kind": assignment.assignment_kind,
"assignment_id": assignment.assignment_id,
"label": (assignment.assignment_label or "")[:500] or None,
}
for assignment in task.assignments
if account_id
and assignment.assignment_kind == "account"
and assignment.assignment_id == account_id
]
if len(assignments) > 100:
raise ValueError("Task account assignments exceed the DSAR bound.")
return DsarRecordRef(
provider_id="tasks",
module_id="tasks",
resource_type="assigned_task",
resource_id=task.id,
category="assigned_institutional_work",
title=f"Assigned task: {task.title[:500]}",
data={
"title": task.title[:500],
"summary": (task.summary or "")[:4_000] or None,
"status": task.status,
"priority": task.priority,
"required_action": (task.required_action or "")[:500] or None,
"action_url": (task.action_url or "")[:1_500] or None,
"due_at": _iso(task.due_at),
"deferred_until": _iso(task.deferred_until),
"completed_at": _iso(task.completed_at),
"cancelled_at": _iso(task.cancelled_at),
"revision": task.revision,
"assignments": assignments,
"sources": _source_projection(task.sources),
"actor_activities": _actor_activities(task, actor_ids),
"created_at": _iso(task.created_at),
"updated_at": _iso(task.updated_at),
},
observed_at=_aware(task.updated_at),
retention_reason=(
"The task may be shared institutional work and requires source-owner "
"and retention review before its account assignment can be changed."
),
)
def _actor_attribution_record(
task: TaskItem,
actor_ids: Sequence[str],
) -> DsarRecordRef:
return DsarRecordRef(
provider_id="tasks",
module_id="tasks",
resource_type="task_actor_attribution",
resource_id=task.id,
category="operator_accountability_evidence",
title="Task lifecycle attribution",
data={
"activities": _actor_activities(task, actor_ids),
"status": task.status,
"priority": task.priority,
"source_module": task.source_module,
"source_resource_type": task.source_resource_type,
"source_resource_id": task.source_resource_id,
"source_revision": task.source_revision,
"revision": task.revision,
"created_at": _iso(task.created_at),
"updated_at": _iso(task.updated_at),
"completed_at": _iso(task.completed_at),
},
observed_at=_aware(task.updated_at),
immutable_evidence=True,
retention_reason=(
"Task creation, update, and completion attribution is immutable "
"accountability evidence; task content and metadata are excluded."
),
)
def _actor_activities(task: TaskItem, actor_ids: Sequence[str]) -> list[str]:
actor_set = set(actor_ids)
activities = []
if task.created_by in actor_set:
activities.append("created_task")
if task.updated_by in actor_set:
activities.append("updated_task")
if task.completed_by in actor_set:
activities.append("completed_task")
return activities
def _source_projection(value: object) -> list[dict[str, str | None]]:
if not isinstance(value, list) or len(value) > _MAX_SOURCES:
raise ValueError("Tasks source references exceed the DSAR bound.")
projected: list[dict[str, str | None]] = []
for item in value:
if not isinstance(item, Mapping):
raise ValueError("Task source reference is invalid.")
projected.append(
{
"module_id": _bounded(item.get("module_id"), 100),
"resource_type": _bounded(item.get("resource_type"), 100),
"resource_id": _bounded(item.get("resource_id"), 255),
"revision": _bounded(item.get("revision"), 255),
"url": _bounded(item.get("url"), 1_500),
"label": _bounded(item.get("label"), 500),
}
)
return projected
def _bounded(value: object, limit: int) -> str | None:
return str(value)[:limit] if value is not None else None
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("Tasks DSAR requires a SQLAlchemy Session.")
return value
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "tasks" or record.module_id != "tasks":
raise ValueError("Tasks DSAR cannot plan a foreign provider record.")
if (
record.resource_type
not in {
"assigned_task",
"task_actor_attribution",
}
or not record.resource_id
):
raise ValueError("Tasks DSAR record identity is invalid.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "tasks" or action.module_id != "tasks":
raise ValueError("Tasks DSAR cannot execute a foreign provider action.")
if not action.action_id.startswith("tasks:"):
raise ValueError("Tasks DSAR action identity is invalid.")
__all__ = ["TASKS_DSAR_CAPABILITY", "TasksDsarProvider"]
src/govoplan_tasks/backend/manifest.py
+70 -2
View File
@@ -34,6 +34,10 @@ from govoplan_core.core.tasks import (
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_tasks.backend.db import models as task_models
from govoplan_tasks.backend.dsar_provider import (
TASKS_DSAR_CAPABILITY,
TasksDsarProvider,
)
from govoplan_tasks.backend.service import SqlTaskService
@@ -69,6 +73,10 @@ def _service(context: ModuleContext) -> SqlTaskService:
return SqlTaskService(context.registry)
def _dsar_provider(_context: ModuleContext) -> TasksDsarProvider:
return TasksDsarProvider()
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
total = (
session.query(task_models.TaskItem)
@@ -122,6 +130,54 @@ ROLE_TEMPLATES = (
)
DOCUMENTATION = (
DocumentationTopic(
id="tasks.data-subject-requests",
title="Task data-subject requests",
summary=(
"Export account-assigned task data and lifecycle attribution while "
"keeping shared institutional work under owner review."
),
body=(
"Tasks correlates exact account and actor identifiers only inside the "
"active tenant. Account-assigned explicit Tasks contribute bounded "
"title, summary, required action, lifecycle state, due dates, the exact "
"matching account assignment, source references, and any create, update, "
"or completion activities performed by the subject. Group, role, function, "
"and anyone visibility is not inferred from external directories and other "
"assignment targets are excluded. When the subject acted on a Task without "
"being its direct account assignee, only minimized lifecycle attribution "
"and source identity are exported. Provenance, arbitrary metadata, request "
"hashes, idempotency keys, and source-module payloads are excluded; source "
"references are never traversed. Task attribution is retained as immutable "
"accountability evidence. Assignment or content erasure requires manual "
"source-owner and retention review because a Task can be shared institutional "
"work; the provider performs no automatic mutation."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "tenant_admin", "operator", "auditor"),
related_modules=("core", "workflow_engine", "approvals", "notifications"),
metadata={
"help_contexts": [
"tasks.page.inbox",
"tasks.page.detail",
"tasks.field.assignment",
"privacy.data-subject-requests",
],
"consequence_classes": {
"export_assigned_task": (
"Returns bounded Task-owned data and exact account assignment only."
),
"review_assignment_erasure": (
"Requires the Task source owner and retention authority before "
"changing shared work."
),
"retain_actor_attribution": (
"Preserves minimized Task lifecycle accountability evidence."
),
},
},
),
DocumentationTopic(
id="tasks.quick-access-and-product-area",
title="Work in product navigation and Quick Access",
@@ -232,6 +288,7 @@ manifest = ModuleManifest(
provides_interfaces=(
ModuleInterfaceProvider(name=CAPABILITY_TASK_COMMANDS, version="1.0.0"),
ModuleInterfaceProvider(name="tasks.work_items", version="1.0.0"),
ModuleInterfaceProvider(name=TASKS_DSAR_CAPABILITY, version="0.1.0"),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
@@ -336,13 +393,24 @@ manifest = ModuleManifest(
),
),
tenant_summary_providers=(_tenant_summary,),
capability_factories={CAPABILITY_TASK_COMMANDS: _service},
capability_factories={
CAPABILITY_TASK_COMMANDS: _service,
TASKS_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={
CAPABILITY_TASK_COMMANDS: CapabilityDocumentation(
label="Task commands",
summary="Creates replay-safe explicit tasks without importing the Tasks implementation.",
contract_version="1.0.0",
)
),
TASKS_DSAR_CAPABILITY: CapabilityDocumentation(
label="Tasks data-subject request provider",
summary=(
"Exports account-assigned Tasks and minimized actor attribution "
"with governed non-executable erasure outcomes."
),
contract_version="0.1.0",
),
},
work_item_providers=(
WorkItemProviderRegistration(id="tasks.explicit", factory=_service, order=10),
tests/test_dsar_provider.py
+376
View File
@@ -0,0 +1,376 @@
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_tasks.backend.db.models import TaskAssignment, TaskItem
from govoplan_tasks.backend.dsar_provider import (
TASKS_DSAR_CAPABILITY,
TasksDsarProvider,
)
from govoplan_tasks.backend.manifest import manifest
class _Registry:
def __init__(self, provider: TasksDsarProvider, *, active: bool = True) -> None:
self.provider = provider
self.active = active
def capability_names(self):
return (TASKS_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "tasks"
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": ("tasks",) 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": "tasks"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != TASKS_DSAR_CAPABILITY:
raise KeyError(name)
class TasksDsarProviderTests(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 = TasksDsarProvider()
self.assertIsInstance(self.provider, DsarProvider)
self._seed()
self.session.commit()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def _task(
self,
task_id: str,
*,
tenant_id: str = "tenant-1",
created_by: str = "account-other",
updated_by: str = "account-other",
completed_by: str | None = None,
) -> TaskItem:
return TaskItem(
id=task_id,
tenant_id=tenant_id,
title=f"Private title for {task_id}",
summary=f"Private summary for {task_id}",
status="completed" if completed_by else "open",
priority="high",
required_action="Review the source decision",
action_url="/cases/case-1",
source_module="cases",
source_resource_type="case",
source_resource_id="case-1",
source_revision="4",
sources=[
{
"module_id": "cases",
"resource_type": "case",
"resource_id": "case-1",
"revision": "4",
"url": "/cases/case-1",
"label": "Case reference",
}
],
provenance={"secret": "provenance-secret-do-not-export"},
metadata_={"secret": "metadata-secret-do-not-export"},
revision=2,
idempotency_key=f"idempotency-{task_id}-do-not-export",
request_sha256="a" * 64,
created_by=created_by,
updated_by=updated_by,
completed_by=completed_by,
)
def _seed(self) -> None:
assigned = self._task("task-assigned")
assigned.assignments.extend(
(
TaskAssignment(
id="assignment-account",
tenant_id="tenant-1",
assignment_kind="account",
assignment_id="account-1",
assignment_label="Resident account",
),
TaskAssignment(
id="assignment-group",
tenant_id="tenant-1",
assignment_kind="group",
assignment_id="group-private",
assignment_label="Private group label do not export",
),
)
)
actor_only = self._task(
"task-actor-only",
created_by="account-1",
updated_by="account-1",
completed_by="account-1",
)
actor_only.assignments.append(
TaskAssignment(
id="assignment-other",
tenant_id="tenant-1",
assignment_kind="account",
assignment_id="account-other",
assignment_label="Other account",
)
)
other = self._task("task-other")
other.assignments.append(
TaskAssignment(
id="assignment-other-task",
tenant_id="tenant-1",
assignment_kind="account",
assignment_id="account-other",
)
)
other_tenant = self._task(
"task-other-tenant",
tenant_id="tenant-2",
created_by="account-1",
updated_by="account-1",
)
other_tenant.assignments.append(
TaskAssignment(
id="assignment-other-tenant",
tenant_id="tenant-2",
assignment_kind="account",
assignment_id="account-1",
)
)
self.session.add_all((assigned, actor_only, other, other_tenant))
@staticmethod
def _subject() -> DsarSubjectRef:
return DsarSubjectRef(account_id="account-1")
def test_search_exports_assigned_task_and_minimized_actor_attribution(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=self._subject(),
)
by_id = {record.resource_id: record for record in records}
self.assertEqual(
{"task-assigned", "task-actor-only"},
set(by_id),
)
self.assertEqual("assigned_task", by_id["task-assigned"].resource_type)
self.assertEqual(
"task_actor_attribution",
by_id["task-actor-only"].resource_type,
)
exported = json.dumps([record.to_dict() for record in records])
self.assertIn("Private summary for task-assigned", exported)
self.assertIn('"module_id": "cases"', exported)
self.assertIn('"resource_id": "case-1"', exported)
self.assertIn("completed_task", exported)
self.assertNotIn("Private summary for task-actor-only", exported)
self.assertNotIn("Private group label do not export", exported)
self.assertNotIn("group-private", exported)
self.assertNotIn("provenance-secret-do-not-export", exported)
self.assertNotIn("metadata-secret-do-not-export", exported)
self.assertNotIn("idempotency-task-assigned-do-not-export", exported)
self.assertNotIn("task-other-tenant", exported)
def test_exact_task_reference_narrows_and_conflicts_fail_closed(self) -> None:
narrowed = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"tasks.task": "task-assigned"},
),
)
conflict = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={"tasks.account": "account-other"},
),
)
reference_only = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(external_references={"tasks.task": "task-assigned"}),
)
self.assertEqual(["task-assigned"], [item.resource_id for item in narrowed])
self.assertEqual((), conflict)
self.assertEqual((), reference_only)
def test_erasure_requires_review_or_retention_and_changes_nothing(self) -> None:
subject = self._subject()
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(
{"manual_review", "retain"},
{action.kind for action in actions},
)
self.assertTrue(all(not action.executable for action in actions))
results = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
self.assertTrue(all(result.status == "blocked" for result in results))
self.assertIsNotNone(self.session.get(TaskItem, "task-assigned"))
self.assertIsNotNone(self.session.get(TaskAssignment, "assignment-account"))
def test_foreign_records_and_actions_are_rejected(self) -> None:
subject = self._subject()
with self.assertRaisesRegex(ValueError, "foreign provider record"):
self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(
DsarRecordRef(
provider_id="workflow_engine",
module_id="workflow_engine",
resource_type="assigned_task",
resource_id="task-assigned",
category="work",
title="Foreign task",
),
),
)
with self.assertRaisesRegex(ValueError, "foreign provider action"):
self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(
DsarErasureActionRef(
action_id="workflow_engine:retain:task:task-assigned",
provider_id="workflow_engine",
module_id="workflow_engine",
kind="retain",
resource_type="task_actor_attribution",
resource_id="task-assigned",
title="Retain task",
rationale="Foreign action",
executable=False,
),
),
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-TASKS-1",
request_kind="access",
subject=self._subject(),
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([TASKS_DSAR_CAPABILITY], row.coverage["provider_capabilities"])
self.assertEqual(2, row.search_result["record_count"])
inactive = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-TASKS-2",
request_kind="access",
subject=self._subject(),
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(
[TASKS_DSAR_CAPABILITY],
inactive.coverage["inactive_provider_capabilities"],
)
self.assertIn(TASKS_DSAR_CAPABILITY, manifest.capability_factories)
self.assertIn(TASKS_DSAR_CAPABILITY, manifest.capability_documentation)
self.assertIn(
TASKS_DSAR_CAPABILITY,
{item.name for item in manifest.provides_interfaces},
)
self.assertTrue(
any(
topic.id == "tasks.data-subject-requests"
and {"admin", "user"}.issubset(topic.documentation_types)
for topic in manifest.documentation
)
)
if __name__ == "__main__":
unittest.main()