0dba6b7672
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
298 lines
15 KiB
Python
298 lines
15 KiB
Python
import pytest
|
||
|
||
from jira_hub import db, mirror, queue, settings
|
||
from jira_hub.commands import CommandError
|
||
from jira_hub.queue import ClaimedError
|
||
|
||
LT = {"связанные": "ASUPR_Связанные"}
|
||
|
||
|
||
def fresh():
|
||
conn = db.connect(":memory:")
|
||
settings.ensure_defaults(conn)
|
||
settings.set_(conn, "link_types", LT)
|
||
return conn
|
||
|
||
|
||
def fields(key, **extra):
|
||
return {"type": "ensure_fields", "params": {"key": key, "duedate": "2026-10-23", **extra}}
|
||
|
||
|
||
def test_create_batch_adds_tail_refresh_and_estimate():
|
||
conn = fresh()
|
||
b = queue.create_batch(conn, "срок", "claude", [fields("A-1"), {"type": "ensure_comment", "params": {"key": "A-2", "marker": "m", "body": "b"}}])
|
||
view = queue.batch_view(conn, b["id"])
|
||
assert [c["type"] for c in view["commands"]] == ["ensure_fields", "ensure_comment", "refresh"]
|
||
assert view["commands"][2]["params"] == {"keys": ["A-1", "A-2"]}
|
||
assert b["estimate_seconds"] == 120 + 30 + 2 and b["count"] == 3
|
||
assert view["status"] == "queued" and view["author"] == "claude"
|
||
|
||
|
||
def test_refresh_only_batch_has_no_tail():
|
||
conn = fresh()
|
||
b = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
|
||
assert b["count"] == 1
|
||
with pytest.raises(CommandError):
|
||
queue.create_batch(conn, "x", "claude", [{"type": "nope"}])
|
||
|
||
|
||
def test_claim_order_pause_and_report():
|
||
conn = fresh()
|
||
b1 = queue.create_batch(conn, "первый", "claude", [fields("A-1")])
|
||
b2 = queue.create_batch(conn, "второй", "claude", [fields("A-2")])
|
||
c = queue.claim(conn, agent_version=1, settings_version=1)
|
||
assert c["type"] == "ensure_fields" and c["params"]["key"] == "A-1" and c["pause_seconds"] == 120 and c["batch_id"] == b1["id"]
|
||
assert queue.claim(conn) is None # одна команда в полёте
|
||
assert queue.agent_view(conn)["current_command_id"] == c["id"]
|
||
done = queue.report_result(conn, c["id"], "done", result={"changed": ["duedate"]}, jira_calls=2)
|
||
assert done["status"] == "done" and done["jira_calls"] == 2
|
||
again = queue.report_result(conn, c["id"], "failed", error="поздно")
|
||
assert again["status"] == "done" # повторный отчёт не меняет
|
||
tail = queue.claim(conn)
|
||
assert tail["type"] == "refresh" and tail["pause_seconds"] == 2
|
||
assert conn.execute("SELECT kind, query FROM sync_run WHERE command_id = ?", (tail["id"],)).fetchone()["query"] == "A-1"
|
||
queue.record_state(conn, tail["id"], 1, 0)
|
||
queue.report_result(conn, tail["id"], "done", jira_calls=1)
|
||
run = conn.execute("SELECT * FROM sync_run WHERE command_id = ?", (tail["id"],)).fetchone()
|
||
assert run["finished_at"] and run["issues_count"] == 1 and run["jira_calls"] == 1
|
||
assert queue.batch_view(conn, b1["id"])["status"] == "done"
|
||
assert queue.claim(conn)["params"]["key"] == "A-2" and queue.batch_view(conn, b2["id"])["status"] == "running"
|
||
|
||
|
||
def test_claim_is_exclusive(tmp_path):
|
||
import threading
|
||
path = tmp_path / "hub.sqlite"
|
||
setup = db.connect(path)
|
||
settings.ensure_defaults(setup)
|
||
queue.create_batch(setup, "x", "claude", [fields("A-1")])
|
||
conns = [db.connect(path), db.connect(path)]
|
||
barrier, results = threading.Barrier(2), []
|
||
|
||
def worker(c):
|
||
barrier.wait()
|
||
results.append(queue.claim(c))
|
||
|
||
threads = [threading.Thread(target=worker, args=(c,)) for c in conns]
|
||
[t.start() for t in threads]
|
||
[t.join() for t in threads]
|
||
assert sorted(r is None for r in results) == [False, True]
|
||
assert setup.execute("SELECT COUNT(*) AS n FROM command WHERE status = 'claimed'").fetchone()["n"] == 1
|
||
|
||
|
||
def test_backup_failure_does_not_block_claim(caplog):
|
||
conn = fresh()
|
||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||
|
||
def broken(label):
|
||
raise OSError("диск полон")
|
||
|
||
c = queue.claim(conn, backup=broken)
|
||
assert c is not None and c["params"]["key"] == "A-1"
|
||
assert queue.get_command(conn, c["id"])["status"] == "claimed"
|
||
assert "не создана" in caplog.text
|
||
|
||
|
||
def test_stale_claimed_does_not_block():
|
||
conn = fresh()
|
||
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||
c = queue.claim(conn)
|
||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
|
||
nxt = queue.claim(conn)
|
||
assert nxt["params"]["key"] == "A-2"
|
||
assert queue.agent_view(conn)["stuck"][0]["id"] == c["id"]
|
||
|
||
|
||
def test_temp_key_substitution_and_skip():
|
||
conn = fresh()
|
||
mirror.set_note(conn, "new:60", {"fact_status": "В работе"}, "import")
|
||
queue.create_batch(conn, "создать", "claude", [
|
||
{"type": "ensure_issue", "params": {"temp_key": "new:60", "issue_type": "task", "summary": "Новая"}},
|
||
{"type": "ensure_link", "params": {"source": "new:60", "target": "A-1", "link_type": "связанные"}},
|
||
{"type": "ensure_issue", "params": {"temp_key": "new:61", "issue_type": "task", "summary": "Вторая"}},
|
||
{"type": "ensure_sprint", "params": {"key": "new:61", "sprint": "Спринт 2"}},
|
||
])
|
||
c1 = queue.claim(conn)
|
||
queue.report_result(conn, c1["id"], "done", result={"key": "A-90"})
|
||
assert mirror.get_note(conn, "A-90")["fact_status"] == "В работе" and mirror.get_note(conn, "new:60") is None
|
||
c2 = queue.claim(conn)
|
||
assert c2["type"] == "ensure_link" and c2["params"]["source"] == "A-90"
|
||
queue.report_result(conn, c2["id"], "done")
|
||
c3 = queue.claim(conn)
|
||
queue.report_result(conn, c3["id"], "failed", error="HTTP 400")
|
||
tail = queue.claim(conn) # ensure_sprint пропущена, refresh без new:61
|
||
assert tail["type"] == "refresh" and tail["params"]["keys"] == ["A-90", "A-1"]
|
||
view = queue.batch_view(conn, c1["batch_id"])
|
||
assert [c["status"] for c in view["commands"]][:4] == ["done", "done", "failed", "skipped"]
|
||
assert view["commands"][3]["error"] == "задача new:61 не создана"
|
||
|
||
|
||
def test_cancel_and_retry():
|
||
conn = fresh()
|
||
b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||
c = queue.claim(conn)
|
||
with pytest.raises(ClaimedError):
|
||
queue.cancel_command(conn, c["id"])
|
||
view = queue.cancel_batch(conn, b["id"])
|
||
assert [x["status"] for x in view["commands"]] == ["claimed", "cancelled", "cancelled"]
|
||
assert queue.claim(conn) is None
|
||
forced = queue.cancel_command(conn, c["id"], force=True)
|
||
assert forced["status"] == "cancelled"
|
||
after = queue.report_result(conn, c["id"], "done", result={"changed": []})
|
||
assert after["status"] == "cancelled" and after["result"] is None # отчёт после force-отмены не затирает
|
||
assert queue.batch_view(conn, b["id"])["status"] == "cancelled"
|
||
r = queue.retry_command(conn, c["id"], "claude")
|
||
assert queue.batch_view(conn, r["id"])["title"] == f"Повтор команды {c['id']}"
|
||
assert queue.list_batches(conn, limit=5)[0]["id"] == r["id"]
|
||
|
||
|
||
def test_report_after_force_cancel_keeps_cancelled():
|
||
conn = fresh()
|
||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||
c = queue.claim(conn)
|
||
queue.cancel_command(conn, c["id"], force=True)
|
||
assert queue.report_result(conn, c["id"], "failed", error="e")["status"] == "cancelled"
|
||
|
||
|
||
def test_reconcile_confirms_by_state():
|
||
conn = fresh()
|
||
queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}])
|
||
c = queue.claim(conn)
|
||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
|
||
b = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||
r = queue.claim(conn)
|
||
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "sprint": "Спринт 2", "labels": []}], [])
|
||
queue.report_result(conn, r["id"], "done")
|
||
assert queue.get_command(conn, c["id"])["status"] == "confirmed_by_state"
|
||
assert r["batch_id"] == c["batch_id"] # выдался хвостовой refresh первого пакета
|
||
assert queue.batch_view(conn, c["batch_id"])["status"] == "done"
|
||
assert queue.batch_view(conn, b["id"])["status"] == "queued"
|
||
|
||
|
||
def test_auto_refresh_conditions():
|
||
conn = fresh()
|
||
assert queue.maybe_auto_refresh(conn) is None # агент не на связи
|
||
queue.claim(conn, agent_version=1, settings_version=1) # отметился, очередь пуста
|
||
b = queue.maybe_auto_refresh(conn)
|
||
assert b and queue.batch_view(conn, b["id"])["title"] == "Автообновление"
|
||
assert queue.maybe_auto_refresh(conn) is None # очередь занята
|
||
c = queue.claim(conn)
|
||
queue.report_result(conn, c["id"], "done")
|
||
assert queue.maybe_auto_refresh(conn) is None # интервал не прошёл
|
||
conn.execute("UPDATE sync_run SET started_at = '2020-01-01T00:00:00+00:00'")
|
||
assert queue.maybe_auto_refresh(conn) is not None
|
||
settings.set_(conn, "auto_refresh_minutes", 0)
|
||
conn.execute("UPDATE sync_run SET started_at = '2020-01-01T00:00:00+00:00'")
|
||
assert queue.maybe_auto_refresh(conn) is None
|
||
|
||
|
||
def test_backup_called_for_first_write_only(tmp_path):
|
||
conn = fresh()
|
||
calls = []
|
||
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||
c1 = queue.claim(conn, backup=calls.append)
|
||
queue.report_result(conn, c1["id"], "done")
|
||
queue.claim(conn, backup=calls.append)
|
||
assert calls == [f"batch-{c1['batch_id']}"]
|
||
|
||
|
||
def test_no_backup_for_refresh_batch():
|
||
conn = fresh()
|
||
calls = []
|
||
queue.create_batch(conn, "x", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
|
||
queue.claim(conn, backup=calls.append)
|
||
assert calls == []
|
||
|
||
|
||
def test_make_backup_rotates(tmp_path):
|
||
from jira_hub import backup
|
||
conn = db.connect(tmp_path / "hub.sqlite")
|
||
for i in range(4):
|
||
backup.make_backup(conn, tmp_path / "backups", f"b{i}", keep=2)
|
||
names = sorted(p.name for p in (tmp_path / "backups").glob("*.sqlite"))
|
||
assert len(names) == 2 and names[-1].endswith("-b3.sqlite")
|
||
assert db.connect(tmp_path / "backups" / names[-1]).execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 1
|
||
|
||
|
||
def test_reconcile_respects_reporter_setting():
|
||
conn = fresh()
|
||
settings.set_(conn, "reporter", "robot")
|
||
queue.create_batch(conn, "x", "claude", [{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}}])
|
||
c = queue.claim(conn)
|
||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
|
||
created = db.utcnow().isoformat()
|
||
mirror.apply_state(conn, [{"key": "A-7", "summary": "Тема", "reporter": "other", "created": created, "labels": []}], [])
|
||
assert queue.reconcile(conn) == [] # чужая задача не подтверждает
|
||
mirror.apply_state(conn, [{"key": "A-8", "summary": "Тема", "reporter": "robot", "created": created, "labels": []}], [])
|
||
assert queue.reconcile(conn) == [c["id"]]
|
||
assert queue.get_command(conn, c["id"])["result"] == {"key": "A-8"}
|
||
|
||
|
||
def _claimed_ensure_issue(conn, summary="Тема"):
|
||
queue.create_batch(conn, "x", "claude", [{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": summary}}])
|
||
c = queue.claim(conn)
|
||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
|
||
return c
|
||
|
||
|
||
def test_reconcile_does_not_confirm_ensure_issue_without_reporter():
|
||
"""R-11: настройка reporter не задана — совпадение темы не подтверждает, заметка new:N не переезжает."""
|
||
conn = fresh()
|
||
c = _claimed_ensure_issue(conn)
|
||
mirror.set_note(conn, "new:1", {"work_kind": "разработка"}, "t")
|
||
created = db.utcnow().isoformat()
|
||
mirror.apply_state(conn, [{"key": "A-7", "summary": "Тема", "reporter": "other", "created": created, "labels": []}], [])
|
||
assert queue.reconcile(conn) == []
|
||
assert queue.get_command(conn, c["id"])["status"] == "claimed"
|
||
assert mirror.get_note(conn, "new:1") is not None and mirror.get_note(conn, "A-7") is None
|
||
|
||
|
||
def test_reconcile_does_not_overwrite_finished_command(monkeypatch):
|
||
"""M-1: отчёт агента/отмена между check_confirmed и UPDATE не перезаписываются."""
|
||
conn = fresh()
|
||
settings.set_(conn, "reporter", "robot")
|
||
c = _claimed_ensure_issue(conn)
|
||
mirror.set_note(conn, "new:1", {"work_kind": "разработка"}, "t")
|
||
mirror.apply_state(conn, [{"key": "A-8", "summary": "Тема", "reporter": "robot",
|
||
"created": db.utcnow().isoformat(), "labels": []}], [])
|
||
real = queue.C.check_confirmed
|
||
|
||
def racing(*a, **kw):
|
||
out = real(*a, **kw)
|
||
queue.cancel_command(conn, c["id"], force=True) # гонка: отмена после проверки
|
||
return out
|
||
|
||
monkeypatch.setattr(queue.C, "check_confirmed", racing)
|
||
assert queue.reconcile(conn) == []
|
||
assert queue.get_command(conn, c["id"])["status"] == "cancelled"
|
||
assert mirror.get_note(conn, "new:1") is not None # rename_note не выполнялся
|
||
|
||
|
||
def _two_batches_with_dependent(conn):
|
||
a = queue.create_batch(conn, "A", "claude", [
|
||
{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}},
|
||
{"type": "ensure_link", "params": {"source": "new:1", "target": "A-1", "link_type": "связанные"}}])
|
||
first = queue.claim(conn)
|
||
assert first["type"] == "ensure_issue"
|
||
b = queue.create_batch(conn, "B", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
|
||
return a, b, first
|
||
|
||
|
||
def test_stale_definer_does_not_block_other_batches():
|
||
"""R-13: зависший ensure_issue с зависимыми не держит очередь целиком."""
|
||
conn = fresh()
|
||
a, b, first = _two_batches_with_dependent(conn)
|
||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (first["id"],))
|
||
got = queue.claim(conn)
|
||
assert got is not None and got["batch_id"] == b["id"] and got["type"] == "refresh"
|
||
# остаток пакета A остаётся в очереди, не выдаётся и не пропускается
|
||
assert [c["status"] for c in queue.batch_view(conn, a["id"])["commands"]][1:] == ["queued", "queued"]
|
||
|
||
|
||
def test_fresh_definer_still_blocks_queue():
|
||
conn = fresh()
|
||
a, b, first = _two_batches_with_dependent(conn)
|
||
assert queue.claim(conn) is None # одна в полёте
|
||
conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],))
|
||
nxt = queue.claim(conn)
|
||
assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9"
|