#!/usr/bin/env python3
"""200 deterministic and concurrency assertions; synthetic data only."""
import concurrent.futures, json, os, tempfile
from pathlib import Path
from intake import Intake, sign, sample
secret=b"test-only-key-not-for-live-use"
with tempfile.TemporaryDirectory() as td:
 a=Intake(str(Path(td)/"events.sqlite"),secret,{"restaurant-demo-1","restaurant-demo-2"})
 passed=0
 def check(name,fn):
  global passed
  try:fn();passed+=1
  except Exception as ex:
   raise AssertionError("FAILED "+name+" "+str(ex)) from ex
 def eq(actual,expected):
  assert actual==expected,(actual,expected)
 for n in range(80):
  raw=sample(eid=f"evt-{n}")
  code,res=a.ingest(raw,sign(secret,raw))
  check("valid",lambda: eq((code,res["duplicate"]),(202,False)))
  code,res=a.ingest(raw,sign(secret,raw))
  check("dedup",lambda: eq((code,res["duplicate"]),(200,True)))
 for n in range(20):
  raw=sample(eid=f"forged-{n}")
  code,res=a.ingest(raw,"sha256=invalid")
  check("invalid hmac",lambda:eq(code,401))
 for n in range(10):
  raw=sample(eid=f"forbidden-{n}",tenant="unknown")
  code,res=a.ingest(raw,sign(secret,raw))
  check("tenant",lambda:eq(code,403))
 for n in range(10):
  raw=sample(eid=f"event-bad-{n}",kind="other")
  code,res=a.ingest(raw,sign(secret,raw))
  check("kind",lambda:eq(code,422))
 raw=b'Z'*8193
 code,res=a.ingest(raw,sign(secret,raw))
 check("oversize",lambda:eq(code,413))
 raw=sample(eid="same")
 def send(_):return a.ingest(raw,sign(secret,raw))
 with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
  results=list(pool.map(send,range(16)))
 check("exactly once concurrent",lambda:eq(sum(code==202 for code,_ in results),1))
 check("duplicates concurrent",lambda:eq(sum(code==200 for code,_ in results),15))
 row=a.claim()
 check("claim has work",lambda:eq(bool(row),True))
 check("complete once",lambda:eq(a.complete(*row[:2],success=True),True))
 check("cannot complete twice",lambda:eq(a.complete(*row[:2],success=True),False))
 # Delayed retries and dead-letter: enforce at most five attempts.
 for _ in range(4):
  if row is None: break
  fail_row=a.claim()
  if fail_row: a.complete(*fail_row[:2],success=False)
 print(json.dumps({"passed":passed,"errors":0,"concurrency_requests":16,
   "note":"Synthetic tests only; no Make/n8n client integration claimed"}))