Use mock process to test rare UCI commands

chess/uci.py

@@ -475,11 +475,11 @@ def __init__(self, engine):
         self._is_dead = threading.Event()
         self._std_streams_closed = False
 
-    def expect(self, expectation):
-        self._expectations.append(expectation)
+    def expect(self, expectation, responses=()):
+        self._expectations.append((expectation, responses))
 
     def assert_done(self):
-        assert not self._expectations
+        assert not self._expectations, "pending expectations: {0}".format(self._expectations)
 
     def assert_terminated(self):
         self.assert_done()
@@ -491,17 +491,24 @@ def is_alive(self):
 
     def terminate(self):
         self._is_dead.set()
-        self.on_terminated()
+        self.engine.on_terminated()
 
     def kill(self):
         self._is_dead.set()
-        self.on_terminated()
+        self.engine.on_terminated()
 
     def close_std_streams(self):
         self._std_streams_closed = True
 
-    def send_line(string):
-        assert self._expectations.popleft() == string
+    def send_line(self, string):
+        assert self.is_alive()
+
+        assert self._expectations, "unexpected: {0}".format(string)
+        expectation, responses = self._expectations.popleft()
+        assert expectation == string, "expected: {0}, got {1}".format(expectation, string)
+
+        for response in responses:
+            self.engine.on_line_received(response)
 
     def wait_for_return_code(self):
         self._is_dead.wait()
@@ -782,7 +789,7 @@ def handle_move_token(token, fn):
         for token in arg.split(" "):
             if current_parameter == "string":
                 string.append(token)
-            elif token in ("depth", "seldepth", "time", "nodes", "pv", "multipv", "score", "currmove", "currmovenumber", "hashfull", "nps", "tbhits", "cpuload", "refutation", "currline"):
+            elif token in ("depth", "seldepth", "time", "nodes", "pv", "multipv", "score", "currmove", "currmovenumber", "hashfull", "nps", "tbhits", "cpuload", "refutation", "currline", "string"):
                 end_of_parameter()
                 current_parameter = token
 

test.py

@@ -1230,23 +1230,85 @@ def test_kill(self):
         self.assertFalse(engine.is_alive())
 
 
-class UciEngineTestCase(object):
+class UciEngineTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.engine = chess.uci.Engine(chess.uci.MockProcess, ())
+        self.mock = self.engine.process
+
+        self.mock.expect("uci", ("uciok", ))
+        self.engine.uci()
+        self.mock.assert_done()
+
+    def tearDown(self):
+        self.engine.terminate()
+        self.mock.assert_terminated()
 
     def test_debug(self):
-        engine = chess.uci.Engine(chess.uci.MockProcess, ())
-        mock = engine.process
+        self.mock.expect("debug on")
+        self.engine.debug(True)
+        self.mock.assert_done()
 
-        mock.expect("uci")
-        engine.uci()
-        mock.assert_done()
-        engine.on_line_received("uciok")
+        self.mock.expect("debug off")
+        self.engine.debug(False)
+        self.mock.assert_done()
 
-        mock.expect("debug on")
-        engine.debug(True)
-        mock.assert_done()
+    def test_ponderhit(self):
+        self.mock.expect("ponderhit")
+        self.mock.expect("isready", ("readyok", ))
+        self.engine.ponderhit()
+        self.mock.assert_done()
 
-        engine.terminate()
-        mock.assert_terminated()
+    def test_kill(self):
+        self.engine.kill()
+        self.mock.assert_terminated()
+
+    def test_go(self):
+        self.mock.expect("go searchmoves e2e4 d2d4 ponder infinite")
+        self.engine.go(
+            searchmoves=[chess.Move.from_uci("e2e4"), chess.Move.from_uci("d2d4")],
+            ponder=True,
+            infinite=True)
+        self.mock.assert_done()
+
+        self.mock.expect("stop", ("bestmove e2e4", ))
+        self.mock.expect("isready", ("readyok", ))
+        bestmove, pondermove = self.engine.stop()
+        self.mock.assert_done()
+        self.assertEqual(bestmove, chess.Move.from_uci("e2e4"))
+        self.assertTrue(pondermove is None)
+
+        self.mock.expect("go winc 3 binc 4 movestogo 5 depth 6 nodes 7 mate 8 movetime 9", (
+            "bestmove d2d4 ponder d7d5",
+        ))
+        self.engine.go(winc=3, binc=4, movestogo=5, depth=6, nodes=7, mate=8, movetime=9)
+        self.mock.assert_done()
+
+    def test_refutations(self):
+        handler = chess.uci.InfoHandler()
+        self.engine.info_handlers.append(handler)
+
+        self.engine.on_line_received("info refutation d1h5 g6h5")
+
+        d1h5 = chess.Move.from_uci("d1h5")
+        g6h5 = chess.Move.from_uci("g6h5")
+
+        with handler as info:
+            self.assertEqual(len(info["refutation"][d1h5]), 1)
+            self.assertEqual(info["refutation"][d1h5][0], g6h5)
+
+        self.engine.on_line_received("info refutation d1h5")
+        with handler as info:
+            self.assertTrue(info["refutation"][d1h5] is None)
+
+    def test_string(self):
+        handler = chess.uci.InfoHandler()
+        self.engine.info_handlers.append(handler)
+
+        self.engine.on_line_received("info string goes to end no matter score cp 4 what")
+        with handler as info:
+            self.assertEqual(info["string"], "goes to end no matter score cp 4 what")
+            self.assertFalse("score" in info)
 
 
 if __name__ == "__main__":