import concurrent.futures, hashlib, hmac, http.client, json, os, socket, sqlite3, tempfile, threading, time, unittest from pathlib import Path from delivery_queue import Queue, Problem, DestinationPolicy, sign, send_request from receiver import Receiver, make_server, SECRET from app import make_server as api_server class Clock: def __init__(self): self.now=1000.0 def __call__(self): return self.now class QueueTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.clock=Clock() self.q=Queue(Path(self.tmp.name)/'q.sqlite',DestinationPolicy(test_origins=['http://127.0.0.1:9081']),self.clock) self.q.add_endpoint('a','http://127.0.0.1:9081/ok',SECRET,retry_base=1) def tearDown(self): self.tmp.cleanup() def event(self,ident='e1',targets=None): return self.q.accept(ident,'asset.checked',{'v':1},targets or ['a']) def test_accept_is_atomic_across_destinations(self): with self.assertRaises(Problem):self.event(targets=['a','missing']) with self.q.connect() as db:self.assertEqual(db.execute('SELECT COUNT(*) FROM events').fetchone()[0],0) def test_identical_retry_and_conflicting_id(self): self.assertFalse(self.event()['duplicate']);self.assertTrue(self.event()['duplicate']) with self.assertRaises(Problem) as e:self.q.accept('e1','asset.checked',{'v':2},['a']) self.assertEqual(e.exception.status,409);self.assertEqual(len(self.q.snapshot()['jobs']),1) def test_restart_preserves_committed_event(self): self.event();again=Queue(self.q.path,self.q.policy,self.clock) self.assertEqual(again.claim()['event_id'],'e1') def test_worker_crash_expires_and_fences_stale_completion(self): self.event();old=self.q.claim(lease_seconds=5);self.clock.now+=6;new=self.q.claim() self.assertNotEqual(old['lease_token'],new['lease_token']) self.assertFalse(self.q.complete(old['id'],old['lease_token'],200)) self.assertTrue(self.q.complete(new['id'],new['lease_token'],200)) self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'delivered') def test_expired_owner_cannot_finish_without_replacement(self): self.event();job=self.q.claim(1);self.clock.now+=2 self.assertFalse(self.q.complete(job['id'],job['lease_token'],200)) def test_paused_endpoint_and_resume(self): self.event();self.q.pause('a',True);self.assertIsNone(self.q.claim());self.q.pause('a',False);self.assertIsNotNone(self.q.claim()) def test_slow_endpoint_capacity_does_not_block_other_endpoint(self): self.q.add_endpoint('b','http://127.0.0.1:9081/ok',SECRET) self.event('e1',['a']);self.event('e2',['a']);self.event('e3',['b']) first=self.q.claim();second=self.q.claim() self.assertEqual(first['endpoint_id'],'a');self.assertEqual(second['endpoint_id'],'b');self.assertIsNone(self.q.claim()) def test_bounded_backoff_and_dead_letter(self): self.event() for n in range(1,5): job=self.q.claim();self.assertEqual(job['attempts'],n);self.q.complete(job['id'],job['lease_token'],503) if n<4: self.assertIsNone(self.q.claim());self.clock.now+=2**(n-1) self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'dead') def test_permanent_rejection_does_not_retry(self): self.event();j=self.q.claim();self.q.complete(j['id'],j['lease_token'],422) self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'dead');self.assertIsNone(self.q.claim()) def test_redirect_is_terminal(self): self.event();j=self.q.claim();self.q.complete(j['id'],j['lease_token'],302) self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'dead') def test_replay_keeps_event_id_and_attempt_history(self): self.event();j=self.q.claim();self.q.complete(j['id'],j['lease_token'],422);self.q.replay(j['id']) replay=self.q.claim();self.assertEqual(replay['event_id'],'e1');self.assertEqual(replay['generation'],1) self.q.complete(replay['id'],replay['lease_token'],200);self.assertEqual(len(self.q.snapshot()['attempts']),2) with self.assertRaises(Problem):self.q.replay(j['id']) def test_crash_budget_is_bounded(self): self.event() for _ in range(4):self.assertIsNotNone(self.q.claim(1));self.clock.now+=2 self.assertIsNone(self.q.claim());self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'dead') def test_concurrent_claims_have_one_owner(self): with self.q.transaction() as db:db.execute('UPDATE endpoints SET max_inflight=16') for n in range(16):self.event(f'e{n}') with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:jobs=list(pool.map(lambda _:self.q.claim(),range(24))) claimed=[j for j in jobs if j];self.assertEqual(len(claimed),16);self.assertEqual(len({j['id'] for j in claimed}),16) def test_retention_preserves_pending_and_rejects_expired_replay(self): self.event('done');j=self.q.claim();self.q.complete(j['id'],j['lease_token'],422);self.event('pending');self.clock.now+=100 self.assertEqual(self.q.purge(self.clock.now),1) with self.assertRaises(Problem) as e:self.q.replay(j['id']) self.assertEqual(e.exception.status,410);self.assertEqual(self.q.claim()['event_id'],'pending') def test_signing_secret_is_absent_from_snapshot_and_audit(self): self.event();self.assertNotIn(SECRET,json.dumps(self.q.snapshot())) with self.q.connect() as db:self.assertNotIn(SECRET,str(db.execute('SELECT * FROM audit').fetchall())) def test_bad_payloads_are_rejected(self): for data in [[],{'nan':float('nan')},{'x':'x'*70000}]: with self.assertRaises(Problem):self.q.accept('invalid','type',data,['a']) class PolicyTests(unittest.TestCase): def test_defaults_deny_all_destinations(self): for url in ['http://127.0.0.1/x','https://example.com/x','file:///etc/passwd']: with self.assertRaises(Problem):DestinationPolicy().validate(url) def test_credentials_fragment_and_http_are_rejected(self): p=DestinationPolicy(['example.com']) for url in ['https://user:pw@example.com/','https://example.com/#secret','http://example.com/','https://example.com:8443/']: with self.assertRaises(Problem):p.validate(url,False) def test_dns_mixed_public_private_is_rejected(self): def resolve(*a,**k):return [(socket.AF_INET,socket.SOCK_STREAM,6,'',(x,443)) for x in ['8.8.8.8','10.0.0.1']] with self.assertRaises(Problem):DestinationPolicy(['example.com'],resolver=resolve).validate('https://example.com/') def test_ipv6_mapped_and_linklocal_are_rejected(self): for ip in ['::ffff:127.0.0.1','fe80::1','::1','fc00::1','ff02::1','fec0::1']: p=DestinationPolicy(['example.com'],resolver=lambda *a,ip=ip,**k:[(socket.AF_INET6,socket.SOCK_STREAM,6,'',(ip,443,0,0))]) with self.assertRaises(Problem):p.validate('https://example.com/') def test_global_address_is_returned_for_pinned_connection(self): p=DestinationPolicy(['example.com'],resolver=lambda *a,**k:[(socket.AF_INET,socket.SOCK_STREAM,6,'',('8.8.8.8',443))]) self.assertEqual(p.validate('https://example.com/')[1],'8.8.8.8') def test_test_exception_is_exact_origin(self): p=DestinationPolicy(test_origins=['http://127.0.0.1:9081']) self.assertEqual(p.validate('http://127.0.0.1:9081/ok')[1],'127.0.0.1') for url in ['http://127.0.0.1:9082/','http://localhost:9081/','http://127.0.0.2:9081/']: with self.assertRaises(Problem):p.validate(url) class HttpTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory();self.receiver=Receiver(Path(self.tmp.name)/'r.sqlite');self.server=make_server(self.receiver,0) self.thread=threading.Thread(target=self.server.serve_forever,daemon=True);self.thread.start() self.origin=f'http://127.0.0.1:{self.server.server_port}';self.policy=DestinationPolicy(test_origins=[self.origin]) self.q=Queue(Path(self.tmp.name)/'q.sqlite',self.policy) def tearDown(self):self.server.shutdown();self.server.server_close();self.thread.join();self.tmp.cleanup() def test_lost_response_retries_same_id_with_one_business_effect(self): self.q.add_endpoint('r',self.origin+'/drop-once',SECRET,retry_base=.01) self.q.accept('e1','test',{},['r']);self.q.work_once();self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'retry') time.sleep(.02);self.q.work_once();self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'delivered') receipt=self.receiver.snapshot()[0];self.assertEqual(receipt['hits'],2);self.assertEqual(receipt['effects'],1) def test_valid_signature_and_invalid_signature(self): body=b'{"id":"e1"}' self.assertEqual(send_request(self.policy,self.origin+'/ok',SECRET,'e1',body,1),200) self.assertEqual(send_request(self.policy,self.origin+'/ok','wrong-signing-secret','e1',body,1),401) def test_redirect_is_not_followed(self): self.assertEqual(send_request(self.policy,self.origin+'/redirect',SECRET,'e1',b'{"id":"e1"}',1),302) self.assertEqual(self.receiver.snapshot(),[]) def test_timeout_completes_attempt_as_retry(self): self.q.add_endpoint('r',self.origin+'/slow',SECRET,timeout=.02,retry_base=.01) self.q.accept('e1','test',{},['r']);self.q.work_once();self.assertEqual(self.q.snapshot()['jobs'][0]['status'],'retry') def test_operator_api_auth_and_event_acceptance(self): self.q.add_endpoint('r',self.origin+'/ok',SECRET) api=api_server(self.q,'test-administrator-token',0);thread=threading.Thread(target=api.serve_forever,daemon=True);thread.start() try: def request(method,path,body=None,authorized=True): c=http.client.HTTPConnection('127.0.0.1',api.server_port);headers={'Content-Type':'application/json'} if authorized:headers['Authorization']='Bearer test-administrator-token' c.request(method,path,json.dumps(body) if body is not None else None,headers);r=c.getresponse();v=r.read();status=r.status;c.close();return status,v self.assertEqual(request('GET','/api/state',authorized=False)[0],401) self.assertEqual(request('POST','/api/events',{'id':'api-event','type':'sample','data':{},'endpoints':['r']})[0],201) self.assertEqual(request('POST','/api/pause',{'endpoint':'r','paused':True})[0],200) self.assertIsNone(self.q.claim()) self.assertEqual(request('GET','/../../delivery_queue.py')[0],404) finally:api.shutdown();api.server_close();thread.join() class StartupTests(unittest.TestCase): def test_demo_setup_and_repeat_start(self): from app import setup with tempfile.TemporaryDirectory() as d: q=setup(d+'/queue.sqlite',demo=True);self.assertEqual(len(q.snapshot()['endpoints']),3) q=setup(d+'/queue.sqlite',demo=True);self.assertEqual(len(q.snapshot()['endpoints']),3) if __name__=='__main__':unittest.main(verbosity=2)