tegwick
0d0751d39d
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
86 lines
4.4 KiB
Python
86 lines
4.4 KiB
Python
import unittest
|
|
from dataclasses import replace
|
|
|
|
from user_engine.adapters.local import InMemoryUserEngineStore
|
|
from user_engine.domain import AccountStatus, AuthorizationEffect
|
|
from user_engine.errors import AuthorizationDenied, NotFoundError
|
|
from user_engine.service import UserEngineService
|
|
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, StaticAuthorizationCheckPort, human_actor_claims
|
|
|
|
|
|
class AccountAuthorityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.store = InMemoryUserEngineStore()
|
|
self.policy = StaticAuthorizationCheckPort()
|
|
self.service = UserEngineService(store=self.store,identity_adapter=FixtureIdentityClaimsAdapter(),authorization=self.policy)
|
|
self.session = self.service.me(human_actor_claims())
|
|
self.caller = replace(self.session.actor,subject='fixture-workload')
|
|
|
|
def lookup(self,**overrides):
|
|
return self.service.lookup_account_authority(self.caller,**{
|
|
'issuer':self.session.actor.issuer,'subject':self.session.actor.subject,
|
|
'tenant':self.session.actor.tenant,**overrides})
|
|
|
|
def test_existing_lookup_does_not_mutate_and_keeps_caller_distinct(self):
|
|
before = repr(self.store.__dict__)
|
|
result = self.lookup()
|
|
self.assertEqual(repr(self.store.__dict__),before)
|
|
self.assertEqual(result.identity_id,self.session.identities[0].identity_id)
|
|
self.assertEqual(result.account_status,AccountStatus.ACTIVE)
|
|
self.assertEqual(result.tenant_account_status,AccountStatus.ACTIVE)
|
|
request = self.policy.requests[-1]
|
|
self.assertEqual(request.action,'account.authority.read')
|
|
self.assertEqual(request.actor.subject,'fixture-workload')
|
|
self.assertEqual(request.context['lookup_subject'],self.session.actor.subject)
|
|
self.assertIsNotNone(result.observed_at.tzinfo)
|
|
|
|
def test_unknown_identity_does_not_provision(self):
|
|
before = repr(self.store.__dict__)
|
|
with self.assertRaises(NotFoundError): self.lookup(subject='unknown')
|
|
with self.assertRaises(NotFoundError): self.lookup(issuer='https://other.example')
|
|
self.assertEqual(repr(self.store.__dict__),before)
|
|
|
|
def test_missing_scoped_account_is_explicit(self):
|
|
self.assertIsNone(self.lookup(tenant='unknown-tenant').tenant_account_status)
|
|
|
|
def test_current_global_and_scoped_status_are_returned_each_time(self):
|
|
self.lookup()
|
|
self.store.save_account(replace(self.session.account,status=AccountStatus.DISABLED))
|
|
scoped = self.store.tenant_account(self.session.actor.tenant,self.session.user.user_id)
|
|
self.store.save_tenant_account(replace(scoped,status=AccountStatus.DISABLED))
|
|
result = self.lookup()
|
|
self.assertEqual(result.account_status,AccountStatus.DISABLED)
|
|
self.assertEqual(result.tenant_account_status,AccountStatus.DISABLED)
|
|
|
|
def test_denial_precedes_target_identity_lookup(self):
|
|
self.policy.effect = AuthorizationEffect.DENY
|
|
original = self.store.find_identity
|
|
def checked(issuer,subject):
|
|
if subject == self.session.actor.subject:
|
|
raise AssertionError('target lookup before authorization')
|
|
return original(issuer,subject)
|
|
self.store.find_identity = checked
|
|
with self.assertRaises(AuthorizationDenied): self.lookup()
|
|
|
|
def test_custody_mapping_only_after_independent_authorization(self):
|
|
calls = []
|
|
class Custody:
|
|
def observe(inner, **values):
|
|
calls.append(values)
|
|
return {"fixture": "observed-fact"} if values["tenant"] == "platform" else None
|
|
self.service.custody = Custody()
|
|
result = self.lookup(tenant="platform")
|
|
self.assertEqual(result.root_grant, {"fixture": "observed-fact"})
|
|
self.assertEqual(calls[-1], {"issuer": self.session.actor.issuer, "subject": self.session.actor.subject, "tenant": "platform"})
|
|
self.policy.effect = AuthorizationEffect.DENY
|
|
with self.assertRaises(AuthorizationDenied):
|
|
self.lookup(tenant="platform")
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
|
|
def test_does_not_emit_root_grants_or_profile_details(self):
|
|
result = self.lookup()
|
|
self.assertIsNone(result.root_grant)
|
|
self.assertFalse(hasattr(result,'root_entitled'))
|
|
self.assertFalse(hasattr(result,'roles'))
|
|
self.assertFalse(hasattr(result,'primary_email'))
|