test_wp10_tenant_postgres.py 41 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549
  1. from __future__ import annotations
  2. import os
  3. import subprocess
  4. import threading
  5. import uuid
  6. from concurrent.futures import ThreadPoolExecutor
  7. from pathlib import Path
  8. import pytest
  9. from sqlalchemy import create_engine, text
  10. from sqlalchemy.engine import make_url
  11. from sqlalchemy.exc import SQLAlchemyError
  12. from app.core.system.tenant_repository import SqlAlchemyTenantFoundationRepository
  13. pytestmark = pytest.mark.integration
  14. ROOT = Path(__file__).resolve().parents[2]
  15. def test_wp10_real_postgres_runtime_is_rls_scoped_and_two_connections_cannot_overreserve():
  16. # Superseded by claim-bound tests below: direct runtime repository access
  17. # must now fail before any payload or tenant GUC can reach PostgreSQL.
  18. with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
  19. SqlAlchemyTenantFoundationRepository(object())
  20. return
  21. database_url = os.getenv("TEST_DATABASE_URL")
  22. if not database_url:
  23. pytest.skip("TEST_DATABASE_URL is required")
  24. tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
  25. engine = create_engine(database_url, pool_pre_ping=True)
  26. try:
  27. with engine.begin() as connection:
  28. repository = SqlAlchemyTenantFoundationRepository(connection)
  29. repository.provision(tenant_id, "private_single_tenant", "provision-1")
  30. first = engine.connect()
  31. second = engine.connect()
  32. first_transaction = first.begin()
  33. second_transaction = second.begin()
  34. try:
  35. first_repo = SqlAlchemyTenantFoundationRepository(first)
  36. second_repo = SqlAlchemyTenantFoundationRepository(second)
  37. fence = first_repo.reserve_quota(tenant_id, "records", 1, "reserve-1")
  38. first_transaction.commit()
  39. with pytest.raises(ValueError, match="tenant_quota_exhausted"):
  40. second_repo.reserve_quota(tenant_id, "records", 1, "reserve-2")
  41. second_transaction.rollback()
  42. with engine.begin() as check:
  43. runtime = SqlAlchemyTenantFoundationRepository(check)
  44. assert runtime.list_audit(tenant_id)[0]["tenant_id"] == tenant_id
  45. with pytest.raises(ValueError, match="tenant_scope_denied"):
  46. runtime.list_audit("wrong-tenant")
  47. assert fence == 1
  48. finally:
  49. if first_transaction.is_active:
  50. first_transaction.rollback()
  51. if second_transaction.is_active:
  52. second_transaction.rollback()
  53. first.close()
  54. second.close()
  55. finally:
  56. with engine.begin() as cleanup:
  57. cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  58. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  59. cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  60. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  61. engine.dispose()
  62. def test_wp10_runtime_role_cannot_directly_mutate_tenant_audit_after_role_init():
  63. with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
  64. SqlAlchemyTenantFoundationRepository(object())
  65. return
  66. database_url = os.getenv("TEST_DATABASE_URL")
  67. runtime_url = os.getenv("TEST_RUNTIME_DATABASE_URL")
  68. if not database_url or not runtime_url:
  69. pytest.skip("TEST_DATABASE_URL and TEST_RUNTIME_DATABASE_URL are required")
  70. tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
  71. admin = create_engine(database_url, pool_pre_ping=True)
  72. runtime = create_engine(runtime_url, pool_pre_ping=True)
  73. try:
  74. with admin.begin() as connection:
  75. SqlAlchemyTenantFoundationRepository(connection).provision(
  76. tenant_id, "private_single_tenant", "provision-1"
  77. )
  78. with runtime.begin() as connection:
  79. connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  80. with pytest.raises(SQLAlchemyError):
  81. connection.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  82. finally:
  83. with admin.begin() as cleanup:
  84. cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  85. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  86. cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  87. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  88. runtime.dispose()
  89. admin.dispose()
  90. def test_wp10_lifecycle_fence_replay_and_frozen_task_rejection_persist_in_postgres():
  91. with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
  92. SqlAlchemyTenantFoundationRepository(object())
  93. return
  94. database_url = os.getenv("TEST_DATABASE_URL")
  95. if not database_url:
  96. pytest.skip("TEST_DATABASE_URL is required")
  97. tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
  98. engine = create_engine(database_url, pool_pre_ping=True)
  99. try:
  100. with engine.begin() as connection:
  101. repository = SqlAlchemyTenantFoundationRepository(connection)
  102. repository.provision(tenant_id, "private_single_tenant", "provision-1")
  103. frozen = repository.transition_lifecycle(tenant_id, "freeze", 0, "freeze-1")
  104. replay = repository.transition_lifecycle(tenant_id, "freeze", 0, "freeze-1")
  105. assert replay == frozen
  106. with pytest.raises(ValueError, match="tenant_fence_conflict"):
  107. repository.transition_lifecycle(tenant_id, "begin_recovery", 0, "recover-1", approval_ref="approval-1")
  108. with pytest.raises(ValueError, match="tenant_not_active"):
  109. repository.reserve_quota(tenant_id, "background_tasks", 1, "task-while-frozen")
  110. recovered = repository.transition_lifecycle(tenant_id, "begin_recovery", frozen, "recover-1", approval_ref="approval-1")
  111. active = repository.transition_lifecycle(tenant_id, "activate", recovered, "activate-1", approval_ref="approval-1")
  112. assert active > recovered > frozen
  113. finally:
  114. with engine.begin() as cleanup:
  115. cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  116. cleanup.execute(text("DELETE FROM public.tenant_lifecycle_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  117. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  118. cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  119. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  120. engine.dispose()
  121. def test_wp10_postgres_quota_accepts_bounded_decimal_without_overreserve():
  122. with pytest.raises(RuntimeError, match="tenant_legacy_gateway_sealed"):
  123. SqlAlchemyTenantFoundationRepository(object())
  124. return
  125. database_url = os.getenv("TEST_DATABASE_URL")
  126. if not database_url:
  127. pytest.skip("TEST_DATABASE_URL is required")
  128. tenant_id = f"wp10-{uuid.uuid4().hex[:16]}"
  129. engine = create_engine(database_url, pool_pre_ping=True)
  130. try:
  131. with engine.begin() as connection:
  132. repository = SqlAlchemyTenantFoundationRepository(connection)
  133. repository.provision(tenant_id, "private_single_tenant", "provision-1")
  134. assert repository.reserve_quota(tenant_id, "records", 0.5, "decimal-1") == 1
  135. with pytest.raises(ValueError, match="tenant_quota_exhausted"):
  136. repository.reserve_quota(tenant_id, "records", 0.75, "decimal-2")
  137. finally:
  138. with engine.begin() as cleanup:
  139. cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  140. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  141. cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  142. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  143. engine.dispose()
  144. def test_wp10_runtime_cannot_claim_another_tenant_by_payload_or_guc():
  145. database_url = os.getenv("TEST_DATABASE_URL")
  146. if not database_url:
  147. pytest.skip("TEST_DATABASE_URL is required")
  148. alpha = f"wp10-alpha-{uuid.uuid4().hex[:12]}"
  149. beta = f"wp10-beta-{uuid.uuid4().hex[:12]}"
  150. runtime_name = f"wp10_runtime_{uuid.uuid4().hex[:12]}"
  151. runtime_password = f"wp10_{uuid.uuid4().hex}"
  152. admin = create_engine(database_url, pool_pre_ping=True)
  153. runtime = None
  154. try:
  155. with admin.begin() as connection:
  156. quoted_runtime = connection.dialect.identifier_preparer.quote(runtime_name)
  157. connection.execute(text(f"CREATE ROLE {quoted_runtime} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": runtime_password})
  158. connection.execute(text(f"GRANT dataops_app_runtime TO {quoted_runtime}"))
  159. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": alpha})
  160. runtime = create_engine(make_url(database_url).set(username=runtime_name, password=runtime_password).render_as_string(hide_password=False), pool_pre_ping=True)
  161. with runtime.begin() as connection:
  162. connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": alpha})
  163. with pytest.raises(SQLAlchemyError, match="permission denied"):
  164. connection.execute(text("SELECT public.tenant_foundation_write('provision', CAST(:payload AS jsonb))"), {"payload": f'{{"tenant_id":"{beta}","delivery_mode":"private_single_tenant","idempotency_key":"beta-provision"}}'})
  165. with admin.connect() as connection:
  166. assert connection.execute(text("SELECT count(*) FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": beta}).scalar_one() == 0
  167. finally:
  168. with admin.begin() as cleanup:
  169. for tenant_id in (alpha, beta):
  170. cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  171. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  172. cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  173. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  174. if runtime is not None:
  175. runtime.dispose()
  176. with admin.begin() as connection:
  177. connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(runtime_name)}"))
  178. admin.dispose()
  179. def test_wp10_control_claim_is_membership_derived_one_shot_and_runtime_cannot_issue():
  180. database_url = os.getenv("TEST_DATABASE_URL")
  181. if not database_url:
  182. pytest.skip("TEST_DATABASE_URL is required")
  183. tenant_id = f"wp10-claim-{uuid.uuid4().hex[:12]}"
  184. principal = str(uuid.uuid4())
  185. claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
  186. control_name = f"wp10_control_{uuid.uuid4().hex[:12]}"
  187. control_password = f"wp10_{uuid.uuid4().hex}"
  188. runtime_name = f"wp10_runtime_{uuid.uuid4().hex[:12]}"
  189. runtime_password = f"wp10_{uuid.uuid4().hex}"
  190. admin = create_engine(database_url, pool_pre_ping=True)
  191. runtime = None
  192. control = None
  193. try:
  194. with admin.begin() as connection:
  195. quoted_control = connection.dialect.identifier_preparer.quote(control_name)
  196. connection.execute(text(
  197. f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
  198. ), {"password": control_password})
  199. connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
  200. quoted_runtime = connection.dialect.identifier_preparer.quote(runtime_name)
  201. connection.execute(text(
  202. f"CREATE ROLE {quoted_runtime} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
  203. ), {"password": runtime_password})
  204. connection.execute(text(f"GRANT dataops_app_runtime TO {quoted_runtime}"))
  205. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
  206. connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
  207. control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
  208. control = create_engine(control_url, pool_pre_ping=True)
  209. runtime = create_engine(make_url(database_url).set(username=runtime_name, password=runtime_password).render_as_string(hide_password=False), pool_pre_ping=True)
  210. with control.begin() as connection:
  211. assert connection.execute(text("SELECT current_user")).scalar_one() == control_name
  212. issued = connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"private.example","action":"quota_reserve","request":{{"quota_name":"records","amount":"1","idempotency_key":"claim-reserve-1"}}}}'}).scalar_one()
  213. assert dict(issued)["tenant_id"] == tenant_id
  214. with runtime.begin() as connection, pytest.raises(SQLAlchemyError):
  215. connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": "{}"})
  216. with control.begin() as connection:
  217. assert dict(connection.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": '{"quota_name":"records","amount":"1","idempotency_key":"claim-reserve-1"}'}).scalar_one())["tenant_id"] == tenant_id
  218. with pytest.raises(SQLAlchemyError):
  219. connection.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": '{"quota_name":"records","amount":"1","idempotency_key":"claim-reserve-1"}'})
  220. finally:
  221. with admin.begin() as cleanup:
  222. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  223. cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  224. cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  225. cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  226. cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  227. # The append-only tenant audit table is RLS-scoped even for the
  228. # owner role. Test cleanup must declare the exact generated
  229. # tenant scope; it must not weaken RLS or use a broad delete.
  230. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  231. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  232. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  233. if control is not None:
  234. control.dispose()
  235. if runtime is not None:
  236. runtime.dispose()
  237. with admin.begin() as connection:
  238. connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(control_name)}"))
  239. connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(runtime_name)}"))
  240. admin.dispose()
  241. def test_wp10_control_claim_two_connections_allow_one_consumer_and_reject_after_restart():
  242. database_url = os.getenv("TEST_DATABASE_URL")
  243. if not database_url:
  244. pytest.skip("TEST_DATABASE_URL is required")
  245. tenant_id = f"wp10-claim-race-{uuid.uuid4().hex[:10]}"
  246. principal = str(uuid.uuid4())
  247. claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
  248. request_json = '{"quota_name":"records","amount":"1","idempotency_key":"claim-race-1"}'
  249. control_name = f"wp10_control_{uuid.uuid4().hex[:12]}"
  250. control_password = f"wp10_{uuid.uuid4().hex}"
  251. admin = create_engine(database_url, pool_pre_ping=True)
  252. control = None
  253. try:
  254. with admin.begin() as connection:
  255. quoted_control = connection.dialect.identifier_preparer.quote(control_name)
  256. connection.execute(text(
  257. f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"
  258. ), {"password": control_password})
  259. connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
  260. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
  261. connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
  262. control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
  263. control = create_engine(control_url, pool_pre_ping=True)
  264. with control.begin() as connection:
  265. connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"private.example","action":"quota_reserve","request":{request_json}}}'})
  266. barrier = threading.Barrier(2)
  267. def consume_once() -> bool:
  268. with control.begin() as connection:
  269. barrier.wait(timeout=10)
  270. try:
  271. connection.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": request_json}).scalar_one()
  272. return True
  273. except SQLAlchemyError:
  274. return False
  275. with ThreadPoolExecutor(max_workers=2) as workers:
  276. outcomes = list(workers.map(lambda _: consume_once(), range(2)))
  277. assert outcomes.count(True) == 1
  278. assert outcomes.count(False) == 1
  279. control.dispose()
  280. control = create_engine(control_url, pool_pre_ping=True)
  281. with control.begin() as restarted, pytest.raises(SQLAlchemyError, match="tenant_claim_replayed_or_expired"):
  282. restarted.execute(text("SELECT public.tenant_control_consume_claim(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "action": "quota_reserve", "request": request_json})
  283. finally:
  284. if control is not None:
  285. control.dispose()
  286. with admin.begin() as cleanup:
  287. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  288. cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  289. cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  290. cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  291. cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  292. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  293. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  294. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  295. cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
  296. admin.dispose()
  297. def test_wp10_control_claim_binds_decimal_quota_reserve_and_release_atomically():
  298. database_url = os.getenv("TEST_DATABASE_URL")
  299. if not database_url:
  300. pytest.skip("TEST_DATABASE_URL is required")
  301. tenant_id = f"wp10-quota-{uuid.uuid4().hex[:14]}"
  302. principal = str(uuid.uuid4())
  303. control_name = f"wp10_control_{uuid.uuid4().hex[:12]}"
  304. control_password = f"wp10_{uuid.uuid4().hex}"
  305. request_json = '{"quota_name":"records","amount":"0.5","idempotency_key":"quota-1"}'
  306. admin = create_engine(database_url, pool_pre_ping=True)
  307. control = None
  308. try:
  309. with admin.begin() as connection:
  310. quoted_control = connection.dialect.identifier_preparer.quote(control_name)
  311. connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
  312. connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
  313. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
  314. connection.execute(text("INSERT INTO public.tenant_quotas(tenant_id,quota_name,limit_units) VALUES(:tenant_id,'records',1)"), {"tenant_id": tenant_id})
  315. connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
  316. control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
  317. control = create_engine(control_url, pool_pre_ping=True)
  318. def claim(action: str, request: str) -> tuple[str, str]:
  319. claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
  320. with control.begin() as connection:
  321. connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"private.example","action":"{action}","request":{request}}}'})
  322. return claim_uid, nonce
  323. reserve_claim, reserve_nonce = claim("quota_reserve", request_json)
  324. with control.begin() as connection:
  325. reserved = connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": reserve_claim, "nonce": reserve_nonce, "action": "quota_reserve", "request": request_json}).scalar_one()
  326. assert dict(reserved)["status"] == "reserved"
  327. with control.begin() as connection, pytest.raises(SQLAlchemyError):
  328. connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": reserve_claim, "nonce": reserve_nonce, "action": "quota_reserve", "request": request_json})
  329. release_request = '{"quota_name":"records","idempotency_key":"quota-1","lease_fence":"1"}'
  330. release_claim, release_nonce = claim("quota_release", release_request)
  331. with control.begin() as connection:
  332. released = connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": release_claim, "nonce": release_nonce, "action": "quota_release", "request": release_request}).scalar_one()
  333. assert dict(released)["status"] == "released"
  334. with admin.connect() as check:
  335. assert check.execute(text("SELECT reserved_units FROM public.tenant_quotas WHERE tenant_id=:tenant_id AND quota_name='records'"), {"tenant_id": tenant_id}).scalar_one() == 0
  336. settle_request = '{"quota_name":"records","amount":"0.25","idempotency_key":"settle-1"}'
  337. settle_reserve_claim, settle_reserve_nonce = claim("quota_reserve", settle_request)
  338. with control.begin() as connection:
  339. assert dict(connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": settle_reserve_claim, "nonce": settle_reserve_nonce, "action": "quota_reserve", "request": settle_request}).scalar_one())["status"] == "reserved"
  340. settle_request = '{"quota_name":"records","idempotency_key":"settle-1","lease_fence":"3"}'
  341. settle_claim, settle_nonce = claim("quota_settle", settle_request)
  342. with control.begin() as connection:
  343. assert dict(connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": settle_claim, "nonce": settle_nonce, "action": "quota_settle", "request": settle_request}).scalar_one())["status"] == "settled"
  344. late_claim, late_nonce = claim("quota_release", settle_request)
  345. with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_quota_terminal"):
  346. connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": late_claim, "nonce": late_nonce, "action": "quota_release", "request": settle_request})
  347. invalid_request = '{"quota_name":"records","amount":1.0,"idempotency_key":"float-1"}'
  348. invalid_claim, invalid_nonce = claim("quota_reserve", invalid_request)
  349. with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_payload_closed"):
  350. connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": invalid_claim, "nonce": invalid_nonce, "action": "quota_reserve", "request": invalid_request})
  351. overflow_request = '{"quota_name":"records","amount":"1000000000000000000000001","idempotency_key":"overflow-1"}'
  352. overflow_claim, overflow_nonce = claim("quota_reserve", overflow_request)
  353. with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_payload_invalid"):
  354. connection.execute(text("SELECT public.tenant_control_quota_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),:action,CAST(:request AS jsonb))"), {"claim_uid": overflow_claim, "nonce": overflow_nonce, "action": "quota_reserve", "request": overflow_request})
  355. finally:
  356. if control is not None:
  357. control.dispose()
  358. with admin.begin() as cleanup:
  359. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  360. cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  361. cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  362. cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  363. cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  364. cleanup.execute(text("DELETE FROM public.tenant_quota_reservations WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  365. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  366. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  367. cleanup.execute(text("DELETE FROM public.tenant_quotas WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  368. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  369. cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
  370. admin.dispose()
  371. def test_wp10_unapproved_shared_delivery_mode_cannot_issue_control_claim():
  372. database_url = os.getenv("TEST_DATABASE_URL")
  373. if not database_url:
  374. pytest.skip("TEST_DATABASE_URL is required")
  375. tenant_id, principal = f"wp10-shared-{uuid.uuid4().hex[:12]}", str(uuid.uuid4())
  376. control_name, control_password = f"wp10_control_{uuid.uuid4().hex[:12]}", f"wp10_{uuid.uuid4().hex}"
  377. claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
  378. admin = create_engine(database_url, pool_pre_ping=True)
  379. control = None
  380. try:
  381. with admin.begin() as connection:
  382. quoted_control = connection.dialect.identifier_preparer.quote(control_name)
  383. connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
  384. connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
  385. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'shared_control_plane','active')"), {"tenant_id": tenant_id})
  386. connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'shared.example')"), {"tenant_id": tenant_id, "principal": principal})
  387. control = create_engine(make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False), pool_pre_ping=True)
  388. with control.begin() as connection, pytest.raises(SQLAlchemyError, match="tenant_membership_denied"):
  389. connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": f'{{"claim_uid":"{claim_uid}","nonce":"{nonce}","principal_id":"{principal}","host":"shared.example","action":"quota_reserve","request":{{"quota_name":"records","amount":"1","idempotency_key":"shared-1"}}}}'})
  390. finally:
  391. if control is not None:
  392. control.dispose()
  393. with admin.begin() as cleanup:
  394. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  395. cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  396. cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  397. cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  398. cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  399. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  400. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  401. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  402. cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
  403. admin.dispose()
  404. def test_wp10_claim_bound_lifecycle_fence_hold_backup_rollback_and_delete():
  405. database_url = os.getenv("TEST_DATABASE_URL")
  406. if not database_url:
  407. pytest.skip("TEST_DATABASE_URL is required")
  408. control_name, control_password = f"wp10_legacy_control_{uuid.uuid4().hex[:12]}", f"wp10_{uuid.uuid4().hex}"
  409. admin = create_engine(database_url, pool_pre_ping=True)
  410. try:
  411. with admin.begin() as connection:
  412. quoted_control = connection.dialect.identifier_preparer.quote(control_name)
  413. connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
  414. connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
  415. control_url = make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False)
  416. with create_engine(control_url).connect() as connection, pytest.raises(SQLAlchemyError):
  417. connection.execute(text("SELECT public.tenant_control_lifecycle_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),'{}'::jsonb)"), {"claim_uid": str(uuid.uuid4()), "nonce": str(uuid.uuid4())})
  418. finally:
  419. with admin.begin() as connection:
  420. connection.execute(text(f"DROP ROLE IF EXISTS {connection.dialect.identifier_preparer.quote(control_name)}"))
  421. admin.dispose()
  422. return
  423. tenant_id, principal = f"wp10-life-{uuid.uuid4().hex[:12]}", str(uuid.uuid4())
  424. control_name, control_password = f"wp10_control_{uuid.uuid4().hex[:12]}", f"wp10_{uuid.uuid4().hex}"
  425. admin = create_engine(database_url, pool_pre_ping=True)
  426. control = None
  427. try:
  428. with admin.begin() as connection:
  429. quoted_control = connection.dialect.identifier_preparer.quote(control_name)
  430. connection.execute(text(f"CREATE ROLE {quoted_control} LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION PASSWORD :password"), {"password": control_password})
  431. connection.execute(text(f"GRANT dataops_tenant_control TO {quoted_control}"))
  432. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
  433. connection.execute(text("INSERT INTO public.tenant_control_memberships(tenant_id,principal_id,host) VALUES(:tenant_id,:principal,'private.example')"), {"tenant_id": tenant_id, "principal": principal})
  434. control = create_engine(make_url(database_url).set(username=control_name, password=control_password).render_as_string(hide_password=False), pool_pre_ping=True)
  435. def transition(operation: str, fence: int, key: str, *, approval: str | None = None, backup: str | None = None, retention: str | None = None):
  436. request = {"operation": operation, "expected_fence": str(fence), "idempotency_key": key, "approval_ref": approval, "backup_digest": backup, "retention_seconds": retention}
  437. claim_uid, nonce = str(uuid.uuid4()), str(uuid.uuid4())
  438. with control.begin() as connection:
  439. connection.execute(text("SELECT public.tenant_control_issue_claim(CAST(:payload AS jsonb))"), {"payload": __import__('json').dumps({"claim_uid": claim_uid, "nonce": nonce, "principal_id": principal, "host": "private.example", "action": "lifecycle_transition", "request": request})})
  440. return dict(connection.execute(text("SELECT public.tenant_control_lifecycle_mutate(CAST(:claim_uid AS uuid),CAST(:nonce AS uuid),CAST(:request AS jsonb))"), {"claim_uid": claim_uid, "nonce": nonce, "request": __import__('json').dumps(request)}).scalar_one())
  441. assert transition("freeze", 0, "freeze-1")["state"] == "frozen"
  442. assert transition("begin_recovery", 1, "recover-1", approval="approval-1")["state"] == "recovering"
  443. assert transition("activate", 2, "activate-1", approval="approval-2")["state"] == "active"
  444. with admin.begin() as connection:
  445. connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  446. connection.execute(text("INSERT INTO public.tenant_legal_holds(hold_ref,tenant_id,state,evidence_digest) VALUES('hold-1',:tenant_id,'active',:digest)"), {"tenant_id": tenant_id, "digest": "a" * 64})
  447. with pytest.raises(SQLAlchemyError, match="tenant_hold_active"):
  448. transition("mark_deletion_candidate", 3, "delete-candidate-1", approval="approval-3", backup="b" * 64, retention="60")
  449. with admin.begin() as connection:
  450. connection.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  451. connection.execute(text("UPDATE public.tenant_legal_holds SET state='released',released_at=clock_timestamp() WHERE hold_ref='hold-1'"))
  452. candidate = transition("mark_deletion_candidate", 3, "delete-candidate-1", approval="approval-3", backup="b" * 64, retention="60")
  453. assert candidate["state"] == "deletion_candidate"
  454. assert transition("rollback", 4, "rollback-1")["state"] == "frozen"
  455. candidate_again = transition("mark_deletion_candidate", 5, "delete-candidate-2", approval="approval-4", backup="c" * 64, retention="120")
  456. assert candidate_again["state"] == "deletion_candidate"
  457. assert transition("delete", 6, "delete-1", approval="approval-5", backup="d" * 64, retention="120")["state"] == "deleted"
  458. # Deleted tenants cannot even receive a lifecycle claim, so rollback
  459. # is intentionally impossible after the destructive transition.
  460. with pytest.raises(SQLAlchemyError, match="tenant_membership_denied"):
  461. transition("rollback", 7, "rollback-after-delete")
  462. finally:
  463. if control is not None:
  464. control.dispose()
  465. with admin.begin() as cleanup:
  466. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  467. cleanup.execute(text("DELETE FROM public.tenant_control_outbox_scope_lookup WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  468. cleanup.execute(text("DELETE FROM public.tenant_control_outbox WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  469. cleanup.execute(text("DELETE FROM public.tenant_control_claims WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  470. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  471. cleanup.execute(text("DELETE FROM public.tenant_legal_holds WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  472. cleanup.execute(text("DELETE FROM public.tenant_lifecycle_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  473. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  474. cleanup.execute(text("DELETE FROM public.tenant_control_memberships WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  475. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  476. cleanup.execute(text(f"DROP ROLE IF EXISTS {cleanup.dialect.identifier_preparer.quote(control_name)}"))
  477. admin.dispose()
  478. def test_wp10_live_tenant_data_blocks_downgrade_before_503_gateway_can_return():
  479. database_url = os.getenv("TEST_DATABASE_URL")
  480. if not database_url:
  481. pytest.skip("TEST_DATABASE_URL is required")
  482. tenant_id = f"wp10-downgrade-{uuid.uuid4().hex[:12]}"
  483. engine = create_engine(database_url, pool_pre_ping=True)
  484. try:
  485. with engine.begin() as connection:
  486. connection.execute(text("INSERT INTO public.tenants(tenant_id,delivery_mode,state) VALUES(:tenant_id,'private_single_tenant','active')"), {"tenant_id": tenant_id})
  487. result = subprocess.run(
  488. [str(ROOT / ".venv/bin/alembic"), "-c", str(ROOT / "alembic.ini"), "downgrade", "20260817_520"],
  489. cwd=ROOT,
  490. env={**os.environ, "MIGRATION_DATABASE_URL": database_url},
  491. capture_output=True,
  492. text=True,
  493. )
  494. assert result.returncode != 0
  495. assert "downgrade refused" in result.stderr and "tenant" in result.stderr
  496. with engine.connect() as connection:
  497. assert connection.execute(text("SELECT version_num FROM public.alembic_version")).scalar_one() == "20260817_525"
  498. finally:
  499. with engine.begin() as cleanup:
  500. cleanup.execute(text("SELECT set_config('dataops.tenant_id', :tenant_id, true)"), {"tenant_id": tenant_id})
  501. cleanup.execute(text("DELETE FROM public.tenant_audit_events WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  502. cleanup.execute(text("DELETE FROM public.tenants WHERE tenant_id=:tenant_id"), {"tenant_id": tenant_id})
  503. engine.dispose()
  504. def test_wp10_low_privilege_migrator_reaches_the_tenant_foundation_head(wp06_head):
  505. engine = create_engine(wp06_head.migration_url, pool_pre_ping=True)
  506. try:
  507. with engine.connect() as connection:
  508. head = connection.execute(text("SELECT version_num FROM public.alembic_version")).scalar_one()
  509. privileges = connection.execute(text("SELECT rolsuper,rolcreaterole FROM pg_roles WHERE rolname=current_user")).one()
  510. assert head == "20260817_525"
  511. assert tuple(privileges) == (False, False)
  512. finally:
  513. engine.dispose()