tegwick
1127f852dd
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
94 lines
4.9 KiB
Python
94 lines
4.9 KiB
Python
"""Opt-in acceptance against a disposable PostgreSQL database, never production."""
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
|
|
from test_web import FakeProvisioning
|
|
from user_engine.adapters import LocalAuthorizationCheckPort
|
|
from user_engine.adapters.postgres import PostgresUserEngineStore
|
|
from user_engine.domain import AccountStatus, User
|
|
from user_engine.errors import ConflictError
|
|
from user_engine.service import UserEngineService
|
|
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
|
|
from user_engine.testing.postgres_provider import postgres_provider_test_config, connect_postgres_provider, reset_user_engine_postgres_tables
|
|
from user_engine.web import PortalApplication
|
|
|
|
class PostgresJourneyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.config,reason=postgres_provider_test_config()
|
|
if reason:self.skipTest(reason)
|
|
self.connections=[]
|
|
self.seed=self.connect()
|
|
reset_user_engine_postgres_tables(self.seed)
|
|
self.tenant='tenant:trial:concurrency'
|
|
claims=human_actor_claims(subject='operator',tenant='tenant:platform:root')
|
|
claims['roles']=['platform-operator']
|
|
self.actor=FixtureIdentityClaimsAdapter().normalize(claims)
|
|
self.provider=FakeProvisioning()
|
|
|
|
def connect(self):
|
|
connection=connect_postgres_provider(self.config.dsn)
|
|
self.connections.append(connection)
|
|
return connection
|
|
|
|
def service(self,connection):
|
|
return UserEngineService(store=PostgresUserEngineStore(connection),identity_adapter=FixtureIdentityClaimsAdapter(),authorization=LocalAuthorizationCheckPort())
|
|
|
|
def tearDown(self):
|
|
for connection in self.connections:connection.close()
|
|
|
|
def test_two_connections_cannot_disable_both_admins(self):
|
|
service=self.service(self.seed)
|
|
users=[]
|
|
for name in ['first','second']:
|
|
u=service.create_user(self.actor,display_name=name,primary_email=name+'@example.test')
|
|
service.set_tenant_account_status(self.actor,u.user_id,AccountStatus.ACTIVE,tenant=self.tenant)
|
|
service.add_membership(self.actor,u.user_id,tenant=self.tenant,scope_type='tenant',scope_id=self.tenant,kind='tenant-admin')
|
|
service.link_identity(self.actor,u.user_id,issuer='urn:netkingdom:directory',subject=name,provider='netkingdom-lldap')
|
|
users.append(u)
|
|
apps=[PortalApplication(self.service(self.connect()),trusted_proxy_secret='disposable-test-marker-only',login_url='https://test.example',provisioning=self.provider) for _ in users]
|
|
barrier=Barrier(2)
|
|
def disable(pair):
|
|
app,u=pair
|
|
barrier.wait(timeout=5)
|
|
try:
|
|
app._change_status(self.actor,self.tenant,u.user_id,AccountStatus.SUSPENDED,idempotency_key='test-'+u.user_id,correlation_id='concurrent-test')
|
|
return 'changed'
|
|
except ConflictError:return 'conflict'
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results=list(pool.map(disable,zip(apps,users)))
|
|
self.assertEqual(['changed','conflict'],sorted(results))
|
|
self.assertEqual(1,len(self.provider.actions))
|
|
self.assertEqual(1,sum(service.store.tenant_account(self.tenant,u.user_id).status==AccountStatus.ACTIVE for u in users))
|
|
|
|
def test_nested_bootstrap_rolls_back_all_local_records(self):
|
|
service=self.service(self.seed)
|
|
user_id=None
|
|
with self.assertRaises(RuntimeError):
|
|
with service.store.tenant_lifecycle_guard(self.tenant), service.store.transaction():
|
|
user=service.create_user(self.actor,display_name='Partial',primary_email='partial@example.test')
|
|
user_id=user.user_id
|
|
service.set_tenant_account_status(self.actor,user_id,AccountStatus.INVITED,tenant=self.tenant)
|
|
raise RuntimeError('failure before first-admin membership')
|
|
self.assertIsNone(service.store.user(user_id))
|
|
self.assertIsNone(service.store.tenant_account(self.tenant,user_id))
|
|
|
|
def test_guard_releases_after_exception(self):
|
|
first=PostgresUserEngineStore(self.seed)
|
|
second=PostgresUserEngineStore(self.connect())
|
|
with self.assertRaises(RuntimeError):
|
|
with first.tenant_lifecycle_guard(self.tenant):raise RuntimeError('simulated provider failure')
|
|
cursor=second.connection.cursor()
|
|
cursor.execute("SET statement_timeout = '2s'")
|
|
with second.tenant_lifecycle_guard(self.tenant):pass
|
|
cursor.close()
|
|
|
|
def test_successful_guard_does_not_commit_or_discard_caller_work(self):
|
|
first=PostgresUserEngineStore(self.seed)
|
|
second=PostgresUserEngineStore(self.connect())
|
|
u=User(user_id='pending-user',display_name='Pending')
|
|
first.save_user(u)
|
|
with first.tenant_lifecycle_guard(self.tenant):pass
|
|
self.assertIsNone(second.user(u.user_id))
|
|
self.seed.commit()
|
|
self.assertEqual(u,second.user(u.user_id))
|