235 lines
8.1 KiB
Python
235 lines
8.1 KiB
Python
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, 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_identity.backend.db.models import Identity, IdentityAccountLink
|
|
from govoplan_identity.backend.dsar_provider import (
|
|
IDENTITY_DSAR_CAPABILITY,
|
|
IdentityDsarProvider,
|
|
)
|
|
from govoplan_identity.backend.manifest import manifest
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, provider: IdentityDsarProvider) -> None:
|
|
self.provider = provider
|
|
|
|
def capability_names(self):
|
|
return (IDENTITY_DSAR_CAPABILITY,)
|
|
|
|
def capability_owner(self, name):
|
|
if name != IDENTITY_DSAR_CAPABILITY:
|
|
raise KeyError(name)
|
|
return "identity"
|
|
|
|
def tenant_entitlement_resolver(self):
|
|
class _Resolver:
|
|
@staticmethod
|
|
def resolve(session, tenant_id):
|
|
del session, tenant_id
|
|
return type("State", (), {"effective_modules": ("identity",)})()
|
|
|
|
return _Resolver()
|
|
|
|
def require_tenant_capability(self, name, session, **kwargs):
|
|
del session, kwargs
|
|
if name != IDENTITY_DSAR_CAPABILITY:
|
|
raise KeyError(name)
|
|
return self.provider
|
|
|
|
def manifests(self):
|
|
return (type("Manifest", (), {"id": "identity"})(),)
|
|
|
|
|
|
class IdentityDsarProviderTests(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 = IdentityDsarProvider()
|
|
self.assertIsInstance(self.provider, DsarProvider)
|
|
identity = Identity(
|
|
id="identity-1",
|
|
display_name="Ada Example",
|
|
external_subject="external-ada",
|
|
source="directory",
|
|
is_active=True,
|
|
settings={"secret": "identity-setting-do-not-export"},
|
|
)
|
|
other = Identity(
|
|
id="identity-other",
|
|
display_name="Other Person",
|
|
external_subject="external-other",
|
|
source="local",
|
|
is_active=True,
|
|
settings={"private": "other-setting"},
|
|
)
|
|
self.session.add_all((identity, other))
|
|
self.session.flush()
|
|
self.session.add_all(
|
|
(
|
|
IdentityAccountLink(
|
|
id="link-1",
|
|
identity_id="identity-1",
|
|
account_id="account-1",
|
|
is_primary=True,
|
|
source="directory",
|
|
),
|
|
IdentityAccountLink(
|
|
id="link-secondary",
|
|
identity_id="identity-1",
|
|
account_id="account-secondary",
|
|
is_primary=False,
|
|
source="local",
|
|
),
|
|
IdentityAccountLink(
|
|
id="link-other",
|
|
identity_id="identity-other",
|
|
account_id="account-other",
|
|
is_primary=True,
|
|
source="local",
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
@staticmethod
|
|
def _subject() -> DsarSubjectRef:
|
|
return DsarSubjectRef(identity_id="identity-1", account_id="account-1")
|
|
|
|
def test_search_requires_and_exports_one_corroborated_link(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self._subject(),
|
|
)
|
|
|
|
self.assertEqual(1, len(records))
|
|
exported = json.dumps(records[0].to_dict())
|
|
self.assertIn("Ada Example", exported)
|
|
self.assertIn("external-ada", exported)
|
|
self.assertIn("link-1", exported)
|
|
self.assertNotIn("account-secondary", exported)
|
|
self.assertNotIn("Other Person", exported)
|
|
self.assertNotIn("identity-setting-do-not-export", exported)
|
|
|
|
def test_incomplete_conflicting_and_mismatched_selectors_fail_closed(self) -> None:
|
|
subjects = (
|
|
DsarSubjectRef(identity_id="identity-1"),
|
|
DsarSubjectRef(account_id="account-1"),
|
|
DsarSubjectRef(identity_id="identity-1", account_id="account-other"),
|
|
DsarSubjectRef(
|
|
identity_id="identity-1",
|
|
account_id="account-1",
|
|
external_references={"identity.account": "account-other"},
|
|
),
|
|
)
|
|
for subject in subjects:
|
|
with self.subTest(subject=subject):
|
|
self.assertEqual(
|
|
(),
|
|
self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=subject,
|
|
),
|
|
)
|
|
|
|
def test_exact_link_can_corroborate_identity(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
identity_id="identity-1",
|
|
external_references={"identity.link": "link-1"},
|
|
),
|
|
)
|
|
self.assertEqual(["identity-1"], [record.resource_id for record in records])
|
|
|
|
def test_erasure_requires_cross_tenant_manual_review(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session, tenant_id="tenant-1", subject=self._subject()
|
|
)
|
|
actions = self.provider.plan_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self._subject(),
|
|
records=records,
|
|
)
|
|
self.assertEqual(["manual_review"], [action.kind for action in actions])
|
|
results = self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self._subject(),
|
|
actions=actions,
|
|
request_id="dsar-1",
|
|
)
|
|
self.assertEqual(["blocked"], [result.status for result in results])
|
|
self.assertEqual(3, self.session.query(IdentityAccountLink).count())
|
|
|
|
def test_foreign_actions_are_rejected(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=self._subject(),
|
|
actions=(
|
|
DsarErasureActionRef(
|
|
action_id="other:manual_review:x",
|
|
provider_id="other",
|
|
module_id="other",
|
|
kind="manual_review",
|
|
resource_type="canonical_identity",
|
|
resource_id="identity-1",
|
|
title="Foreign",
|
|
rationale="Foreign",
|
|
executable=False,
|
|
),
|
|
),
|
|
request_id="dsar-1",
|
|
)
|
|
|
|
def test_manifest_and_core_workflow_discover_provider(self) -> None:
|
|
self.assertIn(IDENTITY_DSAR_CAPABILITY, manifest.capability_factories)
|
|
self.assertIn(
|
|
IDENTITY_DSAR_CAPABILITY,
|
|
{item.name for item in manifest.provides_interfaces},
|
|
)
|
|
row = create_data_subject_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
reference="DSAR-IDENTITY-1",
|
|
request_kind="access",
|
|
subject=self._subject(),
|
|
purpose="Identity access request",
|
|
legal_basis=None,
|
|
due_at=None,
|
|
requested_by_account_id="operator-1",
|
|
)
|
|
search_data_subject_request(
|
|
self.session,
|
|
registry=_Registry(self.provider),
|
|
row=row,
|
|
expected_revision=row.resource_revision,
|
|
)
|
|
self.assertEqual("searched", row.status)
|
|
self.assertEqual(1, row.search_result["record_count"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|