test_bridge.py

1import unittest,tempfile,copy,json,io
2from pathlib import Path
3import chess,chess.pgn
4from core import Game,challenge_reason,reconstruct,budget
5from unittest.mock import Mock
6from runtime import Policy
7import sys
8sys.path.insert(0,str(Path(__file__).resolve().parent.parent))
9import harness as h
10
11def full(moves='',status='started'):
12    return {'id':'AbCd1234','rated':False,'variant':{'key':'standard'},'initialFen':'startpos',
13        'white':{'id':'littleply','name':'LittlePly','title':'BOT'},'black':{'id':'private_test_user','name':'private_test_user'},
14        'clock':{'initial':600000,'increment':5000},
15        'state':{'moves':moves,'status':status,'wtime':600000,'btime':600000,'winc':5000,'binc':5000}}
16def policy(b,seconds):
17    return sorted(b.legal_moves,key=lambda m:m.uci())[0],{'decision_source':'model_policy','model_query':'queried'}
18
19class Tests(unittest.TestCase):
20    def test_challenge_whitelist(self):
21        c={'challenger':{'id':'private_test_user'},'destUser':{'id':'littleply'},'rated':False,
22            'variant':{'key':'standard'},'timeControl':{'type':'clock','limit':600,'increment':5}}
23        self.assertIsNone(challenge_reason(c,'private_test_user','LittlePly'))
24        self.assertEqual(challenge_reason(c,'private_test_user','LittlePly',True),'later')
25        for k,v in [('rated',True),('challenger',{'id':'someoneelse'}),('variant',{'key':'chess960'}),
26            ('initialFen','arbitrary'),('timeControl',{'type':'clock','limit':60,'increment':0}),('timeControl',{'type':'correspondence'})]:
27            x=copy.deepcopy(c);x[k]=v;self.assertIsNotNone(challenge_reason(x,'private_test_user','LittlePly'))
28    def test_full_game_scope(self):
29        with tempfile.TemporaryDirectory() as d:
30            for key,value in [('rated',True),('initialFen','arbitrary'),('white',{'id':'stranger'}),('clock',{'initial':60000,'increment':0})]:
31                data=full();data[key]=value
32                with self.assertRaises(ValueError):Game(data,'littleply','private_test_user',d)
33    def test_idempotence_and_resync(self):
34        with tempfile.TemporaryDirectory() as d:
35            game=Game(full(),'littleply','private_test_user',d);entry=game.decide(policy,fresh=True,now=10)
36            game.mark(entry,'sending',now=10);game.mark(entry,'uncertain')
37            self.assertIsNone(game.decide(policy,now=20))
38            self.assertIsNone(game.decide(policy,fresh=True,now=11))
39            self.assertIs(game.decide(policy,fresh=True,now=20),entry)
40            game.mark(entry,'accepted');self.assertIsNone(game.decide(policy,fresh=True,now=30))
41            game.ingest(full(entry['uci'])['state']);self.assertIsNone(game.decide(policy,fresh=True))
42    def test_uncertain_move_already_committed(self):
43        with tempfile.TemporaryDirectory() as d:
44            game=Game(full(),'littleply','private_test_user',d);entry=game.decide(policy)
45            game.mark(entry,'sending');game.mark(entry,'uncertain')
46            game.ingest(full(entry['uci']+' e7e5')['state'])
47            next_entry=game.decide(policy,fresh=True)
48            self.assertNotEqual(next_entry['key'],entry['key']);self.assertEqual(next_entry['ply'],3)
49    def test_pending_persists_restart(self):
50        with tempfile.TemporaryDirectory() as d:
51            game=Game(full(),'littleply','private_test_user',d);entry=game.decide(policy,now=0)
52            game.mark(entry,'sending',now=0);game.mark(entry,'uncertain')
53            restored=Game(full(),'littleply','private_test_user',d)
54            def never(*a):raise AssertionError('must reuse exact pending move')
55            self.assertEqual(restored.decide(never,fresh=True,now=10)['uci'],entry['uci'])
56    def test_retry_bound(self):
57        with tempfile.TemporaryDirectory() as d:
58            g=Game(full(),'littleply','private_test_user',d);e=g.decide(policy)
59            for n in range(3):g.mark(e,'sending',now=n);g.mark(e,'uncertain')
60            self.assertIsNone(g.decide(policy,fresh=True,now=100))
61    def test_history_castling_ep_promotion(self):
62        castle='e2e4 e7e5 g1f3 b8c6 f1c4 g8f6 e1h1'
63        b=reconstruct(castle);self.assertEqual(b.peek().uci(),'e1g1');self.assertEqual(b.piece_at(chess.F1).piece_type,chess.ROOK)
64        b=reconstruct('e2e4 a7a6 e4e5 d7d5 e5d6');self.assertIsNone(b.piece_at(chess.D5))
65        b=reconstruct('a2a4 b8c6 a4a5 h7h5 a5a6 h5h4 a6b7 h4h3 b7a8n')
66        self.assertEqual(b.piece_at(chess.A8).piece_type,chess.KNIGHT)
67        with self.assertRaises(ValueError):reconstruct('e2e5')
68    def test_clock_budget(self):
69        s=full()['state'];self.assertLessEqual(budget(s,True),8)
70        s['wtime']=1000;self.assertLess(budget(s,True),.2)
71        s['wtime']=600000;s['expiration']={'millisToMove':5000,'idleMillis':1000}
72        self.assertLessEqual(budget(s,True,age=1),1)
73    def test_pgn_server_result_and_annotations(self):
74        with tempfile.TemporaryDirectory() as d:
75            f=full('f2f3 e7e5 g2g4');f['white'],f['black']=f['black'],f['white']
76            g=Game(f,'littleply','private_test_user',d)
77            def mate(b,s):return chess.Move.from_uci('d8h4'),{'decision_source':'forced_mate_override','model_query':'not_queried','mate_search':{'mate_plies':1}}
78            e=g.decide(mate);g.mark(e,'accepted');state=full('f2f3 e7e5 g2g4 d8h4','mate')['state'];state['winner']='black';g.ingest(state)
79            parsed=chess.pgn.read_game(io.StringIO((Path(d)/g.id/'game.pgn').read_text()))
80            self.assertEqual(parsed.headers['Result'],'0-1');self.assertTrue(parsed.end().board().is_checkmate())
81            self.assertIn('MATE HELPER',parsed.end().comment);self.assertIsNone(g.decide(policy))
82    def test_policy_helper_skips_worker(self):
83        b=reconstruct('f2f3 e7e5 g2g4');worker=Mock();worker.context_length=1024
84        move,meta=Policy(worker,h)(b,5)
85        self.assertEqual(move.uci(),'d8h4');self.assertEqual(meta['model_query'],'not_queried');worker.choose.assert_not_called()
86    def test_policy_fallback_context_and_timeout(self):
87        worker=Mock();worker.context_length=1;worker.p.is_alive.return_value=True
88        move,meta=Policy(worker,h)(chess.Board(),5)
89        self.assertIn(move,chess.Board().legal_moves);self.assertEqual(meta['decision_source'],'legal_fallback')
90        self.assertIn('no truncation',meta['reason']);worker.choose.assert_not_called()
91        worker.context_length=1024;worker.choose.side_effect=TimeoutError()
92        move,meta=Policy(worker,h)(chess.Board(),5)
93        self.assertIn(move,chess.Board().legal_moves);self.assertIn('TimeoutError',meta['reason'])
94    def test_policy_clock_fallback(self):
95        worker=Mock();worker.context_length=1024
96        move,meta=Policy(worker,h)(chess.Board(),.02)
97        self.assertIn(move,chess.Board().legal_moves);self.assertEqual(meta['decision_source'],'legal_fallback');worker.choose.assert_not_called()
98
99if __name__=='__main__':unittest.main(verbosity=2)
100