test_http_commands.py 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. from __future__ import annotations
  2. import asyncio
  3. from datetime import UTC, datetime
  4. from typing import Any
  5. import pytest
  6. from sqlalchemy import insert, select
  7. from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
  8. from script_build_host.domain.canonical_json import canonical_sha256
  9. from script_build_host.domain.errors import HttpIdempotencyConflict, ProtocolViolation
  10. from script_build_host.domain.records import Principal
  11. from script_build_host.infrastructure.http_commands import HttpCommandJournal
  12. from script_build_host.infrastructure.tables import http_command_table
  13. @pytest.mark.asyncio
  14. async def test_http_command_replays_completed_response_and_rejects_fingerprint_conflict(
  15. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  16. ) -> None:
  17. _, sessions = database
  18. journal = HttpCommandJournal(sessions)
  19. principal = Principal("subject", "tenant")
  20. calls = 0
  21. async def command() -> tuple[int, dict[str, Any]]:
  22. nonlocal calls
  23. calls += 1
  24. return 201, {"success": True, "resource_id": 9}
  25. first = await journal.execute(
  26. principal=principal,
  27. route_family="script-build:test",
  28. idempotency_key="same",
  29. request_payload={"value": 1},
  30. command=command,
  31. resource_id=9,
  32. )
  33. second = await journal.execute(
  34. principal=principal,
  35. route_family="script-build:test",
  36. idempotency_key="same",
  37. request_payload={"value": 1},
  38. command=command,
  39. resource_id=9,
  40. )
  41. assert first == second == (201, {"success": True, "resource_id": 9})
  42. assert calls == 1
  43. with pytest.raises(HttpIdempotencyConflict):
  44. await journal.execute(
  45. principal=principal,
  46. route_family="script-build:test",
  47. idempotency_key="same",
  48. request_payload={"value": 2},
  49. command=command,
  50. )
  51. @pytest.mark.asyncio
  52. async def test_concurrent_same_http_command_executes_once_and_waits_for_completion(
  53. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  54. ) -> None:
  55. _, sessions = database
  56. journal = HttpCommandJournal(sessions)
  57. entered = asyncio.Event()
  58. release = asyncio.Event()
  59. calls = 0
  60. async def command() -> tuple[int, dict[str, Any]]:
  61. nonlocal calls
  62. calls += 1
  63. entered.set()
  64. await release.wait()
  65. return 200, {"success": True}
  66. first = asyncio.create_task(
  67. journal.execute(
  68. principal=Principal("subject"),
  69. route_family="script-build:concurrent",
  70. idempotency_key="one",
  71. request_payload={"build": 1},
  72. command=command,
  73. )
  74. )
  75. await entered.wait()
  76. second = asyncio.create_task(
  77. journal.execute(
  78. principal=Principal("subject"),
  79. route_family="script-build:concurrent",
  80. idempotency_key="one",
  81. request_payload={"build": 1},
  82. command=command,
  83. )
  84. )
  85. release.set()
  86. assert await first == await second == (200, {"success": True})
  87. assert calls == 1
  88. @pytest.mark.asyncio
  89. async def test_failed_http_command_is_not_blindly_replayed(
  90. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  91. ) -> None:
  92. _, sessions = database
  93. journal = HttpCommandJournal(sessions)
  94. async def command() -> tuple[int, dict[str, Any]]:
  95. raise RuntimeError("lost acknowledgement")
  96. values = dict(
  97. principal=Principal("subject"),
  98. route_family="script-build:failed",
  99. idempotency_key="one",
  100. request_payload={"build": 1},
  101. command=command,
  102. )
  103. with pytest.raises(RuntimeError, match="lost acknowledgement"):
  104. await journal.execute(**values)
  105. with pytest.raises(ProtocolViolation, match="reconciliation"):
  106. await journal.execute(**values)
  107. @pytest.mark.asyncio
  108. async def test_reserved_http_command_resumes_preallocated_identity(
  109. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  110. ) -> None:
  111. _, sessions = database
  112. principal = Principal("subject", "tenant")
  113. route = "script-build:start"
  114. key = "crash-window"
  115. payload = {"topic_id": 3}
  116. scope = canonical_sha256(
  117. {"subject": principal.subject, "tenant_id": principal.tenant_id}
  118. ).hex_value
  119. fingerprint = canonical_sha256({"route_family": route, "request": payload}).hex_value
  120. now = datetime.now(UTC)
  121. async with sessions() as session, session.begin():
  122. await session.execute(
  123. insert(http_command_table).values(
  124. principal_scope_sha256=scope,
  125. route_family=route,
  126. idempotency_key=key,
  127. request_fingerprint=fingerprint,
  128. preallocated_root_trace_id="root-preallocated",
  129. resource_id=41,
  130. state="reserved",
  131. created_at=now,
  132. updated_at=now,
  133. )
  134. )
  135. async def command() -> tuple[int, dict[str, Any]]:
  136. raise AssertionError("reserved command must not be blindly replayed")
  137. async def resume(root: str | None, resource: int | None) -> tuple[int, dict[str, Any]]:
  138. assert (root, resource) == ("root-preallocated", 41)
  139. return 200, {"script_build_id": 41, "root_trace_id": root}
  140. result = await HttpCommandJournal(sessions).execute(
  141. principal=principal,
  142. route_family=route,
  143. idempotency_key=key,
  144. request_payload=payload,
  145. command=command,
  146. resume_reserved=resume,
  147. )
  148. assert result == (
  149. 200,
  150. {"script_build_id": 41, "root_trace_id": "root-preallocated"},
  151. )
  152. async with sessions() as session:
  153. assert (
  154. await session.scalar(
  155. select(http_command_table.c.state).where(
  156. http_command_table.c.idempotency_key == key
  157. )
  158. )
  159. == "completed"
  160. )
  161. @pytest.mark.asyncio
  162. async def test_oversized_idempotent_response_is_failed_not_left_reserved(
  163. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  164. ) -> None:
  165. _, sessions = database
  166. async def command() -> tuple[int, dict[str, Any]]:
  167. return 200, {"value": "x" * (65 * 1024)}
  168. with pytest.raises(ProtocolViolation, match="exceeds"):
  169. await HttpCommandJournal(sessions).execute(
  170. principal=Principal("subject"),
  171. route_family="script-build:oversized",
  172. idempotency_key="oversized",
  173. request_payload={},
  174. command=command,
  175. )
  176. async with sessions() as session:
  177. assert (
  178. await session.scalar(
  179. select(http_command_table.c.state).where(
  180. http_command_table.c.idempotency_key == "oversized"
  181. )
  182. )
  183. == "failed"
  184. )
  185. @pytest.mark.asyncio
  186. async def test_idempotency_key_is_scoped_by_principal_and_route(
  187. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  188. ) -> None:
  189. _, sessions = database
  190. journal = HttpCommandJournal(sessions)
  191. calls: list[str] = []
  192. async def execute(label: str) -> tuple[int, dict[str, Any]]:
  193. calls.append(label)
  194. return 200, {"label": label}
  195. common = {
  196. "idempotency_key": "same-key",
  197. "request_payload": {"script_build_id": 7},
  198. }
  199. first = await journal.execute(
  200. principal=Principal("subject", "tenant-a"),
  201. route_family="script-build:phase-three-advance",
  202. command=lambda: execute("advance-a"),
  203. **common,
  204. )
  205. second = await journal.execute(
  206. principal=Principal("subject", "tenant-b"),
  207. route_family="script-build:phase-three-advance",
  208. command=lambda: execute("advance-b"),
  209. **common,
  210. )
  211. third = await journal.execute(
  212. principal=Principal("subject", "tenant-a"),
  213. route_family="script-build:finalize",
  214. command=lambda: execute("finalize-a"),
  215. **common,
  216. )
  217. assert first[1]["label"] == "advance-a"
  218. assert second[1]["label"] == "advance-b"
  219. assert third[1]["label"] == "finalize-a"
  220. assert calls == ["advance-a", "advance-b", "finalize-a"]