chendelian
1c51139883
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 17s
Introduce SAS v1 emit/finish/journal API, sibling CLI bridge, RPA video session integration, and tests for the host Job Runner activity timeline. Co-authored-by: Cursor <cursoragent@cursor.com>
73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""sibling_bridge: mock subprocess sibling CLI calls."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from jiangchang_skill_core.sibling_bridge import (
|
|
call_sibling_json,
|
|
call_sibling_json_array,
|
|
get_sibling_main_path,
|
|
)
|
|
|
|
|
|
class TestSiblingBridge(unittest.TestCase):
|
|
def test_get_sibling_main_path(self) -> None:
|
|
with patch.dict(os.environ, {"JIANGCHANG_SKILLS_ROOT": r"D:\skills"}):
|
|
path = get_sibling_main_path("account-manager")
|
|
self.assertTrue(path.replace("\\", "/").endswith("account-manager/scripts/main.py"))
|
|
|
|
@patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
|
|
def test_call_sibling_json_success(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=[],
|
|
returncode=0,
|
|
stdout='log line\n{"id": 1, "name": "test"}\n',
|
|
stderr="",
|
|
)
|
|
result = call_sibling_json("account-manager", ["list", "all"])
|
|
self.assertEqual(result, {"id": 1, "name": "test"})
|
|
cmd = mock_run.call_args[0][0]
|
|
self.assertIn("account-manager", cmd[1])
|
|
self.assertIn("list", cmd)
|
|
self.assertTrue(mock_run.call_args[1]["capture_output"])
|
|
|
|
@patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
|
|
def test_call_sibling_json_nonzero_exit(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=[], returncode=1, stdout="", stderr="ERROR:fail"
|
|
)
|
|
self.assertIsNone(call_sibling_json("content-manager", ["get", "x"]))
|
|
|
|
@patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
|
|
def test_call_sibling_json_array(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=[],
|
|
returncode=0,
|
|
stdout=json.dumps([{"a": 1}, {"b": 2}]),
|
|
stderr="",
|
|
)
|
|
result = call_sibling_json_array("account-manager", ["list", "all"])
|
|
self.assertEqual(result, [{"a": 1}, {"b": 2}])
|
|
|
|
@patch("jiangchang_skill_core.sibling_bridge.subprocess.run")
|
|
def test_uses_subprocess_env_with_trace(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=[], returncode=0, stdout='{"ok":true}', stderr=""
|
|
)
|
|
with patch(
|
|
"jiangchang_skill_core.sibling_bridge.subprocess_env_with_trace",
|
|
return_value={"JIANGCHANG_TRACE_ID": "abc123"},
|
|
) as mock_env:
|
|
call_sibling_json("api-key-vault", ["get", "key"])
|
|
self.assertEqual(mock_run.call_args[1]["env"], {"JIANGCHANG_TRACE_ID": "abc123"})
|
|
mock_env.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|