feat(cosy): CoSy subprocess integration with stdin/stdout pipe, vocab policy, backup-before-redefine

src/rook/cosy.py

@@ -0,0 +1,39 @@
+import subprocess
+from rook.policy import is_approved_dir, backup_before_edit
+
+
+class PolicyError(Exception):
+ pass
+
+
+class CoSyProcess:
+ def __init__(self, cosy_bat: str = 'CoSy.bat') -> None:
+  self._bat = cosy_bat
+  self._proc = None
+
+ def start(self) -> None:
+  self._proc = subprocess.Popen(
+   ['cmd', '/c', self._bat],
+   stdin=subprocess.PIPE,
+   stdout=subprocess.PIPE,
+   stderr=subprocess.STDOUT,
+   text=True,
+  )
+
+ def send(self, expr: str) -> str:
+  self._proc.stdin.write(expr + '\n')
+  self._proc.stdin.flush()
+  return self._proc.stdout.readline().rstrip('\n')
+
+ def stop(self) -> None:
+  if self._proc:
+   self._proc.terminate()
+
+ def load_vocab(self, vocab_path: str) -> None:
+  if not is_approved_dir(vocab_path):
+   raise PolicyError(f'vocab path not approved: {vocab_path}')
+  self.send(f'FLOAD {vocab_path}')
+
+ def redefine_word(self, name: str, definition: str) -> None:
+  backup_before_edit(name)
+  self.send(definition)

tests/test_cosy.py

@@ -0,0 +1,71 @@
+import subprocess
+import pytest
+from unittest.mock import patch, MagicMock, call
+
+from rook.cosy import CoSyProcess, PolicyError
+
+
+def test_start_launches_popen():
+ with patch('rook.cosy.subprocess.Popen') as mock_popen:
+  mock_proc = MagicMock()
+  mock_popen.return_value = mock_proc
+  cosy = CoSyProcess()
+  cosy.start()
+  mock_popen.assert_called_once_with(
+   ['cmd', '/c', 'CoSy.bat'],
+   stdin=subprocess.PIPE,
+   stdout=subprocess.PIPE,
+   stderr=subprocess.STDOUT,
+   text=True,
+  )
+
+
+def test_send_writes_expr_and_returns_output():
+ with patch('rook.cosy.subprocess.Popen') as mock_popen:
+  mock_stdin = MagicMock()
+  mock_stdout = MagicMock()
+  mock_stdout.readline.return_value = 'ok\n'
+  mock_proc = MagicMock()
+  mock_proc.stdin = mock_stdin
+  mock_proc.stdout = mock_stdout
+  mock_popen.return_value = mock_proc
+  cosy = CoSyProcess()
+  cosy.start()
+  result = cosy.send('1 2 +')
+  mock_stdin.write.assert_called_with('1 2 +\n')
+  assert result == 'ok'
+
+
+def test_stop_terminates_process():
+ with patch('rook.cosy.subprocess.Popen') as mock_popen:
+  mock_proc = MagicMock()
+  mock_popen.return_value = mock_proc
+  cosy = CoSyProcess()
+  cosy.start()
+  cosy.stop()
+  mock_proc.terminate.assert_called_once()
+
+
+def test_load_vocab_approved():
+ with patch('rook.cosy.is_approved_dir', return_value=True):
+  cosy = CoSyProcess()
+  cosy.send = MagicMock()
+  cosy.load_vocab('/approved/vocab.fs')
+  args, _ = cosy.send.call_args
+  assert 'vocab.fs' in args[0]
+
+
+def test_load_vocab_unapproved_raises():
+ with patch('rook.cosy.is_approved_dir', return_value=False):
+  cosy = CoSyProcess()
+  with pytest.raises(PolicyError):
+   cosy.load_vocab('/bad/path.fs')
+
+
+def test_redefine_word_calls_backup():
+ with patch('rook.cosy.backup_before_edit') as mock_backup:
+  cosy = CoSyProcess()
+  cosy.send = MagicMock()
+  cosy.redefine_word('myword', ': myword 42 ;')
+  mock_backup.assert_called_once()
+  cosy.send.assert_called_once_with(': myword 42 ;')