feat(hub): очередь команд — пакеты, выдача, отчёты, отмена, сверка, автообновление, резервные копии
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
This commit is contained in:
Andrey Epifancev
parent
f353d90ec6
commit
b41ab0ad25
15
src/jira_hub/backup.py
Normal file
15
src/jira_hub/backup.py
Normal file
@ -0,0 +1,15 @@
|
||||
"""Резервная копия файла базы перед пакетом с записями в Jira: VACUUM INTO даёт согласованный снимок при WAL."""
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def make_backup(conn, backups_dir: Path, label: str, keep: int = 30) -> Path:
|
||||
backups_dir = Path(backups_dir)
|
||||
backups_dir.mkdir(parents=True, exist_ok=True)
|
||||
target = backups_dir / f"{datetime.now():%Y%m%d-%H%M%S-%f}-{label}.sqlite"
|
||||
conn.execute("VACUUM INTO ?", (str(target),))
|
||||
files = sorted(backups_dir.glob("*.sqlite"))
|
||||
if keep > 0:
|
||||
for old in files[:-keep]:
|
||||
old.unlink()
|
||||
return target
|
||||
276
src/jira_hub/queue.py
Normal file
276
src/jira_hub/queue.py
Normal file
@ -0,0 +1,276 @@
|
||||
"""Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом, автообновление.
|
||||
Хаб никогда не повторяет команды сам (design.md, решение 4)."""
|
||||
import json
|
||||
from datetime import timedelta
|
||||
|
||||
from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim)
|
||||
from . import commands as C
|
||||
from . import db, mirror, settings
|
||||
|
||||
TERMINAL = {"done", "failed", "skipped", "cancelled", "confirmed_by_state"}
|
||||
STALE_MINUTES = 10 # взятая команда старше — не блокирует выдачу, показывается как «нет ответа»
|
||||
ONLINE_MINUTES = 5
|
||||
|
||||
|
||||
class ClaimedError(Exception):
|
||||
def __init__(self, claimed_at):
|
||||
super().__init__(f"команда взята агентом {claimed_at}; отмена — с force")
|
||||
self.claimed_at = claimed_at
|
||||
|
||||
|
||||
def _cmd(row) -> dict:
|
||||
d = dict(row)
|
||||
d["params"] = json.loads(d["params"])
|
||||
d["result"] = json.loads(d["result"]) if d.get("result") else None
|
||||
return d
|
||||
|
||||
|
||||
def get_command(conn, command_id) -> dict | None:
|
||||
row = conn.execute("SELECT * FROM command WHERE id = ?", (command_id,)).fetchone()
|
||||
return _cmd(row) if row else None
|
||||
|
||||
|
||||
def _dump(value) -> str:
|
||||
return json.dumps(value, ensure_ascii=False)
|
||||
|
||||
|
||||
# --- пакеты -----------------------------------------------------------------
|
||||
|
||||
def create_batch(conn, title, author, cmds) -> dict:
|
||||
cmds = C.validate(cmds)
|
||||
keys = []
|
||||
for c in cmds:
|
||||
keys += [k for k in C.referenced_keys(c) if k not in keys]
|
||||
if any(c["type"] != "refresh" for c in cmds) and keys:
|
||||
cmds.append({"type": "refresh", "params": {"keys": keys}})
|
||||
est = C.estimate([c["type"] for c in cmds], settings.pacing(conn))
|
||||
with db.transaction(conn):
|
||||
cur = conn.execute("INSERT INTO batch(title, author, created_at, estimate_seconds) VALUES (?, ?, ?, ?)",
|
||||
(title, author or "api", db.now(), est))
|
||||
batch_id = cur.lastrowid
|
||||
defined = {}
|
||||
for seq, c in enumerate(cmds, 1):
|
||||
deps = [defined[k] for k in C.temp_keys_used(c) if k in defined]
|
||||
cur = conn.execute("INSERT INTO command(batch_id, seq, type, params, depends_on) VALUES (?, ?, ?, ?, ?)",
|
||||
(batch_id, seq, c["type"], _dump(c["params"]), deps[0] if deps else None))
|
||||
if c["type"] == "ensure_issue":
|
||||
defined[c["params"]["temp_key"]] = cur.lastrowid
|
||||
return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)}
|
||||
|
||||
|
||||
def _status_of(cmds, cancelled_at=None) -> str:
|
||||
sts = [c["status"] for c in cmds]
|
||||
if not sts:
|
||||
return "done"
|
||||
if all(s in TERMINAL for s in sts):
|
||||
return "cancelled" if cancelled_at and all(s == "cancelled" for s in sts) else "done"
|
||||
if any(s == "claimed" for s in sts) or any(s in TERMINAL for s in sts):
|
||||
return "running"
|
||||
return "queued"
|
||||
|
||||
|
||||
def batch_view(conn, batch_id, with_commands=True) -> dict | None:
|
||||
b = conn.execute("SELECT * FROM batch WHERE id = ?", (batch_id,)).fetchone()
|
||||
if b is None:
|
||||
return None
|
||||
cmds = [_cmd(r) for r in conn.execute("SELECT * FROM command WHERE batch_id = ? ORDER BY seq", (batch_id,))]
|
||||
pacing = settings.pacing(conn)
|
||||
d = dict(b)
|
||||
d["status"] = _status_of(cmds, b["cancelled_at"])
|
||||
d["counts"] = {s: sum(1 for c in cmds if c["status"] == s) for s in sorted({c["status"] for c in cmds})}
|
||||
d["remaining_seconds"] = C.estimate([c["type"] for c in cmds if c["status"] in ("queued", "claimed")], pacing)
|
||||
if with_commands:
|
||||
d["commands"] = cmds
|
||||
return d
|
||||
|
||||
|
||||
def list_batches(conn, limit=20) -> list[dict]:
|
||||
ids = [r["id"] for r in conn.execute("SELECT id FROM batch ORDER BY id DESC LIMIT ?", (int(limit),))]
|
||||
return [batch_view(conn, i, with_commands=False) for i in ids]
|
||||
|
||||
|
||||
# --- выдача -----------------------------------------------------------------
|
||||
|
||||
def _mapping(conn, batch_id) -> dict:
|
||||
"""temp_key → настоящий ключ по завершённым ensure_issue пакета."""
|
||||
rows = conn.execute("SELECT json_extract(params, '$.temp_key') AS t, json_extract(result, '$.key') AS k FROM command "
|
||||
"WHERE batch_id = ? AND type = 'ensure_issue' AND status IN ('done', 'confirmed_by_state')",
|
||||
(batch_id,)).fetchall()
|
||||
return {r["t"]: r["k"] for r in rows if r["k"]}
|
||||
|
||||
|
||||
def _definer_status(conn, batch_id, temp_key):
|
||||
r = conn.execute("SELECT status FROM command WHERE batch_id = ? AND type = 'ensure_issue' "
|
||||
"AND json_extract(params, '$.temp_key') = ?", (batch_id, temp_key)).fetchone()
|
||||
return r["status"] if r else "failed"
|
||||
|
||||
|
||||
def _next(conn) -> dict | None:
|
||||
rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id "
|
||||
"WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall()
|
||||
for row in rows:
|
||||
cmd = _cmd(row)
|
||||
needed = C.temp_keys_used(cmd)
|
||||
if needed:
|
||||
mapping = _mapping(conn, cmd["batch_id"])
|
||||
unresolved = [k for k in needed if k not in mapping]
|
||||
pending = [k for k in unresolved if _definer_status(conn, cmd["batch_id"], k) not in TERMINAL]
|
||||
if pending:
|
||||
return None # создание ещё не завершено — ждём
|
||||
cmd["params"] = C.substitute(cmd["type"], cmd["params"], mapping)
|
||||
if cmd["type"] == "refresh":
|
||||
cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not C.is_temp(k)]
|
||||
unresolved = [] if cmd["params"]["keys"] else unresolved
|
||||
if unresolved:
|
||||
conn.execute("UPDATE command SET status = 'skipped', finished_at = ?, error = ? WHERE id = ?",
|
||||
(db.now(), f"задача {unresolved[0]} не создана", cmd["id"]))
|
||||
continue
|
||||
return cmd
|
||||
return None
|
||||
|
||||
|
||||
def _fresh_claimed_exists(conn) -> bool:
|
||||
cutoff = (db.utcnow() - timedelta(minutes=STALE_MINUTES)).isoformat()
|
||||
return conn.execute("SELECT 1 FROM command WHERE status = 'claimed' AND claimed_at > ? LIMIT 1", (cutoff,)).fetchone() is not None
|
||||
|
||||
|
||||
def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=None, backup=None) -> dict | None:
|
||||
now = db.now()
|
||||
to_backup = None
|
||||
with db.transaction(conn):
|
||||
conn.execute("UPDATE agent_status SET last_seen_at = ?, agent_version = COALESCE(?, agent_version), "
|
||||
"settings_version = COALESCE(?, settings_version), last_jira_call_at = COALESCE(?, last_jira_call_at) WHERE id = 1",
|
||||
(now, None if agent_version is None else str(agent_version), settings_version, last_jira_call_at))
|
||||
if _fresh_claimed_exists(conn):
|
||||
return None
|
||||
cmd = _next(conn)
|
||||
if cmd is None:
|
||||
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1")
|
||||
return None
|
||||
pause = int(settings.pacing(conn).get(cmd["type"], 0))
|
||||
first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'",
|
||||
(cmd["batch_id"],)).fetchone()["n"] == 0
|
||||
has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type != 'refresh' LIMIT 1",
|
||||
(cmd["batch_id"],)).fetchone() is not None
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, pause_seconds = ?, params = ? WHERE id = ?",
|
||||
(now, pause, _dump(cmd["params"]), cmd["id"]))
|
||||
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],))
|
||||
if cmd["type"] == "refresh":
|
||||
p = cmd["params"]
|
||||
conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)",
|
||||
(now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"]))
|
||||
if first and has_writes:
|
||||
to_backup = f"batch-{cmd['batch_id']}"
|
||||
if to_backup and backup:
|
||||
backup(to_backup)
|
||||
return {"id": cmd["id"], "type": cmd["type"], "params": cmd["params"], "pause_seconds": pause, "batch_id": cmd["batch_id"]}
|
||||
|
||||
|
||||
# --- отчёты -----------------------------------------------------------------
|
||||
|
||||
def record_state(conn, command_id, issues_count, links_count) -> None:
|
||||
conn.execute("UPDATE sync_run SET issues_count = issues_count + ?, links_count = links_count + ? WHERE command_id = ?",
|
||||
(int(issues_count), int(links_count), command_id))
|
||||
|
||||
|
||||
def report_result(conn, command_id, status, result=None, error=None, jira_calls=0, rate_limited=0) -> dict:
|
||||
if status not in ("done", "failed"):
|
||||
raise ValueError("status — done или failed")
|
||||
with db.transaction(conn):
|
||||
cmd = get_command(conn, command_id)
|
||||
if cmd is None:
|
||||
raise KeyError(command_id)
|
||||
if cmd["status"] in TERMINAL:
|
||||
return cmd # идемпотентно; cancelled не затирается
|
||||
now = db.now()
|
||||
conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?",
|
||||
(status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id))
|
||||
conn.execute("UPDATE agent_status SET current_command_id = NULL, last_seen_at = ? WHERE id = 1 AND current_command_id = ?",
|
||||
(now, command_id))
|
||||
if cmd["type"] == "ensure_issue" and status == "done" and result and result.get("key"):
|
||||
mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"])
|
||||
if cmd["type"] == "refresh":
|
||||
conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?",
|
||||
(now, int(jira_calls or 0), int(rate_limited or 0), command_id))
|
||||
if cmd["type"] == "refresh" and status == "done":
|
||||
reconcile(conn)
|
||||
return get_command(conn, command_id)
|
||||
|
||||
|
||||
def reconcile(conn) -> list[int]:
|
||||
"""Взятые команды без ответа, чей результат уже виден в зеркале → confirmed_by_state."""
|
||||
link_types = settings.get(conn, "link_types") or {}
|
||||
reporter = settings.get(conn, "reporter")
|
||||
changed = []
|
||||
for row in conn.execute("SELECT * FROM command WHERE status = 'claimed'").fetchall():
|
||||
cmd = _cmd(row)
|
||||
ok, result = C.check_confirmed(conn, cmd, link_types, reporter)
|
||||
if not ok:
|
||||
continue
|
||||
with db.transaction(conn):
|
||||
conn.execute("UPDATE command SET status = 'confirmed_by_state', finished_at = ?, result = ? WHERE id = ?",
|
||||
(db.now(), _dump(result) if result else None, cmd["id"]))
|
||||
if cmd["type"] == "ensure_issue" and result:
|
||||
mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"])
|
||||
changed.append(cmd["id"])
|
||||
return changed
|
||||
|
||||
|
||||
# --- отмена и повтор ----------------------------------------------------------
|
||||
|
||||
def cancel_batch(conn, batch_id) -> dict:
|
||||
with db.transaction(conn):
|
||||
if conn.execute("SELECT 1 FROM batch WHERE id = ?", (batch_id,)).fetchone() is None:
|
||||
raise KeyError(batch_id)
|
||||
now = db.now()
|
||||
conn.execute("UPDATE batch SET cancelled_at = ? WHERE id = ? AND cancelled_at IS NULL", (now, batch_id))
|
||||
conn.execute("UPDATE command SET status = 'cancelled', finished_at = ? WHERE batch_id = ? AND status = 'queued'", (now, batch_id))
|
||||
return batch_view(conn, batch_id)
|
||||
|
||||
|
||||
def cancel_command(conn, command_id, force=False) -> dict:
|
||||
with db.transaction(conn):
|
||||
cmd = get_command(conn, command_id)
|
||||
if cmd is None:
|
||||
raise KeyError(command_id)
|
||||
if cmd["status"] == "claimed" and not force:
|
||||
raise ClaimedError(cmd["claimed_at"])
|
||||
if cmd["status"] in ("queued", "claimed"):
|
||||
conn.execute("UPDATE command SET status = 'cancelled', finished_at = ? WHERE id = ?", (db.now(), command_id))
|
||||
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1 AND current_command_id = ?", (command_id,))
|
||||
return get_command(conn, command_id)
|
||||
|
||||
|
||||
def retry_command(conn, command_id, author) -> dict:
|
||||
cmd = get_command(conn, command_id)
|
||||
if cmd is None:
|
||||
raise KeyError(command_id)
|
||||
return create_batch(conn, f"Повтор команды {command_id}", author, [{"type": cmd["type"], "params": cmd["params"]}])
|
||||
|
||||
|
||||
# --- агент и автообновление ----------------------------------------------------
|
||||
|
||||
def agent_view(conn) -> dict:
|
||||
d = dict(conn.execute("SELECT * FROM agent_status WHERE id = 1").fetchone())
|
||||
now = db.utcnow()
|
||||
d["online"] = bool(d["last_seen_at"]) and now - db.parse(d["last_seen_at"]) < timedelta(minutes=ONLINE_MINUTES)
|
||||
min_version = int(settings.get(conn, "min_agent_version") or 0)
|
||||
d["warning"] = "обновите агента" if d["agent_version"] and int(d["agent_version"]) < min_version else None
|
||||
d["rate_limited_24h"] = conn.execute("SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ?",
|
||||
((now - timedelta(days=1)).isoformat(),)).fetchone()["n"]
|
||||
cutoff = (now - timedelta(minutes=STALE_MINUTES)).isoformat()
|
||||
d["stuck"] = [dict(r) for r in conn.execute(
|
||||
"SELECT id, type, claimed_at FROM command WHERE status = 'claimed' AND claimed_at <= ? ORDER BY id", (cutoff,))]
|
||||
return d
|
||||
|
||||
|
||||
def maybe_auto_refresh(conn) -> dict | None:
|
||||
minutes = int(settings.get(conn, "auto_refresh_minutes") or 0)
|
||||
if minutes <= 0 or not agent_view(conn)["online"]:
|
||||
return None
|
||||
if conn.execute("SELECT 1 FROM command WHERE status = 'queued' LIMIT 1").fetchone() or _fresh_claimed_exists(conn):
|
||||
return None
|
||||
last = conn.execute("SELECT MAX(started_at) AS t FROM sync_run").fetchone()["t"]
|
||||
if last and db.utcnow() - db.parse(last) < timedelta(minutes=minutes):
|
||||
return None
|
||||
return create_batch(conn, "Автообновление", "hub", [{"type": "refresh", "params": {"jql": settings.get(conn, "base_jql")}}])
|
||||
201
tests/test_queue.py
Normal file
201
tests/test_queue.py
Normal file
@ -0,0 +1,201 @@
|
||||
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():
|
||||
conn = fresh()
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||||
first, second = queue.claim(conn), queue.claim(conn)
|
||||
assert first is not None and second is None
|
||||
|
||||
|
||||
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"}
|
||||
Loading…
x
Reference in New Issue
Block a user