tegwick
686da7c6bb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929
69 lines
3.5 KiB
Python
69 lines
3.5 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_does_not_emit_root_grants_or_profile_details(self):
|
|
result = self.lookup()
|
|
self.assertFalse(hasattr(result,'root_entitled'))
|
|
self.assertFalse(hasattr(result,'roles'))
|
|
self.assertFalse(hasattr(result,'primary_email'))
|