chore: 更新公共镜像至 1.5.43

tests/test_log_analysis.py

@@ -0,0 +1,93 @@
+import io
+import json
+import unittest
+from contextlib import redirect_stdout
+from unittest import mock
+
+from scripts import oracle_skill as skill
+
+
+class FakeResponse:
+    def __init__(self, payload, ok=True, status_code=200):
+        self._payload = payload
+        self.ok = ok
+        self.status_code = status_code
+        self.text = ""
+
+    def json(self):
+        return self._payload
+
+
+class LogAnalysisSkillTests(unittest.TestCase):
+    def config(self):
+        return {
+            "access_token": "token",
+            "expires_at": "2999-01-01T00:00:00+08:00",
+            "transit_url": "http://transit.example",
+            "client_code": "AHMW",
+        }
+
+    def test_log_info_preserves_path_with_spaces(self):
+        response = FakeResponse({"success": True, "data": {"size": 12}})
+        path = r"D:\Program Files\BOS\logs\app log.txt"
+        with mock.patch.object(skill, "get_config", return_value=self.config()), mock.patch.object(
+            skill.requests, "post", return_value=response
+        ) as post:
+            result = skill.log_info(path, "AHMW")
+        self.assertTrue(result["success"])
+        self.assertEqual(post.call_args.args[0], "http://transit.example/api/log/info")
+        self.assertEqual(post.call_args.kwargs["json"]["path"], path)
+        self.assertEqual(post.call_args.kwargs["json"]["client_code"], "AHMW")
+        self.assertEqual(post.call_args.kwargs["headers"]["Authorization"], "Bearer token")
+
+    def test_log_search_forwards_filters_and_full_scope(self):
+        response = FakeResponse({"success": True, "data": {"matches": 1, "lines": []}})
+        with mock.patch.object(skill, "get_config", return_value=self.config()), mock.patch.object(
+            skill.requests, "post", return_value=response
+        ) as post:
+            result = skill.log_search(
+                r"D:\logs\app.log", ["ERROR", r"ORA-\d+"], "AHMW", regex=True,
+                case_sensitive=True, before_lines=5, after_lines=7, max_matches=20,
+                scan_scope="full", timeout=300,
+            )
+        self.assertTrue(result["success"])
+        payload = post.call_args.kwargs["json"]
+        self.assertEqual(payload["patterns"], ["ERROR", r"ORA-\d+"])
+        self.assertTrue(payload["regex"])
+        self.assertEqual(payload["scan_scope"], "full")
+        self.assertEqual(payload["timeout"], 300)
+
+    def test_log_enable_uses_authenticated_customer_endpoint(self):
+        response = FakeResponse({"success": True, "data": {"enabled": True, "persisted": True}})
+        with mock.patch.object(skill, "get_config", return_value=self.config()), mock.patch.object(
+            skill.requests, "post", return_value=response
+        ) as post:
+            result = skill.set_log_analysis_enabled(True, "AHMW")
+        self.assertTrue(result["success"])
+        self.assertEqual(post.call_args.args[0], "http://transit.example/api/log/config")
+        self.assertEqual(post.call_args.kwargs["json"]["enabled"], True)
+
+    def test_log_error_code_is_preserved(self):
+        response = FakeResponse(
+            {"success": False, "error": "log file not found", "error_code": "file_not_found"},
+            ok=False, status_code=404,
+        )
+        with mock.patch.object(skill, "get_config", return_value=self.config()), mock.patch.object(
+            skill.requests, "post", return_value=response
+        ):
+            result = skill.log_info(r"D:\missing.log", "AHMW")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["error_code"], "file_not_found")
+        self.assertEqual(result["status_code"], 404)
+
+    def test_capabilities_lists_all_log_commands(self):
+        output = io.StringIO()
+        with redirect_stdout(output):
+            skill.cmd_capabilities(type("Args", (), {"json": True})())
+        payload = json.loads(output.getvalue())
+        names = {item["name"] for item in payload["commands"]}
+        self.assertTrue({"log_info", "log_tail", "log_search", "log_enable", "log_disable"}.issubset(names))
+
+
+if __name__ == "__main__":
+    unittest.main()