import json import re import threading from types import SimpleNamespace import pytest from jira_hub import api, mcp_server, settings from jira_hub.client import HubClient @pytest.fixture def live(tmp_path, monkeypatch): h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups") h.claim_wait = 0 conn = h.conn() tok = settings.tokens(conn) conn.close() srv = api.make_server(h, "127.0.0.1", 0) threading.Thread(target=srv.serve_forever, daemon=True).start() base = f"http://127.0.0.1:{srv.server_port}" monkeypatch.setattr(mcp_server, "_client", lambda: HubClient(base, tok["client"])) monkeypatch.setattr(mcp_server, "POLL_SECONDS", 0.05) yield {"agent": HubClient(base, tok["agent"])} srv.shutdown() def test_offline_message(monkeypatch): monkeypatch.setattr(mcp_server, "_client", lambda: HubClient("http://127.0.0.1:9", "t", timeout=1)) assert mcp_server.issues() == mcp_server.OFFLINE assert mcp_server.agent_status() == mcp_server.OFFLINE def test_tools_roundtrip(live): live["agent"].post("/agent/state", {"issues": [{"key": "A-1", "summary": "Тема раз", "status": "Готово", "duedate": "2026-10-23", "assignee": "ivanov", "sprint": "Спринт 2", "labels": ["AWPCA"]}], "links": []}) out = mcp_server.issues(sprint="Спринт 2") assert "A-1" in out and "Тема раз" in out and "ivanov" in out assert "В работе" in mcp_server.notes_set("A-1", fact_status="В работе") detail = mcp_server.issue("A-1") assert "fact_status: В работе" in detail and "Готово" in detail created = mcp_server.batch_create("срок", [{"type": "ensure_fields", "params": {"key": "A-1", "duedate": "2026-10-24"}}]) batch_id = int(created.split("#")[1].split()[0]) assert "2 команд" in created and "мин" in created bad = mcp_server.batch_create("x", [{"type": "ensure_fields", "params": {"key": "A-1"}}]) assert bad.startswith("ошибка хаба 422") cmd = live["agent"].post("/agent/claim", {})["command"] live["agent"].post(f"/agent/commands/{cmd['id']}/result", {"status": "failed", "error": "HTTP 400 поле"}) tail = live["agent"].post("/agent/claim", {})["command"] live["agent"].post(f"/agent/commands/{tail['id']}/result", {"status": "done"}) waited = mcp_server.batch_wait(batch_id, timeout_minutes=1) assert "done" in waited and "failed" in waited and "HTTP 400" in waited assert f"#{batch_id}" in mcp_server.batches(limit=5) assert "Повтор команды" in mcp_server.command_retry(cmd["id"]) assert "cancelled" in mcp_server.batch_cancel(batch_id + 1) r = mcp_server.refresh(full=True, wait=False) assert re.fullmatch(r"пакет #\d+ «Обновление» поставлен", r) assert "онлайн" in mcp_server.agent_status() assert "120" in mcp_server.settings_get("pacing") assert "settings_version" in mcp_server.settings_set("pacing", json.dumps({"refresh": 1})) assert mcp_server.settings_set("tokens", '"x"').startswith("ошибка хаба 422") def test_formatters(): assert mcp_server.fmt_issues([]) == "задач нет" text = mcp_server.fmt_issues([{"key": "A-1", "summary": "x" * 100, "status": "Готово", "duedate": None, "assignee": None, "sprint": None, "fact_status": None}]) assert "x" * 79 + "…" in text and "x" * 80 not in text and "—" in text assert "очередь" in mcp_server.fmt_agent({"online": False, "agent_version": None, "current_command_id": None, "rate_limited_24h": 0, "stuck": [], "warning": None, "last_seen_at": None, "settings_version": None}) def test_settings_set_bad_json(live): assert mcp_server.settings_set("pacing", "{не json").startswith("значение не JSON") def _batch(status): return {"id": 7, "title": "т", "author": "claude", "created_at": "2026-10-11", "status": status, "remaining_seconds": 0, "counts": {status: 1}, "commands": []} class FakeClient: def __init__(self, statuses, runs=None): self.statuses, self.calls, self.runs = list(statuses), 0, runs def get(self, path, **params): if path == "/sync-runs": return {"runs": self.runs} self.calls += 1 return _batch(self.statuses.pop(0) if len(self.statuses) > 1 else self.statuses[0]) def _fake_time(monkeypatch, step=1.0): clock = {"now": 0.0, "sleeps": 0, "deadlines": []} def monotonic(): clock["now"] += step return clock["now"] def sleep(_): clock["sleeps"] += 1 monkeypatch.setattr(mcp_server, "time", SimpleNamespace(monotonic=monotonic, sleep=sleep)) return clock def test_batch_wait_polls_until_done(monkeypatch): fake = FakeClient(["running", "running", "done"]) monkeypatch.setattr(mcp_server, "_client", lambda: fake) clock = _fake_time(monkeypatch) out = mcp_server.batch_wait(7, timeout_minutes=10) assert out == mcp_server.fmt_batch(_batch("done")) and "таймаут" not in out assert fake.calls == 3 and clock["sleeps"] == 2 def test_batch_wait_timeout(monkeypatch): fake = FakeClient(["running"]) monkeypatch.setattr(mcp_server, "_client", lambda: fake) clock = _fake_time(monkeypatch, step=30.0) # 1 минута = 2 шага монотонных часов out = mcp_server.batch_wait(7, timeout_minutes=1) assert "таймаут ожидания" in out assert 1 < fake.calls < 10 and clock["sleeps"] == fake.calls - 1 @pytest.mark.parametrize("asked,minutes", [(0, 1), (1000, 60)]) def test_batch_wait_clamps_timeout(monkeypatch, asked, minutes): fake = FakeClient(["running"]) monkeypatch.setattr(mcp_server, "_client", lambda: fake) clock = _fake_time(monkeypatch, step=60.0) # каждый вызов часов = +1 минута mcp_server.batch_wait(7, timeout_minutes=asked) # первый вызов часов (1 мин) задаёт старт; дедлайн = старт + minutes; выход, когда now >= дедлайна assert fake.calls == minutes def test_client_created_once_per_wait(monkeypatch): fake, made = FakeClient(["running", "running", "done"]), [] monkeypatch.setattr(mcp_server, "_client", lambda: made.append(1) or fake) _fake_time(monkeypatch) mcp_server.batch_wait(7) assert len(made) == 1 def test_sync_log_content(monkeypatch): runs = [{"id": 3, "kind": "full", "query": "project = AWPCA", "started_at": "2026-10-11T10:00", "finished_at": None, "issues_count": 12, "links_count": 5, "jira_calls": 4, "rate_limited": 1}] monkeypatch.setattr(mcp_server, "_client", lambda: FakeClient(["done"], runs)) assert mcp_server.sync_log() == "#3 full project = AWPCA: 2026-10-11T10:00 → —, задач 12, связей 5, вызовов 4, 429: 1"