add skill activity stream and job runtime foundation (v1.1.0)
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 17s
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>
This commit is contained in:
72
tests/test_sibling_bridge.py
Normal file
72
tests/test_sibling_bridge.py
Normal file
@@ -0,0 +1,72 @@
|
||||
# -*- 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()
|
||||
Reference in New Issue
Block a user