import base64 import io import json import os import tempfile import time import unittest from contextlib import redirect_stdout from datetime import datetime, timezone from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_jira import FakeJira from fake_hub import FakeHub from fake_relay import FakeRelay class FolderTransportTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = os.path.join(self.tmp.name, "My Shared Files", "jira-bridge") # пробелы в пути os.makedirs(self.root) self.hub = FakeHub(SETTINGS) self.relay = None def tearDown(self): if self.relay: self.relay.stop() self.hub.stop() self.tmp.cleanup() def client(self, timeout=5): cfg = {"hub_url": "folder:" + self.root, "hub_token": self.hub.token, "timeout": timeout} return agent.HubClient(cfg) def run_relay(self, mode="forward", delay=0.0): self.relay = FakeRelay(self.root, self.hub.base_url, mode, delay).start() return self.relay def listing(self, sub): return sorted(os.listdir(os.path.join(self.root, sub))) def test_settings_through_folder(self): self.run_relay() hub = self.client() self.assertIsNotNone(hub.folder) self.assertEqual(hub.settings()["project_key"], "AURORA") self.assertEqual(self.listing("resp"), []) self.assertEqual(self.listing("req"), []) req = self.relay.requests[0] self.assertEqual(req["method"], "GET") self.assertEqual(req["path"], "/agent/settings") self.assertEqual(req["authorization"], "Bearer " + self.hub.token) self.assertEqual(len(req["id"]), 32) def test_error_status_raises_http_error(self): self.run_relay() with self.assertRaises(agent.HttpError) as e: self.client().result(999, {"status": "done"}) self.assertEqual(e.exception.code, 404) def test_claim_deadline_at_least_45_seconds(self): self.run_relay() before = time.time() self.client(timeout=5).claim(2, 1, None) deadline = datetime.fromisoformat(self.relay.requests[0]["deadline"]) self.assertIsNotNone(deadline.tzinfo) self.assertGreaterEqual(deadline.timestamp() - before, 44.5) def test_no_answer_gives_connection_error_and_removes_request(self): hub = self.client(timeout=0.5) t0 = time.monotonic() with self.assertRaises(ConnectionError) as e: hub.settings() self.assertLess(time.monotonic() - t0, 2) self.assertIn("не ответил через папку", str(e.exception)) self.assertEqual([n for n in self.listing("req") if not n.startswith(".")], []) def test_missing_folder_gives_connection_error(self): cfg = {"hub_url": "folder:" + os.path.join(self.tmp.name, "нет"), "hub_token": "t", "timeout": 5} with self.assertRaises(ConnectionError) as e: agent.HubClient(cfg).settings() self.assertIn("нет папки", str(e.exception)) def test_bad_json_response_gives_connection_error(self): self.run_relay("garbage") hub = self.client() hub.folder.BAD_JSON_WAIT = 0.3 with self.assertRaises(ConnectionError) as e: hub.settings() self.assertIn("нечитаемый ответ", str(e.exception)) def _bad_response(self, payload): path = os.path.join(self.root, "resp-bad.json") with open(path, "w", encoding="utf-8") as f: json.dump(payload, f) ft = agent.FolderTransport(self.root, "Bearer x") ft.BAD_JSON_WAIT = 0.3 return ft, path def test_wrong_status_type_is_unreadable(self): for payload in ({"id": "x", "status": "ok", "body": "{}"}, {"id": "x", "status": True, "body": "{}"}, {"id": "x", "status": 200, "body": 5}): ft, path = self._bad_response(payload) with self.assertRaises(ConnectionError) as e: ft._take(path) self.assertIn("нечитаемый ответ", str(e.exception)) def test_no_answer_message_has_no_hub_prefix(self): hub = self.client(timeout=0.3) with self.assertRaises(ConnectionError) as e: hub.settings() self.assertFalse(str(e.exception).startswith("хаб")) # цикл сам пишет «хаб: …» (O-3 agent#8) def test_late_response_is_not_reused(self): self.run_relay("late", delay=1.0) with self.assertRaises(ConnectionError): self.client(timeout=0.3).settings() time.sleep(1.3) # опоздавший ответ лёг в resp/ late = self.listing("resp") self.assertEqual(len(late), 1) self.relay.mode = "forward" self.assertEqual(self.client().settings()["project_key"], "AURORA") self.assertEqual(self.listing("resp"), late) # чужой (опоздавший) ответ не тронут def test_foreign_files_untouched(self): resp_dir = os.path.join(self.root, "resp") os.makedirs(resp_dir, exist_ok=True) foreign = os.path.join(resp_dir, "0" * 32 + ".json") with open(foreign, "w", encoding="utf-8") as f: json.dump({"id": "0" * 32, "status": 200, "body": "{}"}, f) self.run_relay() self.client().settings() self.assertTrue(os.path.exists(foreign)) def test_http_url_keeps_http_transport(self): cfg = {"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5} hub = agent.HubClient(cfg) self.assertIsNone(hub.folder) self.assertEqual(hub.settings()["project_key"], "AURORA") def test_file_through_folder(self): self.run_relay() data = os.urandom(300 * 1024) sha = self.hub.put_file(data) got = self.client().file(sha) self.assertEqual((got["file"], got["size"]), (sha, len(data))) self.assertEqual(base64.b64decode(got["content"]), data) self.assertEqual(self.relay.requests[0]["path"], "/agent/files/" + sha) self.assertEqual(self.relay.requests[0]["method"], "GET") def test_file_missing_through_folder(self): self.run_relay() with self.assertRaises(agent.HttpError) as e: self.client().file("0" * 64) self.assertEqual(e.exception.code, 404) self.assertIn("no_file", e.exception.text) def test_file_over_http(self): sha = self.hub.put_file(b"abc") hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5}) got = hub.file(sha) self.assertEqual(base64.b64decode(got["content"]), b"abc") self.assertEqual(self.hub.file_requests, 1) class FolderIntegrationTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = os.path.join(self.tmp.name, "My Shared Files", "jira-bridge") os.makedirs(self.root) self.fake = FakeJira() self.hub = FakeHub(SETTINGS) self.relay = FakeRelay(self.root, self.hub.base_url).start() def tearDown(self): self.relay.stop() self.fake.stop() self.hub.stop() self.tmp.cleanup() def cfg(self): return jira_cfg(self.fake, hub_url="folder:" + self.root, hub_token=self.hub.token) def test_version(self): self.assertEqual(agent.AGENT_VERSION, 14) def test_runner_through_folder(self): self.fake.add("AURORA-1", duedate="2026-10-20") cid = self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) cfg = self.cfg() journal = agent.Journal(os.path.join(self.tmp.name, "journal.jsonl")) runner = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), journal) runner.sleep = lambda s: None runner.jira.sleep = no_sleep runner.idle_limit = 1 runner.loop() self.assertEqual(self.hub.results[cid]["status"], "done") self.assertEqual(self.hub.results[cid]["agent_version"], agent.AGENT_VERSION) self.assertEqual(self.fake.issues["AURORA-1"]["duedate"], "2026-10-23") self.assertIn(("acked", cid), [(r["kind"], r["command_id"]) for r in journal.records()]) def test_check_reports_folder(self): out = io.StringIO() with redirect_stdout(out): code = agent.cmd_check(self.cfg()) self.assertEqual(code, 0) self.assertIn("хаб: доступен через папку %s, настройки v1, проект AURORA" % self.root, out.getvalue()) def test_check_folder_silent_relay(self): self.relay.stop() self.relay = FakeRelay(self.root, self.hub.base_url, "silent").start() out = io.StringIO() t0 = time.monotonic() with redirect_stdout(out): code = agent.cmd_check(self.cfg()) self.assertEqual(code, 1) self.assertLess(time.monotonic() - t0, 13) self.assertIn("хаб: НЕДОСТУПЕН", out.getvalue()) self.assertIn("не ответил через папку", out.getvalue()) def test_relative_folder_url_rejected(self): path = os.path.join(self.tmp.name, "agent.json") for url in ("folder:relative/path", "folder:"): cfg = self.cfg() cfg["hub_url"] = url with open(path, "w", encoding="utf-8") as f: json.dump(cfg, f) with self.assertRaises(SystemExit) as e: agent.load_config(path) self.assertIn("абсолютный путь", str(e.exception.code)) if __name__ == "__main__": unittest.main()