121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""Exercise mounted CLI scripts against the exact image Kanban API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
_HOME = tempfile.TemporaryDirectory(prefix="hermes-lane-compatibility-")
|
|
os.environ["HERMES_HOME"] = _HOME.name
|
|
os.environ["HERMES_CLI_LANE_CONCURRENCY"] = "1"
|
|
lane_source = Path(os.environ["HERMES_CLI_LANE_SOURCE"])
|
|
sys.path.insert(0, str(lane_source))
|
|
|
|
from cli_lane_capabilities import ( # noqa: E402
|
|
detect_kanban_capabilities,
|
|
initialize_kanban_capabilities,
|
|
runtime_health,
|
|
)
|
|
from cli_lane_config import TerminalIdentity # noqa: E402
|
|
from cli_lane_dispatch import recover_orphans # noqa: E402
|
|
from cli_lane_finalization import _finalize_document_db # noqa: E402
|
|
from hermes_cli import kanban_db # noqa: E402
|
|
|
|
|
|
class MountedLaneCompatibilityTests(unittest.TestCase):
|
|
"""Verify safe behavior before and after the image API patch."""
|
|
|
|
def setUp(self) -> None:
|
|
self.connection = kanban_db.connect()
|
|
|
|
def tearDown(self) -> None:
|
|
self.connection.close()
|
|
|
|
@staticmethod
|
|
def _document(summary: str) -> dict:
|
|
return {
|
|
"result": json.dumps(
|
|
{
|
|
"status": "completed",
|
|
"summary": summary,
|
|
"changed_files": [],
|
|
"tests_run": ["mixed source compatibility"],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
},
|
|
sort_keys=True,
|
|
),
|
|
"summary": summary,
|
|
"metadata": {"matrix": os.environ["HERMES_COMPATIBILITY_MODE"]},
|
|
}
|
|
|
|
def test_exact_image_api_matrix(self) -> None:
|
|
mode = os.environ["HERMES_COMPATIBILITY_MODE"]
|
|
capabilities = initialize_kanban_capabilities(kanban_db)
|
|
self.assertTrue(capabilities.exact_run_completion)
|
|
self.assertEqual(capabilities.ready, mode == "patched")
|
|
self.assertEqual(runtime_health()["state"], "ready" if mode == "patched" else "deferred")
|
|
self.assertLess(Path(_HOME.name, "cli-lanes/runtime-health.json").stat().st_size, 2048)
|
|
|
|
def test_current_active_run_completion_stays_safe(self) -> None:
|
|
task_id = kanban_db.create_task(self.connection, title="active accepted result")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
run_id = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
outcome = _finalize_document_db(
|
|
kanban_db,
|
|
TerminalIdentity("default", task_id, run_id, "pending"),
|
|
self._document("active result"),
|
|
)
|
|
self.assertEqual(outcome, "committed")
|
|
self.assertEqual(kanban_db.get_task(self.connection, task_id).status, "done")
|
|
|
|
def test_ended_run_replay_is_enabled_only_after_patch(self) -> None:
|
|
mode = os.environ["HERMES_COMPATIBILITY_MODE"]
|
|
task_id = kanban_db.create_task(self.connection, title="ended accepted result")
|
|
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
|
|
run_id = kanban_db.get_task(self.connection, task_id).current_run_id
|
|
self.assertTrue(
|
|
kanban_db.block_task(
|
|
self.connection,
|
|
task_id,
|
|
reason="simulate post-journal error",
|
|
expected_run_id=run_id,
|
|
)
|
|
)
|
|
outcome = _finalize_document_db(
|
|
kanban_db,
|
|
TerminalIdentity("default", task_id, run_id, "pending"),
|
|
self._document("ended result"),
|
|
)
|
|
self.assertEqual(outcome, "committed" if mode == "patched" else "deferred")
|
|
|
|
def test_old_reclaim_api_is_never_called_unguarded(self) -> None:
|
|
capabilities = detect_kanban_capabilities(kanban_db)
|
|
if capabilities.exact_run_reclaim:
|
|
self.skipTest("patched API exercises guarded reclaim in the exact-run suite")
|
|
with mock.patch.object(
|
|
kanban_db,
|
|
"reclaim_task",
|
|
side_effect=AssertionError("legacy unguarded reclaim called"),
|
|
), mock.patch("cli_lane_dispatch.recover_pending_finalizations", return_value=0):
|
|
recover_orphans()
|
|
|
|
|
|
def main() -> int:
|
|
"""Run the compatibility matrix for the selected exact image source."""
|
|
result = unittest.TextTestRunner(verbosity=2).run(
|
|
unittest.defaultTestLoader.loadTestsFromTestCase(MountedLaneCompatibilityTests)
|
|
)
|
|
return 0 if result.wasSuccessful() else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|