94 lines
4.0 KiB
Python
94 lines
4.0 KiB
Python
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()
|