acdd52742e
cmd_check проверяет хаб и Eva (код 0 или 3), флаг run --once, scripts/e2e-hub.sh поднимает тестовый хаб во временном JIRA_HUB_HOME. Решения 33–35 в design.md. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
import contextlib
|
||
import io
|
||
import unittest
|
||
|
||
from _common import SETTINGS, agent, eva_cfg
|
||
from fake_eva import FakeEva
|
||
from fake_hub import FakeHub
|
||
|
||
|
||
class CheckTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.eva = FakeEva()
|
||
self.hub = FakeHub(SETTINGS)
|
||
self.addCleanup(self.eva.stop)
|
||
self.addCleanup(self.hub.stop)
|
||
self.eva.add_doc(1, "DOC-1", "К", project="P")
|
||
|
||
def check(self, **over):
|
||
cfg = eva_cfg(self.eva.base_url, hub_url=self.hub.base_url, hub_token=self.hub.token, **over)
|
||
out = io.StringIO()
|
||
with contextlib.redirect_stdout(out):
|
||
code = agent.cmd_check(cfg)
|
||
return code, out.getvalue()
|
||
|
||
def test_all_ok(self):
|
||
code, out = self.check()
|
||
self.assertEqual(code, 0, out)
|
||
self.assertIn("хаб: ", out)
|
||
self.assertIn("eva: ok", out)
|
||
call = self.eva.api_calls[-1]["body"]
|
||
self.assertEqual(call["method"], "CmfDocument.list")
|
||
self.assertEqual(call["kwargs"], {"filter": [], "fields": ["code"], "slice": [0, 1]})
|
||
|
||
def test_eva_401(self):
|
||
self.eva.token = "другой"
|
||
code, out = self.check()
|
||
self.assertEqual(code, 3)
|
||
self.assertIn("eva: ", out)
|
||
self.assertIn("401", out)
|
||
self.assertNotIn("eva-secret", out)
|
||
|
||
def test_eva_unreachable(self):
|
||
cfg = eva_cfg("http://127.0.0.1:1", hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
out = io.StringIO()
|
||
with contextlib.redirect_stdout(out):
|
||
self.assertEqual(agent.cmd_check(cfg), 3)
|
||
self.assertIn("eva: НЕДОСТУПНА", out.getvalue())
|
||
|
||
def test_hub_down_but_eva_checked(self):
|
||
cfg = eva_cfg(self.eva.base_url) # хаб: порт 1
|
||
out = io.StringIO()
|
||
with contextlib.redirect_stdout(out):
|
||
self.assertEqual(agent.cmd_check(cfg), 3)
|
||
self.assertIn("хаб: НЕДОСТУПЕН", out.getvalue())
|
||
self.assertIn("eva: ok", out.getvalue())
|
||
|
||
def test_selftest_zero_when_ok(self):
|
||
cfg = eva_cfg(self.eva.base_url, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
with contextlib.redirect_stdout(io.StringIO()):
|
||
self.assertEqual(agent.cmd_selftest(cfg), 0)
|
||
|
||
def test_run_once_exits_on_idle(self):
|
||
import json
|
||
import os
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as d:
|
||
cfg = eva_cfg(self.eva.base_url, hub_url=self.hub.base_url, hub_token=self.hub.token, auto_update=False)
|
||
path = os.path.join(d, "agent.json")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(cfg, f)
|
||
with contextlib.redirect_stdout(io.StringIO()):
|
||
code = agent.main(["--config", path, "--journal", os.path.join(d, "j.jsonl"), "run", "--once"])
|
||
self.assertEqual(code, 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|