test_service.py

1import unittest,sys,tempfile,queue,time,threading,json
2from pathlib import Path
3from unittest.mock import Mock,patch
4from app import admission,fresh_history,Service,verify_release
5from transport import Transport,Cooldown,Events
6from core import Game,position_key
7import chess
8
9def challenge():return dict(id='AbCd1234',challenger={'id':'anyhuman'},destUser={'id':'littleply'},rated=False,variant={'key':'standard'},timeControl={'type':'clock','limit':600,'increment':5})
10def full():return dict(type='gameFull',id='AbCd1234',white={'id':'littleply'},black={'id':'anyhuman'},rated=False,variant={'key':'standard'},initialFen='startpos',clock={'initial':600000,'increment':5000},state={'moves':'','status':'started','wtime':600000,'btime':600000,'winc':5000,'binc':5000})
11class Response:
12    def __init__(self,code,body=None):self.status_code=code;self.headers={};self.body=body or {'ok':True}
13    def __enter__(self):return self
14    def __exit__(self,*args):pass
15    def json(self):return self.body
16
17class Tests(unittest.TestCase):
18    def test_game_updates_precede_challenge_backlog(self):
19        q=Events()
20        for i in range(32):q.put_nowait(('account','data',i,1,0))
21        with self.assertRaises(queue.Full):q.put_nowait(('account','data',33,1,0))
22        q.put_nowait(('AbCd1234','data','move',1,0));self.assertEqual(q.get_nowait()[0],'AbCd1234')
23    def test_public_admission(self):
24        c=challenge();self.assertIsNone(admission(c,'LittlePly'));self.assertEqual(admission(c,'LittlePly',True),'later')
25        for field,value in [('rated',True),('variant',{'key':'chess960'}),('initialFen','8/8'),('destUser',{'id':'other'})]:
26            x=dict(c);x[field]=value;self.assertIsNotNone(admission(x,'LittlePly'))
27    def test_clock_bounds(self):
28        for limit,inc in [(60,5),(600,0),(3600,5),(600,60)]:
29            c=challenge();c['timeControl'].update(limit=limit,increment=inc);self.assertIsNotNone(admission(c,'LittlePly'))
30    def test_stale_and_divergent_histories(self):
31        self.assertFalse(fresh_history('e2e4 e7e5','e2e4'));self.assertFalse(fresh_history('e2e4','d2d4'))
32        self.assertTrue(fresh_history('e2e4','e2e4 e7e5'));self.assertTrue(fresh_history('e2e4','e2e4'))
33    def transport(self,code):
34        td=tempfile.TemporaryDirectory();self.addCleanup(td.cleanup);p=Path(td.name)/'secret';p.write_text('test_token')
35        req=Mock(return_value=Response(code));t=Transport(p,req);self.addCleanup(t.close);return t,req
36    def test_429_shared_cooldown_no_immediate_retry(self):
37        t,req=self.transport(429);t.submit('x','POST','/api/challenge/AbCd1234/accept')
38        self.assertEqual(t.results.get(timeout=2)[1],'rate_limited');self.assertGreater(t.cooldown.remaining(),59)
39        t.submit('y','GET','/api/account');time.sleep(.25);self.assertEqual(req.call_count,1)
40    def test_400_is_known_rejection(self):
41        t,_=self.transport(400);t.submit('x','POST','/api/bot/game/AbCd1234/move/e2e4')
42        self.assertEqual(t.results.get(timeout=2)[1],'rejected')
43    def test_500_post_is_uncertain(self):
44        t,_=self.transport(500);t.submit('x','POST','/api/bot/game/AbCd1234/move/e2e4')
45        self.assertEqual(t.results.get(timeout=2)[1],'uncertain')
46    def test_auth_error(self):
47        t,_=self.transport(401);t.submit('x','GET','/api/account');self.assertEqual(t.results.get(timeout=2)[1],'auth_error')
48    def test_guard_cancels_terminal_move(self):
49        t,req=self.transport(200);t.submit('x','POST','/api/bot/game/AbCd1234/move/e2e4',guard=lambda:False)
50        self.assertEqual(t.results.get(timeout=2)[1],'cancelled');req.assert_not_called()
51    def test_control_exception_does_not_kill_owner(self):
52        t,req=self.transport(200);req.side_effect=__import__('requests').ConnectionError()
53        t.submit('x','POST','/api/challenge/AbCd1234/decline');self.assertEqual(t.results.get(timeout=2)[1],'uncertain')
54        self.assertTrue(t.thread.is_alive())
55    def test_invalid_origin(self):
56        t,_=self.transport(200)
57        with self.assertRaises(ValueError):t.submit('x','GET','https://example.com')
58    def test_control_queue_bounded(self):
59        t,_=self.transport(200);t.cooldown.defer()
60        for i in range(32):t.submit(str(i),'GET','/api/account')
61        with self.assertRaises(queue.Full):t.submit('overflow','GET','/api/account')
62    def service(self):
63        td=tempfile.TemporaryDirectory();self.addCleanup(td.cleanup)
64        s=Service.__new__(Service);s.game=Game(full(),'LittlePly','anyhuman',Path(td.name));s.epoch=7
65        s.event=Mock();s.jobs={};s.retry=[];s.api=Mock();s.api.cooldown.remaining.return_value=0
66        s.stream=Mock();s.inflight=None;s.pending=None;s.reservation=None;s.owner=Mock();s.owner.ready=False;s.owner.busy=False
67        s.owner.results=queue.Queue();s.bad_policy=False
68        return s
69    def test_keyed_stale_response_discarded(self):
70        s=self.service();key=(s.game.id,6,position_key(s.game.board));s.owner.results.put((key,'ok',('e2e4',{})))
71        s.work();self.assertFalse(s.game.journal['decisions']);s.event.assert_called_with('stale_inference_discarded')
72    def test_rejected_move_requests_resync(self):
73        s=self.service();entry={'uci':'e2e4','key':'x','ply':1,'metrics':{},'attempts':1,'submission':'sending'}
74        s.game.journal['decisions']['x']=entry;s.inflight=('move:x',entry)
75        s.response('move:x','rate_limited',{});self.assertEqual(entry['submission'],'rejected');s.stream.resync.set.assert_called_once()
76    def test_no_send_after_terminal(self):
77        s=self.service();digest=position_key(s.game.board)
78        s.game.state['status']='resign';self.assertFalse(s.identity_matches(s.game.id,7,digest))
79    def test_model_failure_is_explicit(self):
80        s=self.service();s.api.cooldown.remaining.return_value=60
81        s.owner.results.put(((s.game.id,7,position_key(s.game.board)),'error','TimeoutError'));s.work()
82        entry=next(iter(s.game.journal['decisions'].values()))
83        self.assertEqual(entry['metrics']['decision_source'],'legal_fallback');self.assertTrue(s.bad_policy)
84    def test_manifest_tampering(self):
85        import hashlib
86        with tempfile.TemporaryDirectory() as td:
87            p=Path(td);(p/'x').write_text('safe');(p/'manifest.json').write_text(json.dumps({'files':{'x':hashlib.sha256(b'safe').hexdigest()}}))
88            verify_release(p);(p/'x').write_text('changed')
89            with self.assertRaises(ValueError):verify_release(p)
90    def test_manifest_path_escape(self):
91        with tempfile.TemporaryDirectory() as td:
92            p=Path(td);(p/'manifest.json').write_text(json.dumps({'files':{'../bad':'x'}}))
93            with self.assertRaises(ValueError):verify_release(p)
94    def test_ten_challenges_reserve_only_one_slot(self):
95        s=self.service();s.args=Mock(bot='LittlePly');s.reservation=None;s.stream=None
96        s.accepting=lambda:not s.reservation;s.api_job=Mock(return_value=True)
97        for i in range(10):
98            c=challenge();c['id']='Game%04d'%i
99            s.ingest('account','data',{'type':'challenge','challenge':c},1,time.monotonic())
100        calls=[c for c in s.api_job.call_args_list if c.args[0].startswith('accept:')]
101        self.assertEqual(len(calls),1)
102    def test_duplicate_accepted_move_not_requeued(self):
103        s=self.service();s.args=Mock(bot='LittlePly');s.stream.tag=s.game.id
104        digest=position_key(s.game.board);s.game.journal['decisions'][digest]={'submission':'accepted'}
105        s.ingest(s.game.id,'data',full(),8,time.monotonic())
106        self.assertIsNone(s.pending)
107    def test_old_stream_epoch_ignored(self):
108        s=self.service();s.stream.tag=s.game.id
109        state=dict(full()['state'],moves='e2e4')
110        s.ingest(s.game.id,'data',dict(type='gameState',**state),6,time.monotonic())
111        self.assertEqual(s.game.board.ply(),0)
112    def test_disk_failure_does_not_send_move(self):
113        s=self.service();s.api.cooldown.remaining.return_value=60
114        s.owner.results.put(((s.game.id,7,position_key(s.game.board)),'ok',('e2e4',{'decision_source':'model_policy'})))
115        with patch.object(s.game,'save',side_effect=OSError('disk full')):
116            with self.assertRaises(OSError):s.work()
117        s.api.submit.assert_not_called()
118
119if __name__=='__main__':unittest.main()
120