test_rust_mate.py

1import sys,time,json,unittest,random,subprocess,os
2from pathlib import Path
3from unittest.mock import patch
4HERE=Path(__file__).resolve().parent
5sys.path[:0]=[str(HERE),str(HERE.parent),'./modelcode']
6if os.environ.get('RUST_MATE_RELEASE'):
7 ROOT=Path(os.environ['RUST_MATE_RELEASE']);sys.path[:0]=[str(ROOT),str(ROOT/'modelcode'),str(ROOT/'lichess_service'),str(ROOT/'lichess_bridge')]
8import chess,harness
9from rust_mate import forced_mate,BINARY
10from mate_override import forced_mate as reference
11FENS=[('mate1','7k/5Q2/6K1/8/8/8/8/8 w - - 0 1'),('quiet_mate3','2K4Q/k7/8/8/8/8/8/8 w - - 0 1'),('castle','4rkr1/4p1p1/8/8/8/8/8/4K2R w K - 0 1'),('underpromotion','3nbn2/3pkpP1/8/4K3/8/8/8/8 w - - 0 1'),('ep','8/4N1B1/8/R2pP2k/5K2/6B1/8/8 w - d6 0 1')]
12class RustMateTests(unittest.TestCase):
13 def search(self,b,depth=3,**kw):return forced_mate(b,adjudicate=harness.outcome,max_plies=depth,seconds=kw.pop('seconds',2),**kw)
14 def test_reference_and_special_moves(self):
15  for name,fen in FENS:
16   b=chess.Board(fen);r=self.search(b);ref=reference(b,adjudicate=harness.outcome,seconds=5)
17   self.assertEqual(r['status'],'proven',(name,r));self.assertEqual(r['mate_plies'],ref['mate_plies']);self.assertTrue(r['proof_verified'])
18   if name in ('castle','underpromotion','ep'):self.assertEqual(r['move'],dict(castle='e1g1',underpromotion='g7g8n',ep='e5d6')[name])
19 def test_cooperative_line_not_proof(self):
20  b=chess.Board('1k6/8/8/1K6/1Q6/8/8/8 w - - 0 1');r=self.search(b)
21  self.assertNotEqual(r['move'],'b5a6')
22 def test_draw_claims_and_terminal_priority(self):
23  for fen in ['7k/5Q2/6K1/8/8/8/8/8 b - - 0 1','7k/8/6K1/8/8/8/8/8 w - - 0 1','1k6/8/8/1K6/1Q6/8/8/8 w - - 99 1','1k6/8/8/1K6/1Q6/8/8/8 w - - 100 1']:
24   self.assertEqual(self.search(chess.Board(fen),10)['status'],'no_forced_mate')
25 def test_repetition_history_and_declared_move(self):
26  b=chess.Board()
27  for m in 'g1f3 g8f6 f3g1 f6g8 g1f3 g8f6 f3g1'.split():b.push_uci(m)
28  self.assertTrue(b.can_claim_threefold_repetition());self.assertEqual(self.search(b,10)['status'],'no_forced_mate')
29 def test_board_restoration(self):
30  b=chess.Board();b.push_uci('e2e4');before=(b.fen(),list(b.move_stack));self.search(b,10,max_nodes=20)
31  self.assertEqual((b.fen(),b.move_stack),before)
32 def test_unknown_budgets_not_disproof(self):
33  r=self.search(chess.Board(),10,max_nodes=1);self.assertEqual(r['status'],'unknown');self.assertIsNone(r['move']);self.assertLessEqual(r['nodes'],1)
34  start=time.monotonic();r=self.search(chess.Board(),10,seconds=.01,max_nodes=1000000);self.assertEqual(r['status'],'unknown');self.assertLess(time.monotonic()-start,.25)
35 def test_depth_limit_and_repeatability(self):
36  b=chess.Board(FENS[1][1]);self.assertEqual(self.search(b,2)['status'],'no_forced_mate')
37  a=self.search(b,10);z=self.search(b,10);self.assertEqual(a['move'],z['move']);self.assertEqual(a['mate_plies'],3)
38  for d in (0,11):
39   with self.assertRaises(ValueError):self.search(b,d)
40 def test_parallel_proofs_deterministic_and_shared_limits(self):
41  b=chess.Board('8/8/8/8/8/2Q1K2k/8/8 w - - 0 1');moves=[]
42  for workers in (1,2,4,8):
43   r=self.search(b,7,workers=workers,max_nodes=100000);self.assertEqual(r['status'],'proven');self.assertTrue(r['proof_verified']);moves.append(r['move'])
44   self.assertEqual(r['iteration_depths'],[1,3,5]);self.assertLessEqual(r['nodes'],100000)
45   limited=self.search(chess.Board(),7,workers=workers,max_nodes=100,seconds=.25)
46   self.assertEqual(limited['status'],'unknown');self.assertLessEqual(limited['nodes'],100)
47   self.assertEqual(limited['iteration_depths'],[1,3]);self.assertEqual(limited['completed_depth_plies'],1)
48  self.assertEqual(len(set(moves)),1)
49 def test_pool_exhaustion_and_release_across_calls(self):
50  from mate_cpu_pool import acquire,release
51  held=acquire(8)
52  try:
53   self.assertEqual(len(held),8);r=self.search(chess.Board(),7,workers=4);self.assertEqual(r['reason'],'cpu_capacity');self.assertEqual(r['workers'],0)
54   import mate_cpu_pool
55   script='import sys;sys.path.insert(0,'+repr(str(Path(mate_cpu_pool.__file__).parent))+');from mate_cpu_pool import acquire;print(len(acquire(8)))'
56   child=subprocess.run([sys.executable,'-c',script],capture_output=True,text=True,timeout=3)
57   self.assertEqual(child.stdout.strip(),'0')
58  finally:release(held)
59  self.assertEqual(self.search(chess.Board(FENS[0][1]),workers=4)['status'],'proven')
60 def test_parallel_cancellation_and_odd_horizons(self):
61  for workers in (2,4,8):
62   r=self.search(chess.Board(),7,workers=workers,seconds=.02,max_nodes=1000000)
63   self.assertEqual(r['status'],'unknown');self.assertTrue(all(d%2==1 for d in r['iteration_depths']))
64  a=self.search(chess.Board(FENS[0][1]),9);b=self.search(chess.Board(FENS[0][1]),10)
65  self.assertEqual(a['move'],b['move']);self.assertEqual(a['iteration_depths'],b['iteration_depths'])
66 def test_proves_five_ply_mate_beyond_old_bound(self):
67  b=chess.Board('8/8/8/8/8/2Q1K2k/8/8 w - - 0 1')
68  self.assertEqual(self.search(b,3)['status'],'no_forced_mate')
69  r=self.search(b,10,max_nodes=200000);self.assertEqual(r['status'],'proven');self.assertEqual(r['mate_plies'],5)
70 def test_invalid_history_native(self):
71  p=subprocess.run([str(BINARY)],input=json.dumps(dict(root_fen=chess.STARTING_FEN,moves=['e2e5'],max_plies=10,max_nodes=100,millis=100)),text=True,capture_output=True)
72  self.assertNotEqual(p.returncode,0);self.assertEqual(json.loads(p.stdout)['status'],'unknown')
73 def test_missing_backend_fails_closed(self):
74  r=forced_mate(chess.Board(),adjudicate=harness.outcome,binary=HERE/'missing');self.assertEqual(r['status'],'unknown');self.assertIsNone(r['move'])
75 def test_process_deadline_kills_and_reaps(self):
76  import tempfile,os
77  with tempfile.TemporaryDirectory() as td:
78   script=Path(td)/'stuck';script.write_text('#!'+sys.executable+'\nimport time\ntime.sleep(60)\n');script.chmod(0o700)
79   real=subprocess.Popen;children=[]
80   def spawn(*args,**kwargs):
81    p=real(*args,**kwargs);children.append(p);return p
82   with patch('rust_mate.subprocess.Popen',side_effect=spawn):r=forced_mate(chess.Board(),adjudicate=harness.outcome,binary=script,seconds=.02)
83   self.assertEqual(r['reason'],'process_deadline');self.assertIsNotNone(children[0].poll())
84 def test_deterministic_random_shallow_reference(self):
85  rng=random.Random(7621);checked=0
86  for _ in range(40):
87   b=chess.Board(None)
88   for sq,piece in zip(rng.sample(range(64),3),['K','Q','k']):b.set_piece_at(sq,chess.Piece.from_symbol(piece))
89   if not b.is_valid():continue
90   ref=reference(b,adjudicate=harness.outcome,max_plies=1,seconds=1);r=self.search(b,1)
91   self.assertEqual(r['status'],ref['status']);checked+=1
92  self.assertGreater(checked,10)
93 def test_false_certificate_rejected(self):
94  b=chess.Board(FENS[1][1]);r=self.search(b)
95  r['proof']['branches'][0]['child']['branches'].pop()
96  with patch('rust_mate.invoke',return_value=(0,json.dumps(r))):self.assertEqual(self.search(b)['reason'],'invalid_proof')
97 def test_default_dispatch_remains_python3(self):
98  import mate_backend,os
99  with patch('mate_backend.configuration',return_value=dict(backend='python',max_plies=3,max_nodes=50000,max_seconds=1.)):
100   r=mate_backend.forced_mate(chess.Board(FENS[0][1]),adjudicate=harness.outcome)
101  self.assertEqual(r['max_plies'],3);self.assertNotIn('backend',r)
102 def test_malformed_schema_complete_unknown(self):
103  good=self.search(chess.Board(FENS[0][1]));cases=['[]','null','{}','['*2000+'0'+']'*2000]
104  for key,value in [('nodes',True),('nodes',-1),('seconds',float('nan')),('seconds',float('inf')),('status',[]),('status','garbage'),('proof',[]),('mate_plies',True),('move',42),('max_plies',11)]:
105   bad=dict(good);bad[key]=value;cases.append(json.dumps(bad))
106  for payload in cases:
107   with self.subTest(payload=payload[:80]),patch('rust_mate.invoke',return_value=(0,payload)):
108    r=self.search(chess.Board(FENS[0][1]));self.assertEqual(r['status'],'unknown');self.assertIsNone(r['move']);self.assertIsNone(r['proof'])
109    for field in ('reason','nodes','node_budget','max_plies','completed_depth_plies','deepest_visited_ply','seconds','time_budget_seconds','backend','proof_verified'):self.assertIn(field,r)
110    json.dumps(r,allow_nan=False)
111 def test_bounded_stdout_and_stderr(self):
112  import tempfile
113  for stream in ('stdout','stderr'):
114   with tempfile.TemporaryDirectory() as td:
115    script=Path(td)/'flood';script.write_text('#!'+sys.executable+'\nimport sys,time\nsys.'+stream+'.write("x"*3000000)\nsys.'+stream+'.flush()\ntime.sleep(60)\n');script.chmod(0o700)
116    r=forced_mate(chess.Board(),adjudicate=harness.outcome,binary=script,seconds=.5)
117    self.assertEqual(r['status'],'unknown');self.assertEqual(r['reason'],'output_limit')
118 @unittest.skipUnless(os.environ.get('RUST_MATE_RELEASE'),'frozen future integration only')
119 def test_malformed_falls_back_in_actual_worker_loop(self):
120  import inference,scorer,training_core,hashlib
121  from chessdata import tokenizer
122  class Model:
123   def score(self,board):return next(iter(board.legal_moves)),{'raw_greedy_tokens':[1,2,3]}
124  class Conn:
125   def __init__(self):self.sent=[];self.done=False
126   def send(self,x):self.sent.append(x)
127   def recv(self):
128    if self.done:raise EOFError
129    self.done=True;return ('request',chess.Board(),1.)
130   def close(self):pass
131  meta={'identity':{'tokenizer':{'sha256':hashlib.sha256(Path(tokenizer.__file__).read_bytes()).hexdigest()}}}
132  for payload in ('[]','null','{}','{"status":"proven"}','{"status":"unknown","seconds":NaN}'):
133   conn=Conn()
134   with patch('training_core.load_model_checkpoint',return_value=(None,meta)),patch('scorer.Scorer',return_value=Model()),patch('rust_mate.invoke',return_value=(0,payload)),patch.dict(os.environ,{'LITTLEPLY_DEVICE':'cpu'}):inference.child(conn,'unused')
135   self.assertEqual(conn.sent[0][0],'ready');result=conn.sent[1]
136   self.assertEqual(result[0],'request');self.assertEqual(result[2]['model_query'],'queried');self.assertEqual(result[2]['mate_search']['status'],'unknown')
137 @unittest.skipUnless(os.environ.get('RUST_MATE_RELEASE'),'frozen future integration only')
138 def test_future_eval_game_records_config_and_fallback(self):
139  import tempfile
140  from test_harness import Model,MockEngine
141  with tempfile.TemporaryDirectory() as td:
142   r=harness.game(Model(),MockEngine,chess.WHITE,Path(td)/'mock',cap=2,mate_backend='rust',mate_plies=10,mate_nodes=1,mate_seconds=.1,label='OFFLINE MOCK TEST')
143   self.assertEqual(r['mate_max_plies'],10)
144   self.assertEqual(r['moves'][0]['mate_search']['status'],'unknown')
145   self.assertEqual(r['moves'][0]['model_query'],'queried')
146   self.assertIn('10-ply-override-rust',r['model_policy'])
147if __name__=='__main__':unittest.main()
148