Очередь, релиз и статус по агенту модуля (#91)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
b8c7f20647
commit
3f4c9183b8
@ -13,9 +13,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, unquote, urlparse
|
||||
|
||||
from . import backup, db, files, projects, queue, registry, relay, release, settings
|
||||
from . import backup, db, files, http, projects, queue, registry, relay, release, settings
|
||||
from .errors import CommandError, HttpMapped
|
||||
from . import http
|
||||
from .http import ApiError, Raw, int_group, int_param, mapped, not_found
|
||||
from .paths import backups_dir as default_backups_dir
|
||||
from .paths import secure_dir
|
||||
@ -35,20 +34,21 @@ SWEEP_EVERY = 24 * 3600 # уборка files/ не чаще раза в
|
||||
|
||||
|
||||
class Hub:
|
||||
def __init__(self, db_path, backups_dir=None, release_dir=None, web_dir=None, files_dir=None):
|
||||
def __init__(self, db_path, backups_dir=None, release_root=None, web_dir=None, files_dir=None):
|
||||
self.db_path = Path(db_path)
|
||||
self.web_dir = Path(web_dir) if web_dir else WEB_APP
|
||||
self.backups_dir = Path(backups_dir) if backups_dir else default_backups_dir()
|
||||
self.release_dir = Path(release_dir) if release_dir else self.db_path.parent / "release"
|
||||
self.release_root = Path(release_root) if release_root else self.db_path.parent / "release"
|
||||
self.files_dir = Path(files_dir) if files_dir else self.db_path.parent / "files"
|
||||
self.last_sweep = None # время прошлой уборки files/, в памяти (решение 15)
|
||||
self.module_state: dict[str, dict] = {} # память модулей между тиками (имя модуля → словарь)
|
||||
# последнее из claim; в памяти (R-3). reported — был ли claim с запуска хаба: до него launcher_version
|
||||
# неизвестен, а не null (hub-hygiene-d, решение 7)
|
||||
self.agent_update = {"update_state": None, "launcher_version": None, "reported": False}
|
||||
# последнее из claim по каждому агенту; в памяти (R-3). reported — был ли claim с запуска хаба: до него
|
||||
# launcher_version неизвестен, а не null (hub-hygiene-d, решение 7)
|
||||
self.agent_update: dict[str, dict] = {m.name: self._no_update() for m in registry.agent_modules()}
|
||||
self.claim_wait = 30
|
||||
secure_dir(self.backups_dir) # каталог копий закрыт при каждом старте, а не только после первой копии
|
||||
secure_dir(self.files_dir) # files/ — 0700 при каждом старте, как backups/
|
||||
release.migrate_layout(self.release_root) # release/agent.py → release/jira/ (раскладка по модулям)
|
||||
conn = self.conn()
|
||||
settings.ensure_defaults(conn)
|
||||
conn.close()
|
||||
@ -89,10 +89,14 @@ class Hub:
|
||||
log.info("уборка файлов: удалено %d", len(removed))
|
||||
return removed
|
||||
|
||||
def note_agent_update(self, body):
|
||||
"""update_state и launcher_version из claim: хранятся в памяти, агент шлёт их в каждом claim (R-3).
|
||||
@staticmethod
|
||||
def _no_update() -> dict:
|
||||
return {"update_state": None, "launcher_version": None, "reported": False}
|
||||
|
||||
def note_agent_update(self, agent, body):
|
||||
"""update_state и launcher_version из claim агента: хранятся в памяти, агент шлёт их в каждом claim (R-3).
|
||||
Значение неверного типа игнорируется — claim из-за него не отвергается; сам claim отмечается (reported)."""
|
||||
new = dict(self.agent_update)
|
||||
new = dict(self.agent_update.get(agent) or self._no_update())
|
||||
new["reported"] = True
|
||||
us = body.get("update_state")
|
||||
if us is None or isinstance(us, dict):
|
||||
@ -102,7 +106,7 @@ class Hub:
|
||||
lv = body.get("launcher_version")
|
||||
if lv is None or (isinstance(lv, int) and not isinstance(lv, bool)):
|
||||
new["launcher_version"] = lv
|
||||
self.agent_update = new
|
||||
self.agent_update[agent] = new
|
||||
|
||||
def make_backup(self, label):
|
||||
conn = self.conn()
|
||||
@ -115,17 +119,18 @@ class Hub:
|
||||
# --- обработчики: fn(hub, conn, match, query, body) -> (status, payload) ----------
|
||||
|
||||
def h_claim(hub, conn, m, q, body):
|
||||
hub.note_agent_update(body)
|
||||
agent = http.agent_of(q)
|
||||
hub.note_agent_update(agent, body)
|
||||
deadline = time.monotonic() + hub.claim_wait
|
||||
while True:
|
||||
cmd = queue.claim(conn, body.get("agent_version"), body.get("settings_version"),
|
||||
cmd = queue.claim(conn, agent, body.get("agent_version"), body.get("settings_version"),
|
||||
body.get("last_call_at", body.get("last_jira_call_at")),
|
||||
backup=hub.make_backup, files_dir=hub.files_dir, agent=http.agent_of(q))
|
||||
backup=hub.make_backup, files_dir=hub.files_dir)
|
||||
if cmd or time.monotonic() >= deadline:
|
||||
break
|
||||
time.sleep(1)
|
||||
try: # команда уже выдана: публикация не должна превращать ответ в 500 (I-1)
|
||||
rel = release.current(hub.release_dir)
|
||||
rel = release.current(release.module_dir(hub.release_root, agent))
|
||||
except Exception: # noqa: BLE001
|
||||
log.exception("release: чтение публикации при claim")
|
||||
rel = None
|
||||
@ -134,7 +139,7 @@ def h_claim(hub, conn, m, q, body):
|
||||
|
||||
|
||||
def h_agent_release(hub, conn, m, q, body):
|
||||
rel = release.load(hub.release_dir)
|
||||
rel = release.load(release.module_dir(hub.release_root, http.agent_of(q)))
|
||||
if rel is None:
|
||||
raise ApiError(404, "no_release", "версия агента не опубликована")
|
||||
return 200, rel
|
||||
@ -288,14 +293,26 @@ def h_command_cancel(hub, conn, m, q, body):
|
||||
raise ApiError(409, "claimed", e.claimed_at)
|
||||
|
||||
|
||||
def h_agent_get(hub, conn, m, q, body):
|
||||
view = queue.agent_view(conn)
|
||||
rel = release.current(hub.release_dir)
|
||||
au = hub.agent_update
|
||||
def _agent_view(hub, conn, name) -> dict:
|
||||
view = queue.agent_view(conn, name)
|
||||
rel = release.current(release.module_dir(hub.release_root, name))
|
||||
au = hub.agent_update.get(name) or hub._no_update()
|
||||
us, lv, reported = au["update_state"], au["launcher_version"], au["reported"]
|
||||
view.update(release=rel, update_state=us, launcher_version=lv, agent_reported=reported,
|
||||
update_hint=release.hint(rel, view, us, lv, reported=reported))
|
||||
return 200, view
|
||||
return view
|
||||
|
||||
|
||||
def h_agent_get(hub, conn, m, q, body):
|
||||
"""Статус агента модуля: ?module=<имя>, по умолчанию jira; неизвестный или без агента — 404."""
|
||||
name = q.get("module") or "jira"
|
||||
if name not in [a.name for a in registry.agent_modules()]:
|
||||
raise ApiError(404, "not_found", f"module {name}")
|
||||
return 200, _agent_view(hub, conn, name)
|
||||
|
||||
|
||||
def h_agents_get(hub, conn, m, q, body):
|
||||
return 200, {"agents": [_agent_view(hub, conn, a.name) for a in registry.agent_modules()]}
|
||||
|
||||
|
||||
def _settings_view(conn) -> dict:
|
||||
@ -378,6 +395,7 @@ CORE_ROUTES = [
|
||||
("POST", r"^/commands/(\d+)/retry$", h_command_retry, "client"),
|
||||
("POST", r"^/commands/(\d+)/cancel$", h_command_cancel, "client"),
|
||||
("GET", r"^/agent$", h_agent_get, "client"),
|
||||
("GET", r"^/agents$", h_agents_get, "client"),
|
||||
("GET", r"^/settings$", h_settings_get, "client"),
|
||||
("PUT", r"^/settings$", h_settings_put, "client"),
|
||||
("GET", r"^/projects$", h_projects_get, "client"),
|
||||
|
||||
@ -26,13 +26,18 @@ def main(argv=None) -> int:
|
||||
p = sub.add_parser("agent", help="версия агента для самообновления")
|
||||
agent_sub = p.add_subparsers(dest="agent_cmd", required=True)
|
||||
p = agent_sub.add_parser("publish", help="опубликовать agent.py (версия только вверх)")
|
||||
p.add_argument("--module", default="jira", help="модуль агента (по умолчанию jira)")
|
||||
p.add_argument("path")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.cmd == "agent":
|
||||
from . import release
|
||||
names = [m.name for m in registry.agent_modules()]
|
||||
if args.module not in names:
|
||||
print(f"неизвестный модуль {args.module}; модули с агентами: {', '.join(names)}", file=sys.stderr)
|
||||
return 2
|
||||
try:
|
||||
m = release.publish(paths.release_dir(), args.path)
|
||||
m = release.publish(release.module_dir(paths.release_root(), args.module), args.path)
|
||||
except release.PublishError as e:
|
||||
print(e, file=sys.stderr)
|
||||
return 1
|
||||
|
||||
@ -18,7 +18,8 @@ def backups_dir() -> Path:
|
||||
return home() / "backups"
|
||||
|
||||
|
||||
def release_dir() -> Path:
|
||||
def release_root() -> Path:
|
||||
"""Корень публикаций агентов: release/<модуль>/agent.py + manifest.json."""
|
||||
return home() / "release"
|
||||
|
||||
|
||||
|
||||
@ -110,19 +110,28 @@ def insert_batch(conn, title, author, cmds, project=None) -> dict:
|
||||
return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)}
|
||||
|
||||
|
||||
def has_active(conn) -> bool:
|
||||
"""В очереди есть команда queued или claimed."""
|
||||
return conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone() is not None
|
||||
def _own_types(agent: str) -> list[str]:
|
||||
"""Типы команд модуля агента (имя агента = имя модуля); у модуля без команд — пусто."""
|
||||
m = registry.module(agent)
|
||||
return sorted(m.commands.types) if m and m.commands else []
|
||||
|
||||
|
||||
def _agent(agent: str | None) -> str:
|
||||
"""Имя агента; без имени — первый агент реестра (временно, пока claim не получает агента из запроса)."""
|
||||
return agent or registry.agent_modules()[0].name
|
||||
def _in(types: list[str]) -> str:
|
||||
return ",".join("?" * len(types))
|
||||
|
||||
|
||||
def agent_version(conn, agent: str | None = None):
|
||||
def has_active(conn, module: str | None = None) -> bool:
|
||||
"""В очереди есть команда queued или claimed; module — только типы этого модуля, None — любого."""
|
||||
if module is None:
|
||||
return conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone() is not None
|
||||
types = _own_types(module)
|
||||
return conn.execute(f"SELECT 1 FROM command WHERE status IN ('queued', 'claimed') AND type IN ({_in(types)}) LIMIT 1",
|
||||
types).fetchone() is not None
|
||||
|
||||
|
||||
def agent_version(conn, agent: str):
|
||||
"""Версия агента из его последней выдачи; None — агент не отчитывался."""
|
||||
row = conn.execute("SELECT agent_version FROM agent_status WHERE agent = ?", (_agent(agent),)).fetchone()
|
||||
row = conn.execute("SELECT agent_version FROM agent_status WHERE agent = ?", (agent,)).fetchone()
|
||||
return row["agent_version"] if row else None
|
||||
|
||||
|
||||
@ -181,9 +190,11 @@ def _skip_reason(conn, batch_id, definer, unresolved) -> str:
|
||||
return f"создание задачи {cancelled[0]} отменено" if cancelled else f"задача {unresolved[0]} не создана"
|
||||
|
||||
|
||||
def _next(conn) -> dict | None:
|
||||
def _next(conn, agent: str) -> dict | None:
|
||||
types = _own_types(agent)
|
||||
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()
|
||||
f"WHERE c.status = 'queued' AND b.cancelled_at IS NULL AND c.type IN ({_in(types)}) "
|
||||
"ORDER BY b.created_at, b.id, c.seq", types).fetchall()
|
||||
for row in rows:
|
||||
cmd = _cmd(row)
|
||||
h = _hooks(cmd["type"])
|
||||
@ -203,11 +214,14 @@ def _next(conn) -> dict | None:
|
||||
return None
|
||||
|
||||
|
||||
def _redeliver(conn, now) -> dict | None:
|
||||
def _redeliver(conn, now, agent: str) -> dict | None:
|
||||
"""Самая старая взятая команда без отчёта — выдать снова. Агент приходит за новой командой, только когда своей
|
||||
незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed."""
|
||||
незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed. Берутся только
|
||||
команды типов модуля агента: взятая другим агентом не довыдаётся."""
|
||||
types = _own_types(agent)
|
||||
while True:
|
||||
row = conn.execute("SELECT * FROM command WHERE status = 'claimed' ORDER BY claimed_at, id LIMIT 1").fetchone()
|
||||
row = conn.execute(f"SELECT * FROM command WHERE status = 'claimed' AND type IN ({_in(types)}) "
|
||||
"ORDER BY claimed_at, id LIMIT 1", types).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
cmd = _cmd(row)
|
||||
@ -229,16 +243,15 @@ def deliver_params(conn, cmd, files_dir) -> dict:
|
||||
return h.deliver_params(conn, cmd, files_dir) if h else cmd["params"]
|
||||
|
||||
|
||||
def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=None, backup=None, files_dir=None,
|
||||
agent: str | None = None) -> dict | None:
|
||||
def claim(conn, agent: str, agent_version=None, settings_version=None, last_call_at=None, backup=None,
|
||||
files_dir=None) -> dict | None:
|
||||
now = db.now()
|
||||
agent = _agent(agent)
|
||||
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_call_at = COALESCE(?, last_call_at) WHERE agent = ?",
|
||||
(now, None if agent_version is None else str(agent_version), settings_version, last_jira_call_at, agent))
|
||||
cmd = _redeliver(conn, now)
|
||||
(now, None if agent_version is None else str(agent_version), settings_version, last_call_at, agent))
|
||||
cmd = _redeliver(conn, now, agent)
|
||||
if cmd is not None:
|
||||
pause = _pause(conn, cmd["type"])
|
||||
conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?",
|
||||
@ -249,7 +262,7 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non
|
||||
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE agent = ?", (cmd["id"], agent))
|
||||
log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1)
|
||||
else:
|
||||
cmd = _next(conn)
|
||||
cmd = _next(conn, agent)
|
||||
if cmd is None:
|
||||
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE agent = ?", (agent,))
|
||||
return None
|
||||
@ -377,16 +390,19 @@ def retry_command(conn, command_id, author, files_dir=None, project=None) -> dic
|
||||
|
||||
# --- агент и автообновление ----------------------------------------------------
|
||||
|
||||
def agent_view(conn, agent: str | None = None) -> dict:
|
||||
agent = _agent(agent)
|
||||
def agent_view(conn, agent: str) -> dict:
|
||||
d = dict(conn.execute("SELECT * FROM agent_status WHERE agent = ?", (agent,)).fetchone())
|
||||
d["module"] = agent
|
||||
types = _own_types(agent)
|
||||
now = db.utcnow()
|
||||
d["online"] = bool(d["last_seen_at"]) and now - db.parse(d["last_seen_at"]) < timedelta(minutes=ONLINE_MINUTES)
|
||||
m = registry.module(agent)
|
||||
min_version = int(settings.get(conn, m.agent_min_version_key) or 0) if m and m.agent_min_version_key else 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"]
|
||||
d["rate_limited_24h"] = conn.execute(
|
||||
f"SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ? AND type IN ({_in(types)})",
|
||||
((now - timedelta(days=1)).isoformat(), *types)).fetchone()["n"]
|
||||
d["redelivered"] = [dict(r) for r in conn.execute(
|
||||
"SELECT id, type, claimed_at, deliveries FROM command WHERE status = 'claimed' AND deliveries > 1 ORDER BY id")]
|
||||
"SELECT id, type, claimed_at, deliveries FROM command WHERE status = 'claimed' AND deliveries > 1 "
|
||||
f"AND type IN ({_in(types)}) ORDER BY id", types)]
|
||||
return d
|
||||
|
||||
@ -26,6 +26,31 @@ class PublishError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def module_dir(root, module) -> Path:
|
||||
"""Каталог публикации агента модуля: <root>/<модуль>. Нет каталога — создаётся с правами 0700; существующий не
|
||||
трогаем (права владельца не переписываем на каждом claim)."""
|
||||
d = Path(root) / module
|
||||
if not d.is_dir():
|
||||
secure_dir(d)
|
||||
return d
|
||||
|
||||
|
||||
def migrate_layout(root) -> None:
|
||||
"""Раскладка до модулей: agent.py и manifest.json лежали прямо в root. Если root/jira/ нет — переносим их в
|
||||
root/jira/ (манифест последним: по нему версия считается опубликованной). Сбой — предупреждение в лог, хаб стартует."""
|
||||
root = Path(root)
|
||||
try:
|
||||
old = [root / n for n in ("agent.py", "manifest.json") if (root / n).exists()]
|
||||
if not old or (root / "jira").exists():
|
||||
return
|
||||
dest = module_dir(root, "jira")
|
||||
for f in old:
|
||||
os.replace(f, dest / f.name)
|
||||
log.info("release: публикация агента перенесена в %s", dest)
|
||||
except OSError as e:
|
||||
log.warning("release: раскладка %s не перенесена в jira/: %s", root, e)
|
||||
|
||||
|
||||
def parse_version(text: str) -> int:
|
||||
found = VERSION_RE.findall(text)
|
||||
if len(found) != 1:
|
||||
|
||||
@ -30,7 +30,7 @@ def request_fetch(conn, files_dir, attachment_id: str, author: str | None) -> di
|
||||
raise KeyError(attachment_id)
|
||||
if info["state"] in ("ready", "fetching"):
|
||||
return info
|
||||
v = queue.agent_version(conn)
|
||||
v = queue.agent_version(conn, "jira")
|
||||
if v is not None and str(v).isdigit() and int(v) < FETCH_AGENT_VERSION: # версия неизвестна — не блокируем
|
||||
raise AgentTooOld(f"агент v{v} не умеет скачивать вложения — нужен v{FETCH_AGENT_VERSION}")
|
||||
cmds = commands.validate([{"type": "fetch_attachment", "params": {"key": info["key"], "attachment_id": attachment_id}}])
|
||||
|
||||
@ -24,9 +24,9 @@ def maybe_auto_refresh(conn) -> dict | None:
|
||||
if jql is None:
|
||||
return None
|
||||
minutes = int(settings.get(conn, "auto_refresh_minutes") or 0)
|
||||
if minutes <= 0 or not queue.agent_view(conn)["online"]:
|
||||
if minutes <= 0 or not queue.agent_view(conn, "jira")["online"]:
|
||||
return None
|
||||
if queue.has_active(conn):
|
||||
if queue.has_active(conn, "jira"):
|
||||
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):
|
||||
|
||||
6
tests/fixtures/baseline/routes.expected.json
vendored
6
tests/fixtures/baseline/routes.expected.json
vendored
@ -71,6 +71,12 @@
|
||||
"agent",
|
||||
"h_agent_state"
|
||||
],
|
||||
[
|
||||
"GET",
|
||||
"^/agents$",
|
||||
"client",
|
||||
"h_agents_get"
|
||||
],
|
||||
[
|
||||
"GET",
|
||||
"^/app(?:/(.*))?$",
|
||||
|
||||
@ -1,15 +1,25 @@
|
||||
"""Агенты модулей: токен определяет модуль, миграция v7 сохраняет агента jira (wiki-module, решения 1–3, 6, 15)."""
|
||||
import pytest
|
||||
|
||||
from jira_hub.core import cli, db, queue, settings
|
||||
from jira_hub.core import api, cli, db, queue, release, settings
|
||||
from jira_hub.core.client import HubError
|
||||
from v6db import make_v6_db
|
||||
|
||||
|
||||
def test_v6_upgrade_keeps_jira_agent(tmp_path):
|
||||
make_v6_db(tmp_path / "hub.sqlite")
|
||||
rel = tmp_path / "release"
|
||||
rel.mkdir()
|
||||
(rel / "agent.py").write_text("AGENT_VERSION = 14\n", encoding="utf-8")
|
||||
release.publish(rel, rel / "agent.py")
|
||||
hub = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
|
||||
hub.claim_wait = 0
|
||||
conn = db.connect(tmp_path / "hub.sqlite")
|
||||
try:
|
||||
tok = settings.tokens(conn)["agents"]["jira"]
|
||||
assert tok == "T-agent"
|
||||
code, body = api.h_claim(hub, conn, None, {"_agent": "jira"}, {"agent_version": 14})
|
||||
assert code == 200 and body["agent_release"]["version"] == 14 # релиз v14 перенесён в release/jira/
|
||||
assert db._schema_version(conn) == 7
|
||||
tok = settings.tokens(conn)
|
||||
assert tok["client"] == "T-client" and tok["agents"]["jira"] == "T-agent" and "agent" not in tok
|
||||
@ -91,3 +101,112 @@ def test_agent_view_min_version_from_module_key(conn_two_modules):
|
||||
settings.set_(conn_two_modules, "min_agent_version", 5)
|
||||
assert queue.agent_view(conn_two_modules, "jira")["warning"] == "обновите агента"
|
||||
assert queue.agent_view(conn_two_modules, "stub")["warning"] is None
|
||||
|
||||
|
||||
# --- очередь, релиз и статус по агенту (Task 2) ------------------------------------------------------------------
|
||||
|
||||
def _two_batches(conn):
|
||||
queue.create_batch(conn, "w", "t", [{"type": "stub_cmd", "params": {}}])
|
||||
queue.create_batch(conn, "j", "t", [{"type": "ensure_issue",
|
||||
"params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}}])
|
||||
|
||||
|
||||
def test_wiki_queue_does_not_block_jira(conn_two_modules):
|
||||
conn = conn_two_modules
|
||||
_two_batches(conn)
|
||||
s = queue.claim(conn, "stub")
|
||||
assert s["type"] == "stub_cmd"
|
||||
j = queue.claim(conn, "jira") # stub_cmd ещё claimed, но это очередь другого агента
|
||||
assert j["type"] == "ensure_issue"
|
||||
assert queue.agent_view(conn, "stub")["current_command_id"] == s["id"]
|
||||
assert queue.agent_view(conn, "jira")["current_command_id"] == j["id"]
|
||||
|
||||
|
||||
def test_redeliver_only_own(conn_two_modules):
|
||||
conn = conn_two_modules
|
||||
_two_batches(conn)
|
||||
s = queue.claim(conn, "stub")
|
||||
j = queue.claim(conn, "jira")
|
||||
assert queue.claim(conn, "jira")["id"] == j["id"] # довыдача — своя команда, не stub_cmd
|
||||
assert queue.claim(conn, "stub")["id"] == s["id"]
|
||||
assert queue.agent_view(conn, "jira")["redelivered"][0]["id"] == j["id"]
|
||||
assert [r["id"] for r in queue.agent_view(conn, "stub")["redelivered"]] == [s["id"]]
|
||||
|
||||
|
||||
def test_agent_version_per_agent(conn_two_modules):
|
||||
conn = conn_two_modules
|
||||
queue.claim(conn, "jira", agent_version=14)
|
||||
assert queue.agent_version(conn, "jira") == "14" and queue.agent_version(conn, "stub") is None
|
||||
queue.claim(conn, "stub", agent_version=1)
|
||||
assert queue.agent_version(conn, "jira") == "14" and queue.agent_version(conn, "stub") == "1"
|
||||
|
||||
|
||||
def test_has_active_per_module(conn_two_modules):
|
||||
conn = conn_two_modules
|
||||
assert not queue.has_active(conn)
|
||||
queue.create_batch(conn, "w", "t", [{"type": "stub_cmd", "params": {}}])
|
||||
assert queue.has_active(conn) and queue.has_active(conn, "stub") and not queue.has_active(conn, "jira")
|
||||
|
||||
|
||||
def _publish(root, module, tmp_path, version):
|
||||
src = tmp_path / f"{module}{version}.py"
|
||||
src.write_text(f"AGENT_VERSION = {version}\n", encoding="utf-8")
|
||||
return release.publish(release.module_dir(root, module), src)
|
||||
|
||||
|
||||
def test_release_per_module(hub_server_two_modules, tmp_path):
|
||||
root = hub_server_two_modules["hub"].release_root
|
||||
_publish(root, "jira", tmp_path, 14)
|
||||
_publish(root, "stub", tmp_path, 1)
|
||||
assert release.module_dir(root, "stub") == root / "stub" and oct(release.module_dir(root, "stub").stat().st_mode & 0o777) == "0o700"
|
||||
jira, stub = hub_server_two_modules["agents"]["jira"], hub_server_two_modules["agents"]["stub"]
|
||||
assert jira.post("/agent/claim", {})["agent_release"]["version"] == 14
|
||||
assert stub.post("/agent/claim", {})["agent_release"]["version"] == 1
|
||||
assert jira.get("/agent/release")["version"] == 14 and stub.get("/agent/release")["version"] == 1
|
||||
|
||||
|
||||
def test_release_layout_migration(tmp_path):
|
||||
root = tmp_path / "release"
|
||||
root.mkdir()
|
||||
(root / "agent.py").write_text("AGENT_VERSION = 14\n", encoding="utf-8")
|
||||
release.publish(root, root / "agent.py") # старая раскладка: agent.py и manifest.json в корне
|
||||
hub = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
|
||||
assert not (root / "agent.py").exists() and not (root / "manifest.json").exists()
|
||||
assert release.current(hub.release_root / "jira")["version"] == 14
|
||||
api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups") # повторный старт ничего не ломает
|
||||
assert release.current(root / "jira")["version"] == 14
|
||||
|
||||
|
||||
def test_migrate_layout_failure_is_warning(tmp_path, monkeypatch, caplog):
|
||||
root = tmp_path / "release"
|
||||
root.mkdir()
|
||||
(root / "agent.py").write_text("AGENT_VERSION = 14\n", encoding="utf-8")
|
||||
monkeypatch.setattr("os.replace", lambda *a, **k: (_ for _ in ()).throw(OSError(13, "нет прав")))
|
||||
release.migrate_layout(root) # без исключения
|
||||
assert any("release" in r.getMessage() for r in caplog.records)
|
||||
|
||||
|
||||
def test_publish_unknown_module(tmp_path, monkeypatch, capsys):
|
||||
monkeypatch.setenv("JIRA_HUB_HOME", str(tmp_path / "home"))
|
||||
src = tmp_path / "agent.py"
|
||||
src.write_text("AGENT_VERSION = 1\n", encoding="utf-8")
|
||||
assert cli.main(["agent", "publish", "--module", "nope", str(src)]) == 2
|
||||
assert "неизвестный модуль nope; модули с агентами: jira" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_publish_module_option(tmp_path, monkeypatch, capsys):
|
||||
monkeypatch.setenv("JIRA_HUB_HOME", str(tmp_path / "home"))
|
||||
src = tmp_path / "agent.py"
|
||||
src.write_text("AGENT_VERSION = 1\n", encoding="utf-8")
|
||||
assert cli.main(["agent", "publish", "--module", "jira", str(src)]) == 0
|
||||
assert (tmp_path / "home" / "release" / "jira" / "manifest.json").exists()
|
||||
|
||||
|
||||
def test_agents_endpoint(hub_server_two_modules):
|
||||
c = hub_server_two_modules["client"]
|
||||
agents = c.get("/agents")["agents"]
|
||||
assert [a["module"] for a in agents] == ["jira", "stub"]
|
||||
assert c.get("/agent")["module"] == "jira" and c.get("/agent", module="stub")["module"] == "stub"
|
||||
with pytest.raises(HubError) as e:
|
||||
c.get("/agent", module="nope")
|
||||
assert e.value.status == 404 and e.value.error == "not_found"
|
||||
|
||||
@ -409,7 +409,7 @@ def test_board_route(hub, add_project):
|
||||
def _publish(hub, tmp_path, version):
|
||||
src = tmp_path / f"agent{version}.py"
|
||||
src.write_text(f"AGENT_VERSION = {version}\nprint('v{version}')\n", encoding="utf-8")
|
||||
return release.publish(hub["hub"].release_dir, src)
|
||||
return release.publish(release.module_dir(hub["hub"].release_root, "jira"), src)
|
||||
|
||||
|
||||
def test_claim_release_field(hub, tmp_path):
|
||||
@ -477,8 +477,7 @@ def test_agent_view_without_release(hub):
|
||||
|
||||
|
||||
def test_claim_survives_unreadable_release(hub):
|
||||
rd = hub["hub"].release_dir
|
||||
rd.mkdir()
|
||||
rd = release.module_dir(hub["hub"].release_root, "jira")
|
||||
(rd / "manifest.json").write_text('{"version": 6, "sha256": "x"}', encoding="utf-8")
|
||||
os.chmod(rd, 0)
|
||||
try:
|
||||
@ -500,8 +499,7 @@ def test_claim_survives_unreadable_release(hub):
|
||||
def test_invalid_manifest_one_warning_for_many_claims(hub, monkeypatch, caplog):
|
||||
"""M-3/M-7 #25, спека «Негодный манифест»: version = 6.9 — публикации нет; 10 claim и GET /agent — одно предупреждение."""
|
||||
monkeypatch.setattr(release, "_warned", None)
|
||||
rd = hub["hub"].release_dir
|
||||
rd.mkdir()
|
||||
rd = release.module_dir(hub["hub"].release_root, "jira")
|
||||
(rd / "manifest.json").write_text('{"version": 6.9, "sha256": "' + "a" * 64 + '"}', encoding="utf-8")
|
||||
for _ in range(10):
|
||||
assert hub["agent"].post("/agent/claim", {"agent_version": 5})["agent_release"] is None
|
||||
@ -516,7 +514,7 @@ def test_agent_release_non_utf8_is_404(hub, tmp_path):
|
||||
"""M-5 #25: agent.py не в UTF-8 при совпавшем sha256 — 404 no_release, а не 500."""
|
||||
import hashlib
|
||||
_publish(hub, tmp_path, 6)
|
||||
rd = hub["hub"].release_dir
|
||||
rd = release.module_dir(hub["hub"].release_root, "jira")
|
||||
data = b"AGENT_VERSION = 6\n\xff\n"
|
||||
(rd / "agent.py").write_bytes(data)
|
||||
m = json.loads((rd / "manifest.json").read_text(encoding="utf-8"))
|
||||
|
||||
@ -37,7 +37,7 @@ def test_agent_publish(tmp_path, monkeypatch, capsys):
|
||||
assert cli.main(["agent", "publish", str(src)]) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert out.startswith("опубликована v6, sha256 ") and "…, 18 байт" in out
|
||||
assert (paths.release_dir() / "manifest.json").exists()
|
||||
assert (release.module_dir(paths.release_root(), "jira") / "manifest.json").exists()
|
||||
assert cli.main(["agent", "publish", str(src)]) == 1
|
||||
assert "опубликована v6, новая должна быть больше" in capsys.readouterr().err
|
||||
|
||||
@ -62,8 +62,8 @@ def test_agent_publish_write_failure(tmp_path, monkeypatch, capsys):
|
||||
err = capsys.readouterr().err
|
||||
assert err.startswith("не удалось записать публикацию: ") and "Traceback" not in err
|
||||
monkeypatch.setattr(os, "replace", real)
|
||||
assert release.load(paths.release_dir())["version"] == 8
|
||||
assert sorted(p.name for p in paths.release_dir().iterdir()) == ["agent.py", "manifest.json"]
|
||||
assert release.load(release.module_dir(paths.release_root(), "jira"))["version"] == 8
|
||||
assert sorted(p.name for p in release.module_dir(paths.release_root(), "jira").iterdir()) == ["agent.py", "manifest.json"]
|
||||
|
||||
|
||||
def test_mcp_default_module(monkeypatch):
|
||||
|
||||
@ -238,7 +238,7 @@ def test_migrated_claimed_redelivered_end_to_end(tmp_path):
|
||||
_v1_database(path)
|
||||
conn = db.connect(path)
|
||||
settings.ensure_defaults(conn)
|
||||
cmd = queue.claim(conn)
|
||||
cmd = queue.claim(conn, "jira")
|
||||
assert cmd["id"] == 1
|
||||
assert queue.get_command(conn, 1)["deliveries"] == 2
|
||||
conn.close()
|
||||
|
||||
@ -44,17 +44,17 @@ def test_refresh_only_batch_has_no_tail(conn):
|
||||
def test_claim_order_pause_and_report(conn):
|
||||
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)
|
||||
c = queue.claim(conn, "jira", 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"]
|
||||
again = queue.claim(conn) # ответ потерян — та же команда снова
|
||||
again = queue.claim(conn, "jira") # ответ потерян — та же команда снова
|
||||
assert again["id"] == c["id"] and again["pause_seconds"] == 120
|
||||
assert queue.get_command(conn, c["id"])["deliveries"] == 2
|
||||
assert queue.agent_view(conn)["current_command_id"] == c["id"]
|
||||
assert queue.agent_view(conn, "jira")["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)
|
||||
tail = queue.claim(conn, "jira")
|
||||
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"
|
||||
refresh.record_state(conn, tail["id"], 1, 0)
|
||||
@ -62,7 +62,7 @@ def test_claim_order_pause_and_report(conn):
|
||||
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"
|
||||
assert queue.claim(conn, "jira")["params"]["key"] == "A-2" and queue.batch_view(conn, b2["id"])["status"] == "running"
|
||||
|
||||
|
||||
def test_claim_is_exclusive(tmp_path, open_db):
|
||||
@ -76,7 +76,7 @@ def test_claim_is_exclusive(tmp_path, open_db):
|
||||
|
||||
def worker(c):
|
||||
barrier.wait()
|
||||
results.append(queue.claim(c))
|
||||
results.append(queue.claim(c, "jira"))
|
||||
|
||||
threads = [threading.Thread(target=worker, args=(c,)) for c in conns]
|
||||
[t.start() for t in threads]
|
||||
@ -93,7 +93,7 @@ def test_backup_failure_does_not_block_claim(conn, caplog):
|
||||
def broken(label):
|
||||
raise OSError("диск полон")
|
||||
|
||||
c = queue.claim(conn, backup=broken)
|
||||
c = queue.claim(conn, "jira", 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
|
||||
@ -109,15 +109,15 @@ def test_temp_key_substitution_and_skip(conn, add_project):
|
||||
"labels": ["AWPCA"]}},
|
||||
{"type": "ensure_sprint", "params": {"key": "new:61", "sprint": "Спринт 2"}},
|
||||
])
|
||||
c1 = queue.claim(conn)
|
||||
c1 = queue.claim(conn, "jira")
|
||||
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)
|
||||
c2 = queue.claim(conn, "jira")
|
||||
assert c2["type"] == "ensure_link" and c2["params"]["source"] == "A-90"
|
||||
queue.report_result(conn, c2["id"], "done")
|
||||
c3 = queue.claim(conn)
|
||||
c3 = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, c3["id"], "failed", error="HTTP 400")
|
||||
tail = queue.claim(conn) # ensure_sprint пропущена, refresh без new:61
|
||||
tail = queue.claim(conn, "jira") # 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"]
|
||||
@ -126,12 +126,12 @@ def test_temp_key_substitution_and_skip(conn, add_project):
|
||||
|
||||
def test_cancel_and_retry(conn):
|
||||
b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
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)["id"] == c["id"] # взятая из отменённого пакета довыдаётся
|
||||
assert queue.claim(conn, "jira")["id"] == c["id"] # взятая из отменённого пакета довыдаётся
|
||||
forced = queue.cancel_command(conn, c["id"], force=True)
|
||||
assert forced["status"] == "cancelled"
|
||||
after = queue.report_result(conn, c["id"], "done", result={"changed": []})
|
||||
@ -144,7 +144,7 @@ def test_cancel_and_retry(conn):
|
||||
|
||||
def test_report_after_force_cancel_keeps_cancelled(conn):
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.cancel_command(conn, c["id"], force=True)
|
||||
assert queue.report_result(conn, c["id"], "failed", error="e")["status"] == "cancelled"
|
||||
|
||||
@ -153,7 +153,7 @@ def test_reconcile_confirms_by_state(conn, add_project):
|
||||
add_project(conn)
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "labels": ["AWPCA"]}], [])
|
||||
queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
b = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
r = queue.batch_view(conn, b["id"])["commands"][0]
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, deliveries = 1 WHERE id = ?", (db.now(), r["id"]))
|
||||
@ -164,14 +164,14 @@ def test_reconcile_confirms_by_state(conn, add_project):
|
||||
|
||||
|
||||
def test_auto_refresh_needs_projects(conn):
|
||||
queue.claim(conn, agent_version=1, settings_version=1)
|
||||
queue.claim(conn, "jira", agent_version=1, settings_version=1)
|
||||
assert refresh.maybe_auto_refresh(conn) is None # агент на связи, очередь пуста, проектов нет
|
||||
|
||||
|
||||
def test_auto_refresh_two_projects(conn, add_project): # сценарий «Два проекта»
|
||||
add_project(conn)
|
||||
add_project(conn, "STOCKS_DRRK", board_id=None)
|
||||
queue.claim(conn, agent_version=1, settings_version=1)
|
||||
queue.claim(conn, "jira", agent_version=1, settings_version=1)
|
||||
b = refresh.maybe_auto_refresh(conn)
|
||||
assert queue.batch_view(conn, b["id"])["commands"][0]["params"] == {"jql": 'project = AURORA AND labels in ("AWPCA", "STOCKS_DRRK")'}
|
||||
|
||||
@ -179,11 +179,11 @@ def test_auto_refresh_two_projects(conn, add_project): # сцен
|
||||
def test_auto_refresh_conditions(conn, add_project):
|
||||
add_project(conn)
|
||||
assert refresh.maybe_auto_refresh(conn) is None # агент не на связи
|
||||
queue.claim(conn, agent_version=1, settings_version=1) # отметился, очередь пуста
|
||||
queue.claim(conn, "jira", agent_version=1, settings_version=1) # отметился, очередь пуста
|
||||
b = refresh.maybe_auto_refresh(conn)
|
||||
assert b and queue.batch_view(conn, b["id"])["title"] == "Автообновление"
|
||||
assert refresh.maybe_auto_refresh(conn) is None # очередь занята
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, c["id"], "done")
|
||||
assert refresh.maybe_auto_refresh(conn) is None # интервал не прошёл
|
||||
conn.execute("UPDATE sync_run SET started_at = '2020-01-01T00:00:00+00:00'")
|
||||
@ -196,16 +196,16 @@ def test_auto_refresh_conditions(conn, add_project):
|
||||
def test_backup_called_for_first_write_only(conn, tmp_path):
|
||||
calls = []
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||||
c1 = queue.claim(conn, backup=calls.append)
|
||||
c1 = queue.claim(conn, "jira", backup=calls.append)
|
||||
queue.report_result(conn, c1["id"], "done")
|
||||
queue.claim(conn, backup=calls.append)
|
||||
queue.claim(conn, "jira", backup=calls.append)
|
||||
assert calls == [f"batch-{c1['batch_id']}"]
|
||||
|
||||
|
||||
def test_no_backup_for_refresh_batch(conn):
|
||||
calls = []
|
||||
queue.create_batch(conn, "x", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
|
||||
queue.claim(conn, backup=calls.append)
|
||||
queue.claim(conn, "jira", backup=calls.append)
|
||||
assert calls == []
|
||||
|
||||
|
||||
@ -233,7 +233,7 @@ def test_rotation_keeps_schema_migration_copy(tmp_path, open_db):
|
||||
|
||||
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)
|
||||
c = queue.claim(conn, "jira")
|
||||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
|
||||
return c
|
||||
|
||||
@ -254,7 +254,7 @@ def test_reconcile_does_not_overwrite_finished_command(conn, monkeypatch, add_pr
|
||||
add_project(conn)
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "labels": ["AWPCA"]}], [])
|
||||
queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "sprint": "Спринт 2", "labels": ["AWPCA"]}], [])
|
||||
real = commands.check_confirmed
|
||||
|
||||
@ -272,7 +272,7 @@ 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)
|
||||
first = queue.claim(conn, "jira")
|
||||
assert first["type"] == "ensure_issue"
|
||||
b = queue.create_batch(conn, "B", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
|
||||
return a, b, first
|
||||
@ -280,16 +280,16 @@ def _two_batches_with_dependent(conn):
|
||||
|
||||
def test_redelivered_definer_blocks_dependents(conn):
|
||||
a, b, first = _two_batches_with_dependent(conn)
|
||||
assert queue.claim(conn)["id"] == first["id"] # довыдача определителя, зависимые ждут
|
||||
assert queue.claim(conn, "jira")["id"] == first["id"] # довыдача определителя, зависимые ждут
|
||||
conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],))
|
||||
nxt = queue.claim(conn)
|
||||
nxt = queue.claim(conn, "jira")
|
||||
assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9"
|
||||
|
||||
|
||||
def test_auto_refresh_waits_for_claimed_without_report(conn):
|
||||
queue.claim(conn, agent_version=1, settings_version=1) # агент на связи
|
||||
queue.claim(conn, "jira", agent_version=1, settings_version=1) # агент на связи
|
||||
queue.create_batch(conn, "y", "claude", [fields("A-3")])
|
||||
queue.claim(conn) # выдана, отчёта нет
|
||||
queue.claim(conn, "jira") # выдана, отчёта нет
|
||||
conn.execute("DELETE FROM command WHERE status = 'queued'") # хвостовой refresh не мешает проверке
|
||||
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE status = 'claimed'") # давно без отчёта
|
||||
assert refresh.maybe_auto_refresh(conn) is None # claimed без отчёта — очередь занята
|
||||
@ -297,23 +297,23 @@ def test_auto_refresh_waits_for_claimed_without_report(conn):
|
||||
|
||||
def test_redelivery_of_lost_claim_before_queue(conn):
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||||
c = queue.claim(conn)
|
||||
again = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
again = queue.claim(conn, "jira")
|
||||
assert again["id"] == c["id"] and again["params"] == c["params"]
|
||||
assert queue.agent_view(conn)["current_command_id"] == c["id"]
|
||||
view = queue.agent_view(conn)["redelivered"]
|
||||
assert queue.agent_view(conn, "jira")["current_command_id"] == c["id"]
|
||||
view = queue.agent_view(conn, "jira")["redelivered"]
|
||||
assert [(x["id"], x["deliveries"]) for x in view] == [(c["id"], 2)]
|
||||
queue.report_result(conn, c["id"], "done")
|
||||
assert queue.claim(conn)["params"]["key"] == "A-2"
|
||||
assert queue.agent_view(conn)["redelivered"] == []
|
||||
assert queue.claim(conn, "jira")["params"]["key"] == "A-2"
|
||||
assert queue.agent_view(conn, "jira")["redelivered"] == []
|
||||
|
||||
|
||||
def test_redelivery_limit_fails_command(conn):
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||||
c = queue.claim(conn)
|
||||
queue.claim(conn)
|
||||
queue.claim(conn) # третья выдача
|
||||
nxt = queue.claim(conn) # четвёртой нет — failed и дальше по очереди
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.claim(conn, "jira")
|
||||
queue.claim(conn, "jira") # третья выдача
|
||||
nxt = queue.claim(conn, "jira") # четвёртой нет — failed и дальше по очереди
|
||||
assert nxt["params"]["key"] == "A-2"
|
||||
failed = queue.get_command(conn, c["id"])
|
||||
assert failed["status"] == "failed" and failed["error"] == "агент не отчитался после 3 выдач" and failed["finished_at"]
|
||||
@ -326,21 +326,21 @@ def test_redelivery_oldest_first(conn):
|
||||
rows = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields' ORDER BY id").fetchall()
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-13T10:00:00+00:00', deliveries = 1 WHERE id = ?", (rows[1]["id"],))
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-13T11:00:00+00:00', deliveries = 1 WHERE id = ?", (rows[0]["id"],))
|
||||
assert queue.claim(conn)["id"] == rows[1]["id"]
|
||||
assert queue.claim(conn, "jira")["id"] == rows[1]["id"]
|
||||
|
||||
|
||||
def test_redelivered_refresh_keeps_one_sync_run(conn):
|
||||
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
r = queue.claim(conn)
|
||||
assert queue.claim(conn)["id"] == r["id"]
|
||||
r = queue.claim(conn, "jira")
|
||||
assert queue.claim(conn, "jira")["id"] == r["id"]
|
||||
assert conn.execute("SELECT COUNT(*) AS n FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["n"] == 1
|
||||
|
||||
|
||||
def test_redelivery_does_not_repeat_backup(conn):
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||||
labels = []
|
||||
queue.claim(conn, backup=labels.append)
|
||||
queue.claim(conn, backup=labels.append)
|
||||
queue.claim(conn, "jira", backup=labels.append)
|
||||
queue.claim(conn, "jira", backup=labels.append)
|
||||
assert len(labels) == 1
|
||||
|
||||
|
||||
@ -349,7 +349,7 @@ def test_migrated_claimed_is_redelivered(conn):
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||||
cid = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields'").fetchone()["id"]
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-12T00:00:00+00:00', deliveries = 1 WHERE id = ?", (cid,))
|
||||
assert queue.claim(conn)["id"] == cid
|
||||
assert queue.claim(conn, "jira")["id"] == cid
|
||||
assert queue.get_command(conn, cid)["deliveries"] == 2
|
||||
|
||||
|
||||
@ -365,7 +365,7 @@ def test_late_report_after_force_cancel_keeps_issue_key(conn):
|
||||
mirror.set_note(conn, "new:7", {"fact_status": "В работе"}, "import")
|
||||
queue.create_batch(conn, "x", "claude", [
|
||||
{"type": "ensure_issue", "params": {"temp_key": "new:7", "issue_type": "task", "summary": "Тема"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
cancelled = queue.cancel_command(conn, c["id"], force=True)
|
||||
late = queue.report_result(conn, c["id"], "done", result={"key": "A-20", "created": True}, jira_calls=3)
|
||||
assert late["status"] == "cancelled" and late["result"] == {"key": "A-20", "created": True} and late["jira_calls"] == 3
|
||||
@ -378,7 +378,7 @@ def test_late_report_after_force_cancel_keeps_issue_key(conn):
|
||||
def test_late_failed_report_after_force_cancel_changes_nothing(conn):
|
||||
queue.create_batch(conn, "x", "claude", [
|
||||
{"type": "ensure_issue", "params": {"temp_key": "new:7", "issue_type": "task", "summary": "Тема"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.cancel_command(conn, c["id"], force=True)
|
||||
late = queue.report_result(conn, c["id"], "failed", error="HTTP 400")
|
||||
assert late["status"] == "cancelled" and late["result"] is None and late["error"] is None
|
||||
@ -386,14 +386,14 @@ def test_late_failed_report_after_force_cancel_changes_nothing(conn):
|
||||
|
||||
def _fail_by_limit(conn, cid):
|
||||
for _ in range(queue.MAX_DELIVERIES - 1):
|
||||
assert queue.claim(conn)["id"] == cid
|
||||
queue.claim(conn) # четвёртый claim: cid → failed по пределу
|
||||
assert queue.claim(conn, "jira")["id"] == cid
|
||||
queue.claim(conn, "jira") # четвёртый claim: cid → failed по пределу
|
||||
assert queue.get_command(conn, cid)["status"] == "failed"
|
||||
|
||||
|
||||
def test_error_is_trimmed_to_1500(conn):
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
out = queue.report_result(conn, c["id"], "failed", error="е" * 5000)
|
||||
assert out["error"] == "е" * 1500 and queue.ERROR_LIMIT == 1500
|
||||
|
||||
@ -402,7 +402,7 @@ def test_late_done_after_delivery_limit_keeps_issue_key(conn):
|
||||
mirror.set_note(conn, "new:8", {"fact_status": "План"}, "import")
|
||||
queue.create_batch(conn, "x", "claude", [
|
||||
{"type": "ensure_issue", "params": {"temp_key": "new:8", "issue_type": "task", "summary": "Тема"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
_fail_by_limit(conn, c["id"])
|
||||
late = queue.report_result(conn, c["id"], "done", result={"key": "A-30", "created": True}, jira_calls=2)
|
||||
assert late["status"] == "failed" and late["result"] == {"key": "A-30", "created": True}
|
||||
@ -414,9 +414,9 @@ def test_late_done_after_delivery_limit_keeps_issue_key(conn):
|
||||
|
||||
def test_cancelled_batch_with_done_command_is_cancelled(conn):
|
||||
b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2"), fields("A-3")])
|
||||
first = queue.claim(conn)
|
||||
first = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, first["id"], "done", result={"changed": []})
|
||||
second = queue.claim(conn)
|
||||
second = queue.claim(conn, "jira")
|
||||
queue.cancel_batch(conn, b["id"])
|
||||
queue.report_result(conn, second["id"], "done", result={"changed": []})
|
||||
view = queue.batch_view(conn, b["id"])
|
||||
@ -430,14 +430,14 @@ def test_empty_full_refresh_keeps_neighbors(conn, add_project):
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "x", "labels": []}], [],
|
||||
fetched_at="2020-01-01T00:00:00+00:00") # прочитана давно — до старта прогона
|
||||
queue.create_batch(conn, "полный", "claude", [{"type": "refresh", "params": {"jql": jql}}])
|
||||
r = queue.claim(conn)
|
||||
r = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, r["id"], "done") # загрузок состояния не было: 0 задач
|
||||
assert conn.execute("SELECT neighbor FROM issue WHERE key = 'A-1'").fetchone()["neighbor"] == 0
|
||||
|
||||
|
||||
def test_cancel_finished_batch_stays_done(conn):
|
||||
b = queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||||
while (c := queue.claim(conn)):
|
||||
while (c := queue.claim(conn, "jira")):
|
||||
queue.report_result(conn, c["id"], "done", result={"changed": []})
|
||||
view = queue.cancel_batch(conn, b["id"])
|
||||
assert view["status"] == "done"
|
||||
@ -449,10 +449,10 @@ def test_redelivered_full_refresh_marks_neighbors_once(conn, add_project):
|
||||
jql = jira_projects.base_jql(conn)
|
||||
mirror.apply_state(conn, [{"key": "OLD", "summary": "x", "labels": []}], [], fetched_at="2020-01-01T00:00:00+00:00")
|
||||
queue.create_batch(conn, "полный", "claude", [{"type": "refresh", "params": {"jql": jql}}])
|
||||
r = queue.claim(conn)
|
||||
r = queue.claim(conn, "jira")
|
||||
refresh.record_state(conn, r["id"], 1, 0)
|
||||
mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], [])
|
||||
assert queue.claim(conn)["id"] == r["id"] # отчёт потерян — довыдача
|
||||
assert queue.claim(conn, "jira")["id"] == r["id"] # отчёт потерян — довыдача
|
||||
assert conn.execute("SELECT deliveries FROM command WHERE id = ?", (r["id"],)).fetchone()["deliveries"] == 2
|
||||
refresh.record_state(conn, r["id"], 1, 0)
|
||||
mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], [])
|
||||
@ -464,10 +464,10 @@ def test_redelivered_full_refresh_marks_neighbors_once(conn, add_project):
|
||||
|
||||
def test_redelivered_refresh_resets_counts(conn):
|
||||
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
r = queue.claim(conn)
|
||||
r = queue.claim(conn, "jira")
|
||||
refresh.record_state(conn, r["id"], 150, 40)
|
||||
started = conn.execute("SELECT started_at FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["started_at"]
|
||||
assert queue.claim(conn)["id"] == r["id"] # отчёт потерян — довыдача
|
||||
assert queue.claim(conn, "jira")["id"] == r["id"] # отчёт потерян — довыдача
|
||||
refresh.record_state(conn, r["id"], 150, 40)
|
||||
run = conn.execute("SELECT * FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()
|
||||
assert (run["issues_count"], run["links_count"]) == (150, 40)
|
||||
@ -476,16 +476,16 @@ def test_redelivered_refresh_resets_counts(conn):
|
||||
|
||||
def test_refresh_failed_by_limit_finishes_sync_run(conn):
|
||||
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
r = queue.claim(conn)
|
||||
r = queue.claim(conn, "jira")
|
||||
for _ in range(queue.MAX_DELIVERIES):
|
||||
queue.claim(conn)
|
||||
queue.claim(conn, "jira")
|
||||
assert queue.get_command(conn, r["id"])["status"] == "failed"
|
||||
assert conn.execute("SELECT finished_at FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["finished_at"]
|
||||
|
||||
|
||||
def test_cancel_batch_reports_still_claimed(conn):
|
||||
b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
view = queue.cancel_batch(conn, b["id"])
|
||||
assert view["still_claimed"] == [{"id": c["id"], "type": "ensure_fields"}]
|
||||
assert queue.get_command(conn, c["id"])["status"] == "claimed"
|
||||
@ -496,9 +496,9 @@ def test_dependent_skipped_when_creation_cancelled(conn):
|
||||
b = queue.create_batch(conn, "x", "claude", [
|
||||
{"type": "ensure_issue", "params": {"temp_key": "new:901", "issue_type": "task", "summary": "Отменённая"}},
|
||||
{"type": "ensure_comment", "params": {"key": "new:901", "marker": "m", "body": "b"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.cancel_command(conn, c["id"], force=True)
|
||||
assert queue.claim(conn) is None # комментарий и хвостовой refresh пропущены
|
||||
assert queue.claim(conn, "jira") is None # комментарий и хвостовой refresh пропущены
|
||||
cmds = queue.batch_view(conn, b["id"])["commands"]
|
||||
assert [x["status"] for x in cmds] == ["cancelled", "skipped", "skipped"]
|
||||
assert cmds[1]["error"] == "создание задачи new:901 отменено"
|
||||
@ -510,11 +510,11 @@ def test_dependent_on_failed_and_cancelled_names_cancelled(conn):
|
||||
{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Упавшая"}},
|
||||
{"type": "ensure_issue", "params": {"temp_key": "new:2", "issue_type": "task", "summary": "Отменённая"}},
|
||||
{"type": "ensure_link", "params": {"source": "new:1", "target": "new:2", "link_type": "связанные"}}])
|
||||
first = queue.claim(conn)
|
||||
first = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, first["id"], "failed", error="HTTP 400")
|
||||
second = queue.claim(conn)
|
||||
second = queue.claim(conn, "jira")
|
||||
queue.cancel_command(conn, second["id"], force=True)
|
||||
queue.claim(conn)
|
||||
queue.claim(conn, "jira")
|
||||
link = queue.batch_view(conn, b["id"])["commands"][2]
|
||||
assert link["status"] == "skipped" and link["error"] == "создание задачи new:2 отменено"
|
||||
|
||||
@ -522,7 +522,7 @@ def test_dependent_on_failed_and_cancelled_names_cancelled(conn):
|
||||
def test_reconcile_confirms_fields_with_crlf(conn):
|
||||
"""O-9 #16, решение 11, спека «Описание с CRLF в зеркале»: потерянный отчёт закрывается по состоянию."""
|
||||
queue.create_batch(conn, "x", "claude", [{"type": "ensure_fields", "params": {"key": "A-1", "description": "a\nb"}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "description": "a\r\nb", "labels": []}], [])
|
||||
assert queue.reconcile(conn) == [c["id"]]
|
||||
assert queue.get_command(conn, c["id"])["status"] == "confirmed_by_state"
|
||||
@ -531,15 +531,15 @@ def test_reconcile_confirms_fields_with_crlf(conn):
|
||||
def test_delivery_limit_of_write_command_keeps_open_sync_run(conn):
|
||||
"""T5.2 #16: предел выдач у не-refresh команды не закрывает журнал чужой загрузки (finished_at — только для refresh)."""
|
||||
queue.create_batch(conn, "x", "claude", [fields("A-1")])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
r = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-9"]}}])
|
||||
rid = queue.batch_view(conn, r["id"])["commands"][0]["id"]
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2099-01-01T00:00:00+00:00', deliveries = 1 "
|
||||
"WHERE id = ?", (rid,))
|
||||
conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, 'keys', 'A-9', ?)", (db.now(), rid))
|
||||
for _ in range(queue.MAX_DELIVERIES - 1):
|
||||
assert queue.claim(conn)["id"] == c["id"]
|
||||
assert queue.claim(conn)["id"] == rid # c — failed по пределу, довыдаётся refresh
|
||||
assert queue.claim(conn, "jira")["id"] == c["id"]
|
||||
assert queue.claim(conn, "jira")["id"] == rid # c — failed по пределу, довыдаётся refresh
|
||||
assert queue.get_command(conn, c["id"])["status"] == "failed"
|
||||
assert conn.execute("SELECT finished_at FROM sync_run WHERE command_id = ?", (rid,)).fetchone()["finished_at"] is None
|
||||
|
||||
@ -625,16 +625,16 @@ def test_bug_with_attachment_one_batch(conn, store, add_project):
|
||||
bug(), {"type": "ensure_sprint", "params": {"key": "new:1", "sprint": "Спринт 2"}}, attach(key="new:1")],
|
||||
files_dir=store)
|
||||
assert b["count"] == 4 and b["estimate_seconds"] == 120 + 30 + 30 + 2
|
||||
c1 = queue.claim(conn, files_dir=store)
|
||||
c1 = queue.claim(conn, "jira", files_dir=store)
|
||||
assert c1["type"] == "ensure_issue"
|
||||
queue.report_result(conn, c1["id"], "done", result={"key": "AURORA-1650"})
|
||||
c2 = queue.claim(conn, files_dir=store)
|
||||
c2 = queue.claim(conn, "jira", files_dir=store)
|
||||
assert c2["type"] == "ensure_sprint"
|
||||
queue.report_result(conn, c2["id"], "done")
|
||||
c3 = queue.claim(conn, files_dir=store)
|
||||
c3 = queue.claim(conn, "jira", files_dir=store)
|
||||
assert c3["type"] == "ensure_attachment" and c3["pause_seconds"] == 30
|
||||
assert c3["params"] == {"key": "AURORA-1650", "file": SHA, "filename": "ОСВ 60.xlsx", "size": 3}
|
||||
again = queue.claim(conn, files_dir=store) # довыдача — тоже с size
|
||||
again = queue.claim(conn, "jira", files_dir=store) # довыдача — тоже с size
|
||||
assert again["id"] == c3["id"] and again["params"]["size"] == 3
|
||||
stored = queue.get_command(conn, c3["id"])["params"]
|
||||
assert stored == {"key": "AURORA-1650", "file": SHA, "filename": "ОСВ 60.xlsx"} # в базе без size
|
||||
@ -643,15 +643,15 @@ def test_bug_with_attachment_one_batch(conn, store, add_project):
|
||||
def test_attachment_size_null_when_file_gone(conn, store):
|
||||
queue.create_batch(conn, "вложение", "claude", [attach()], files_dir=store)
|
||||
(store / SHA).unlink()
|
||||
c = queue.claim(conn, files_dir=store)
|
||||
c = queue.claim(conn, "jira", files_dir=store)
|
||||
assert c["type"] == "ensure_attachment" and c["params"]["size"] is None
|
||||
assert queue.claim(conn)["params"]["size"] is None # без хранилища — тоже null, не исключение
|
||||
assert queue.claim(conn, "jira")["params"]["size"] is None # без хранилища — тоже null, не исключение
|
||||
assert queue.deliver_params(conn, queue.get_command(conn, c["id"]), store)["size"] is None
|
||||
|
||||
|
||||
def test_attachment_retry_passes_validation(conn, store):
|
||||
queue.create_batch(conn, "вложение", "claude", [attach()], files_dir=store)
|
||||
c = queue.claim(conn, files_dir=store)
|
||||
c = queue.claim(conn, "jira", files_dir=store)
|
||||
queue.report_result(conn, c["id"], "failed", error="HTTP 413")
|
||||
r = queue.retry_command(conn, c["id"], "claude", files_dir=store)
|
||||
assert queue.batch_view(conn, r["id"])["commands"][0]["params"] == {"key": "A-1", "file": SHA, "filename": "ОСВ 60.xlsx"}
|
||||
@ -663,7 +663,7 @@ def test_attachment_retry_passes_validation(conn, store):
|
||||
def test_attachment_not_confirmed_by_state(conn, store):
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "Т", "labels": []}], [])
|
||||
queue.create_batch(conn, "вложение", "claude", [attach()], files_dir=store)
|
||||
c = queue.claim(conn, files_dir=store)
|
||||
c = queue.claim(conn, "jira", files_dir=store)
|
||||
assert queue.reconcile(conn) == [] and queue.get_command(conn, c["id"])["status"] == "claimed"
|
||||
|
||||
|
||||
@ -900,16 +900,16 @@ def test_refresh_delivered_with_boards(conn, add_project): # сце
|
||||
add_project(conn, "STOCKS_DRRK", board_id=5120, board_type=None)
|
||||
add_project(conn, "C", board_id=4216)
|
||||
b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": "key = A-1"}}])
|
||||
got = queue.claim(conn)
|
||||
got = queue.claim(conn, "jira")
|
||||
assert got["params"]["boards"] == [{"id": 4216, "type": "scrum"}, {"id": 5120, "type": None}]
|
||||
assert "boards" not in queue.batch_view(conn, b["id"])["commands"][0]["params"]
|
||||
assert queue.claim(conn)["params"]["boards"] == got["params"]["boards"] # довыдача — тоже с boards
|
||||
assert queue.claim(conn, "jira")["params"]["boards"] == got["params"]["boards"] # довыдача — тоже с boards
|
||||
|
||||
|
||||
def test_refresh_boards_empty_list(conn, add_project):
|
||||
add_project(conn, board_id=None, board_type=None)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
assert queue.claim(conn)["params"]["boards"] == []
|
||||
assert queue.claim(conn, "jira")["params"]["boards"] == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("result, expected", [
|
||||
@ -925,7 +925,7 @@ def test_refresh_boards_empty_list(conn, add_project):
|
||||
def test_refresh_result_board_types(conn, add_project, result, expected):
|
||||
add_project(conn, "STOCKS_DRRK", board_id=5120, board_type=None)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, c["id"], "done", result=result)
|
||||
assert projects.get(conn, "STOCKS_DRRK")["board_type"] == expected
|
||||
|
||||
@ -933,7 +933,7 @@ def test_refresh_result_board_types(conn, add_project, result, expected):
|
||||
def test_refresh_result_board_type_changes_scrum_to_kanban(conn, add_project): # доска в Jira стала канбаном
|
||||
add_project(conn, "STOCKS_DRRK", board_id=5120, board_type="scrum")
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, c["id"], "done", result={"boards": {"5120": "kanban"}})
|
||||
assert projects.get(conn, "STOCKS_DRRK")["board_type"] == "kanban"
|
||||
|
||||
@ -941,7 +941,7 @@ def test_refresh_result_board_type_changes_scrum_to_kanban(conn, add_project):
|
||||
def test_refresh_failed_result_records_board_types(conn, add_project): # R-3 / T6.1: тип доски — факт о доске, а не часть выборки
|
||||
add_project(conn, "STOCKS_DRRK", board_id=5120, board_type=None)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
queue.report_result(conn, c["id"], "failed", result={"boards": {"5120": "kanban"}}, error="e")
|
||||
assert projects.get(conn, "STOCKS_DRRK")["board_type"] == "kanban"
|
||||
|
||||
@ -951,7 +951,7 @@ def test_refresh_failed_result_records_board_types(conn, add_project): # R-3
|
||||
def _full_run(conn, result, started_at=None, status="done"):
|
||||
"""Полный refresh: поставить, выдать, отчитаться; started_at — подменить время начала прогона."""
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
if started_at:
|
||||
conn.execute("UPDATE sync_run SET started_at = ? WHERE command_id = ?", (started_at, c["id"]))
|
||||
queue.report_result(conn, c["id"], status, result=result)
|
||||
@ -961,7 +961,7 @@ def _full_run(conn, result, started_at=None, status="done"):
|
||||
def test_first_full_refresh_gets_backfill(conn, add_project): # сценарий «Первый refresh после выкатки»
|
||||
add_project(conn)
|
||||
b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}])
|
||||
got = queue.claim(conn)
|
||||
got = queue.claim(conn, "jira")
|
||||
assert got["params"]["history"] == {
|
||||
"jql": '(project = AURORA AND labels in ("AWPCA")) AND sprint in openSprints() AND statusCategory != Done'}
|
||||
assert "history" not in queue.batch_view(conn, b["id"])["commands"][0]["params"]
|
||||
@ -972,7 +972,7 @@ def test_regular_refresh_gets_since(conn, add_project): # «Р
|
||||
_full_run(conn, {"history": {"issues": 1, "changes": 2, "incomplete": []}}, "2026-10-22T09:00:00+00:00")
|
||||
_full_run(conn, {"history": {"error": "HTTP 500"}}, "2026-10-22T10:00:00+00:00")
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}])
|
||||
assert queue.claim(conn)["params"]["history"] == {"since": "2026-10-22T08:45:00Z"}
|
||||
assert queue.claim(conn, "jira")["params"]["history"] == {"since": "2026-10-22T08:45:00Z"}
|
||||
|
||||
|
||||
def test_new_project_gets_since_and_backfill(conn, add_project): # «Новый проект»
|
||||
@ -980,7 +980,7 @@ def test_new_project_gets_since_and_backfill(conn, add_project): # «Н
|
||||
_full_run(conn, {"history": {"issues": 0, "changes": 0, "incomplete": []}}, "2026-10-22T09:00:00+00:00")
|
||||
add_project(conn, "STOCKS_DRRK", board_id=5120)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}])
|
||||
h = queue.claim(conn)["params"]["history"]
|
||||
h = queue.claim(conn, "jira")["params"]["history"]
|
||||
assert h["since"] == "2026-10-22T08:45:00Z"
|
||||
assert h["jql"] == "(" + jira_projects.base_jql(conn) + ") AND sprint in openSprints() AND statusCategory != Done"
|
||||
assert "STOCKS_DRRK" in h["jql"]
|
||||
@ -990,7 +990,7 @@ def test_tail_refresh_since_and_no_mark(conn, add_project): # «Х
|
||||
add_project(conn)
|
||||
_full_run(conn, {"history": {"issues": 0, "changes": 0, "incomplete": []}}, "2026-10-22T09:00:00+00:00")
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
c = queue.claim(conn)
|
||||
c = queue.claim(conn, "jira")
|
||||
assert c["params"]["history"] == {"since": "2026-10-22T08:45:00Z"}
|
||||
queue.report_result(conn, c["id"], "done", result={"history": {"issues": 0, "changes": 0, "incomplete": []}})
|
||||
assert conn.execute("SELECT history_at FROM sync_run WHERE command_id = ?", (c["id"],)).fetchone()["history_at"] is None
|
||||
@ -999,19 +999,19 @@ def test_tail_refresh_since_and_no_mark(conn, add_project): # «Х
|
||||
def test_tail_refresh_without_mark_gets_nothing(conn, add_project): # «Хвост пакета до первой отметки»
|
||||
add_project(conn)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
assert "history" not in queue.claim(conn)["params"]
|
||||
assert "history" not in queue.claim(conn, "jira")["params"]
|
||||
|
||||
|
||||
def test_no_projects_no_history(conn):
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
assert "history" not in queue.claim(conn)["params"]
|
||||
assert "history" not in queue.claim(conn, "jira")["params"]
|
||||
|
||||
|
||||
def test_redelivery_recomputes_history(conn, add_project): # Review Focus 3
|
||||
add_project(conn)
|
||||
b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": jira_projects.base_jql(conn)}}])
|
||||
first = queue.claim(conn)
|
||||
again = queue.claim(conn) # отчёта нет — довыдача
|
||||
first = queue.claim(conn, "jira")
|
||||
again = queue.claim(conn, "jira") # отчёта нет — довыдача
|
||||
assert again["id"] == first["id"] and again["params"]["history"] == first["params"]["history"]
|
||||
assert "history" not in queue.batch_view(conn, b["id"])["commands"][0]["params"]
|
||||
|
||||
@ -1086,7 +1086,7 @@ def test_fetch_size_limit_boundary(fconn):
|
||||
def test_fetch_report_done_non_dict_result_fails(fconn, tmp_path):
|
||||
"""T3.4: done с result не-словарём — failed, вложение не отмечено скачанным."""
|
||||
queue.create_batch(fconn, "x", "claude", [fetch()])
|
||||
c = queue.claim(fconn)
|
||||
c = queue.claim(fconn, "jira")
|
||||
r = queue.report_result(fconn, c["id"], "done", result="ok", files_dir=tmp_path)
|
||||
assert r["status"] == "failed"
|
||||
assert fconn.execute("SELECT COUNT(*) AS n FROM attachment_file").fetchone()["n"] == 0
|
||||
@ -1095,7 +1095,7 @@ def test_fetch_report_done_non_dict_result_fails(fconn, tmp_path):
|
||||
def test_fetch_no_backup(fconn):
|
||||
queue.create_batch(fconn, "x", "claude", [fetch()])
|
||||
calls = []
|
||||
cmd = queue.claim(fconn, backup=calls.append)
|
||||
cmd = queue.claim(fconn, "jira", backup=calls.append)
|
||||
assert cmd["type"] == "fetch_attachment" and cmd["pause_seconds"] == 2 and calls == []
|
||||
|
||||
|
||||
@ -1161,7 +1161,7 @@ def test_request_fetch_agent_too_old(fconn, tmp_path):
|
||||
def test_request_fetch_after_failed(fconn, tmp_path):
|
||||
root = tmp_path / "files"
|
||||
first = jira_fetch.request_fetch(fconn, root, "31337", "x")
|
||||
cmd = queue.claim(fconn)
|
||||
cmd = queue.claim(fconn, "jira")
|
||||
queue.report_result(fconn, cmd["id"], "failed", error="нет сети", files_dir=root)
|
||||
again = jira_fetch.request_fetch(fconn, root, "31337", "x")
|
||||
assert again["state"] == "fetching" and again["batch_id"] != first["batch_id"]
|
||||
@ -1169,7 +1169,7 @@ def test_request_fetch_after_failed(fconn, tmp_path):
|
||||
|
||||
def report_done(conn, root, result):
|
||||
queue.create_batch(conn, "x", "claude", [fetch()])
|
||||
cmd = queue.claim(conn)
|
||||
cmd = queue.claim(conn, "jira")
|
||||
return queue.report_result(conn, cmd["id"], "done", result=result, files_dir=root)
|
||||
|
||||
|
||||
|
||||
@ -57,7 +57,7 @@ def test_on_result_called(tmp_path, open_db, stub):
|
||||
path = tmp_path / "hub.sqlite"
|
||||
conn = open_db(path)
|
||||
queue.create_batch(conn, "x", "claude", [{"type": "w", "params": {}}])
|
||||
cmd = queue.claim(conn)
|
||||
cmd = queue.claim(conn, "t")
|
||||
seen = {}
|
||||
real = stub.append
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user