Support tenant-scoped directory lifecycle with isolation tests
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
This commit is contained in:
parent
0071dba99e
commit
a0cc947bf6
4 changed files with 146 additions and 0 deletions
47
identity-provisioner/tests/test_tenant_access.py
Normal file
47
identity-provisioner/tests/test_tenant_access.py
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
import unittest
|
||||
from provisioner import LLDAPProvisioner, dispatch
|
||||
|
||||
class Directory(LLDAPProvisioner):
|
||||
def __init__(self):
|
||||
self.groups = {'tenant:trial:one:users':1, 'tenant:trial:one:admins':2,
|
||||
'tenant:trial:two:users':3, 'netkingdom-suspended':4, 'unrelated':5}
|
||||
self.calls=[]
|
||||
self.fail=False
|
||||
def _login(self): return 'test-token'
|
||||
def _user(self, token, subject):
|
||||
return {'groups':[{'displayName':n,'id':i} for n,i in self.groups.items()]}
|
||||
def _directory(self, token): return [],[]
|
||||
def _ensure_group(self, token, groups, name): return max(self.groups.values(), default=0)+1
|
||||
def _add_group(self, token, subject, group):
|
||||
self.calls.append(('add',group))
|
||||
if not self.fail:self.groups['tenant:trial:one:users']=group
|
||||
def _remove_group(self, token, subject, group):
|
||||
self.calls.append(('remove',group))
|
||||
if not self.fail:self.groups={n:i for n,i in self.groups.items() if i!=group}
|
||||
|
||||
class TenantAccessTests(unittest.TestCase):
|
||||
def payload(self, **kwargs):
|
||||
return dict(external_subject='uid=member,ou=people,dc=netkingdom,dc=local',
|
||||
tenant='tenant:trial:one', enabled=False, roles=['user'],
|
||||
idempotency_key='test-operation-123456', correlation_id='test', **kwargs)
|
||||
def test_disable_preserves_other_tenants_and_global_suspension(self):
|
||||
d=Directory(); result=dispatch(d,'/v1/identities/tenant-access',self.payload())
|
||||
self.assertEqual('tenant_disabled',result.status)
|
||||
self.assertEqual({'tenant:trial:two:users','netkingdom-suspended','unrelated'},set(d.groups))
|
||||
d.calls.clear(); dispatch(d,'/v1/identities/tenant-access',self.payload())
|
||||
self.assertEqual([],d.calls)
|
||||
def test_reactivation_does_not_remove_platform_suspension(self):
|
||||
d=Directory();d.tenant_access(self.payload());p=self.payload();p['enabled']=True
|
||||
self.assertEqual('tenant_active',d.tenant_access(p).status)
|
||||
self.assertIn('tenant:trial:one:users',d.groups)
|
||||
self.assertIn('netkingdom-suspended',d.groups)
|
||||
self.assertNotIn('tenant:trial:one:admins',d.groups)
|
||||
def test_failed_readback_never_reports_success_and_retry_recovers(self):
|
||||
d=Directory();d.fail=True
|
||||
with self.assertRaises(RuntimeError):d.tenant_access(self.payload())
|
||||
d.fail=False;self.assertEqual('tenant_disabled',d.tenant_access(self.payload()).status)
|
||||
def test_invalid_scope_or_role_causes_no_mutation(self):
|
||||
for key,value in [('tenant','tenant:platform:root'),('tenant','tenant:one:users*'),('enabled','false'),('roles',['platform-operator'])]:
|
||||
d=Directory();p=self.payload();p[key]=value
|
||||
with self.assertRaises(ValueError):d.tenant_access(p)
|
||||
self.assertEqual([],d.calls)
|
||||
Loading…
Add table
Add a link
Reference in a new issue