140 lines
5.1 KiB
Python
140 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from govoplan_access.backend.db.base import AccessBase
|
|
from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, User
|
|
from govoplan_access.backend.dsar_provider import AccessDsarProvider
|
|
from govoplan_core.core.dsar import DsarSubjectRef
|
|
|
|
|
|
class AccessDsarProviderTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine(
|
|
"sqlite+pysqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
AccessBase.metadata.create_all(bind=self.engine)
|
|
self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
|
|
self.account = Account(
|
|
id="account-1",
|
|
email="ada@example.test",
|
|
normalized_email="ada@example.test",
|
|
display_name="Ada Example",
|
|
password_hash="secret-hash",
|
|
)
|
|
self.user = User(
|
|
id="membership-1",
|
|
tenant_id="tenant-1",
|
|
account_id=self.account.id,
|
|
email="ada@example.test",
|
|
display_name="Ada Example",
|
|
password_hash="tenant-secret-hash",
|
|
settings={"locale": "de"},
|
|
mail_profile_policy={"profile": "one"},
|
|
)
|
|
self.key = ApiKey(
|
|
id="key-1",
|
|
tenant_id="tenant-1",
|
|
user_id=self.user.id,
|
|
name="Automation",
|
|
prefix="gpn_example",
|
|
key_hash="do-not-export",
|
|
scopes=["files:read"],
|
|
)
|
|
self.auth_session = AuthSession(
|
|
id="session-1",
|
|
tenant_id="tenant-1",
|
|
user_id=self.user.id,
|
|
account_id=self.account.id,
|
|
token_hash="do-not-export",
|
|
csrf_token_hash="do-not-export",
|
|
expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
|
user_agent="Browser fingerprint",
|
|
ip_address="192.0.2.10",
|
|
)
|
|
self.session.add_all([self.account, self.user, self.key, self.auth_session])
|
|
self.session.commit()
|
|
self.provider = AccessDsarProvider()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
AccessBase.metadata.drop_all(bind=self.engine)
|
|
self.engine.dispose()
|
|
|
|
def test_search_omits_secret_and_client_fingerprint_material(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(email="ADA@example.test"),
|
|
)
|
|
serialized = repr([record.to_dict() for record in records])
|
|
self.assertIn("membership-1", serialized)
|
|
self.assertNotIn("do-not-export", serialized)
|
|
self.assertNotIn("Browser fingerprint", serialized)
|
|
self.assertNotIn("192.0.2.10", serialized)
|
|
|
|
def test_multiple_subject_selectors_must_identify_the_same_membership(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
membership_id=self.user.id,
|
|
email="different@example.test",
|
|
),
|
|
)
|
|
|
|
self.assertEqual((), records)
|
|
|
|
def test_plan_and_execution_anonymize_membership_and_revoke_credentials(self) -> None:
|
|
records = self.provider.search_subject(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id=self.account.id),
|
|
)
|
|
actions = self.provider.plan_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id=self.account.id),
|
|
records=records,
|
|
)
|
|
executable = tuple(action for action in actions if action.executable)
|
|
self.assertEqual(3, len(executable))
|
|
self.assertTrue(any(action.kind == "manual_review" for action in actions))
|
|
|
|
results = self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id=self.account.id),
|
|
actions=executable,
|
|
request_id="dsar-1",
|
|
)
|
|
self.assertEqual({"executed"}, {result.status for result in results})
|
|
self.assertTrue(self.user.email.endswith("@invalid.govoplan"))
|
|
self.assertFalse(self.user.is_active)
|
|
self.assertEqual({}, self.user.settings)
|
|
self.assertIsNotNone(self.key.revoked_at)
|
|
self.assertIsNotNone(self.auth_session.revoked_at)
|
|
self.assertIsNone(self.auth_session.user_agent)
|
|
self.assertIsNone(self.auth_session.ip_address)
|
|
self.assertEqual("ada@example.test", self.account.email)
|
|
|
|
repeated = self.provider.execute_erasure(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id=self.account.id),
|
|
actions=executable,
|
|
request_id="dsar-1",
|
|
)
|
|
self.assertEqual({"unchanged"}, {result.status for result in repeated})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|