sdlc-hub/tests/test_queue.py
Andrey Epifancev bd8c769e75 feat(queue,mcp): отмена пакета сообщает о команде у агента; подсказка про labels (#13)
cancel_batch возвращает still_claimed — команды пакета в статусе claimed,
которые отмена не останавливает. MCP batch_cancel выводит строку на каждую
такую команду с подсказкой про POST /commands/N/cancel с force. Описание
batch_create предупреждает, что ensure_issue нужны labels с базовой меткой
AWPCA, иначе задача выпадет из базовой JQL и из зеркала.

M-11 (hub#12), T3.7 (agent#4).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 12:51:51 +04:00

468 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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"]
again = queue.claim(conn) # ответ потерян — та же команда снова
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"]
done = queue.report_result(conn, c["id"], "done", result={"changed": ["duedate"]}, jira_calls=2)
assert done["status"] == "done" and done["jira_calls"] == 2
again = queue.report_result(conn, c["id"], "failed", error="поздно")
assert again["status"] == "done" # повторный отчёт не меняет
tail = queue.claim(conn)
assert tail["type"] == "refresh" and tail["pause_seconds"] == 2
assert conn.execute("SELECT kind, query FROM sync_run WHERE command_id = ?", (tail["id"],)).fetchone()["query"] == "A-1"
queue.record_state(conn, tail["id"], 1, 0)
queue.report_result(conn, tail["id"], "done", jira_calls=1)
run = conn.execute("SELECT * FROM sync_run WHERE command_id = ?", (tail["id"],)).fetchone()
assert run["finished_at"] and run["issues_count"] == 1 and run["jira_calls"] == 1
assert queue.batch_view(conn, b1["id"])["status"] == "done"
assert queue.claim(conn)["params"]["key"] == "A-2" and queue.batch_view(conn, b2["id"])["status"] == "running"
def test_claim_is_exclusive(tmp_path):
import threading
path = tmp_path / "hub.sqlite"
setup = db.connect(path)
settings.ensure_defaults(setup)
queue.create_batch(setup, "x", "claude", [fields("A-1")])
conns = [db.connect(path), db.connect(path)]
barrier, results = threading.Barrier(2), []
def worker(c):
barrier.wait()
results.append(queue.claim(c))
threads = [threading.Thread(target=worker, args=(c,)) for c in conns]
[t.start() for t in threads]
[t.join() for t in threads]
got = [r for r in results if r is not None]
assert got and len({r["id"] for r in got}) == 1 # один и тот же id при любом порядке потоков
row = setup.execute("SELECT COUNT(*) AS n, MAX(deliveries) AS d FROM command WHERE status = 'claimed'").fetchone()
assert row["n"] == 1 and row["d"] == len(got)
def test_backup_failure_does_not_block_claim(caplog):
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
def broken(label):
raise OSError("диск полон")
c = queue.claim(conn, backup=broken)
assert c is not None and c["params"]["key"] == "A-1"
assert queue.get_command(conn, c["id"])["status"] == "claimed"
assert "не создана" in caplog.text
def test_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)["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": []})
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)
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"]))
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 queue.batch_view(conn, b["id"])["status"] == "done"
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_rotation_keeps_schema_migration_copy(tmp_path):
"""Копия базы перед миграцией схемы не участвует в ротации копий пакетов (M-1 финального ревью)."""
from jira_hub import backup
conn = db.connect(tmp_path / "hub.sqlite")
backup.make_backup(conn, tmp_path / "backups", "schema-v1", keep=0)
for i in range(4):
backup.make_backup(conn, tmp_path / "backups", f"batch-{i}", keep=2)
names = sorted(p.name for p in (tmp_path / "backups").glob("*.sqlite"))
assert len([n for n in names if n.endswith("-schema-v1.sqlite")]) == 1
assert len([n for n in names if "-batch-" in n]) == 2
def _claimed_ensure_issue(conn, summary="Тема"):
queue.create_batch(conn, "x", "claude", [{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": summary}}])
c = queue.claim(conn)
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
return c
def test_reconcile_does_not_confirm_ensure_issue_without_reporter():
"""ensure_issue по состоянию не подтверждается: совпадение темы не закрывает команду, заметка new:N не переезжает."""
conn = fresh()
c = _claimed_ensure_issue(conn)
mirror.set_note(conn, "new:1", {"work_kind": "разработка"}, "t")
created = db.utcnow().isoformat()
mirror.apply_state(conn, [{"key": "A-7", "summary": "Тема", "reporter": "other", "created": created, "labels": []}], [])
assert queue.reconcile(conn) == []
assert queue.get_command(conn, c["id"])["status"] == "claimed"
assert mirror.get_note(conn, "new:1") is not None and mirror.get_note(conn, "A-7") is None
def test_reconcile_does_not_overwrite_finished_command(monkeypatch):
"""M-1: отчёт агента/отмена между check_confirmed и UPDATE не перезаписываются."""
conn = fresh()
queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}])
c = queue.claim(conn)
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "sprint": "Спринт 2", "labels": []}], [])
real = queue.C.check_confirmed
def racing(*a, **kw):
out = real(*a, **kw)
queue.cancel_command(conn, c["id"], force=True) # гонка: отмена после проверки
return out
monkeypatch.setattr(queue.C, "check_confirmed", racing)
assert queue.reconcile(conn) == []
assert queue.get_command(conn, c["id"])["status"] == "cancelled"
def _two_batches_with_dependent(conn):
a = queue.create_batch(conn, "A", "claude", [
{"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}},
{"type": "ensure_link", "params": {"source": "new:1", "target": "A-1", "link_type": "связанные"}}])
first = queue.claim(conn)
assert first["type"] == "ensure_issue"
b = queue.create_batch(conn, "B", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
return a, b, first
def test_redelivered_definer_blocks_dependents():
conn = fresh()
a, b, first = _two_batches_with_dependent(conn)
assert queue.claim(conn)["id"] == first["id"] # довыдача определителя, зависимые ждут
conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],))
nxt = queue.claim(conn)
assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9"
def test_auto_refresh_waits_for_claimed_without_report():
conn = fresh()
queue.claim(conn, agent_version=1, settings_version=1) # агент на связи
queue.create_batch(conn, "y", "claude", [fields("A-3")])
queue.claim(conn) # выдана, отчёта нет
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 queue.maybe_auto_refresh(conn) is None # claimed без отчёта — очередь занята
def test_redelivery_of_lost_claim_before_queue():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
c = queue.claim(conn)
again = queue.claim(conn)
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 [(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"] == []
def test_redelivery_limit_fails_command():
conn = fresh()
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 и дальше по очереди
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"]
assert queue.MAX_DELIVERIES == 3
def test_redelivery_oldest_first():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
queue.create_batch(conn, "y", "claude", [fields("A-2")])
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"]
def test_redelivered_refresh_keeps_one_sync_run():
conn = fresh()
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
assert queue.claim(conn)["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 = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
labels = []
queue.claim(conn, backup=labels.append)
queue.claim(conn, backup=labels.append)
assert len(labels) == 1
def test_migrated_claimed_is_redelivered():
"""Команда, взятая до миграции (deliveries = 1 после неё), довыдаётся со счётчиком 2."""
conn = fresh()
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.get_command(conn, cid)["deliveries"] == 2
def test_report_for_queued_command_rejected():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
cid = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields'").fetchone()["id"]
with pytest.raises(queue.NotClaimedError):
queue.report_result(conn, cid, "done")
assert queue.get_command(conn, cid)["status"] == "queued"
def test_late_report_after_force_cancel_keeps_issue_key():
conn = fresh()
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)
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
assert late["finished_at"] == cancelled["finished_at"]
assert mirror.get_note(conn, "A-20")["fact_status"] == "В работе" and mirror.get_note(conn, "new:7") is None
again = queue.report_result(conn, c["id"], "done", result={"key": "A-21"})
assert again["result"] == {"key": "A-20", "created": True} # второй поздний отчёт ничего не меняет
def test_late_failed_report_after_force_cancel_changes_nothing():
conn = fresh()
queue.create_batch(conn, "x", "claude", [
{"type": "ensure_issue", "params": {"temp_key": "new:7", "issue_type": "task", "summary": "Тема"}}])
c = queue.claim(conn)
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
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.get_command(conn, cid)["status"] == "failed"
def test_error_is_trimmed_to_1500():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
c = queue.claim(conn)
out = queue.report_result(conn, c["id"], "failed", error="е" * 5000)
assert out["error"] == "е" * 1500 and queue.ERROR_LIMIT == 1500
def test_late_done_after_delivery_limit_keeps_issue_key():
conn = fresh()
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)
_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}
assert late["error"] == "агент не отчитался после 3 выдач"
assert mirror.get_note(conn, "A-30")["fact_status"] == "План" and mirror.get_note(conn, "new:8") is None
again = queue.report_result(conn, c["id"], "done", result={"key": "A-31"})
assert again["result"] == {"key": "A-30", "created": True} # повтор ничего не меняет
def test_cancelled_batch_with_done_command_is_cancelled():
conn = fresh()
b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2"), fields("A-3")])
first = queue.claim(conn)
queue.report_result(conn, first["id"], "done", result={"changed": []})
second = queue.claim(conn)
queue.cancel_batch(conn, b["id"])
queue.report_result(conn, second["id"], "done", result={"changed": []})
view = queue.batch_view(conn, b["id"])
assert view["status"] == "cancelled"
assert view["counts"]["done"] == 2
def test_empty_full_refresh_keeps_neighbors():
conn = fresh()
jql = settings.get(conn, "base_jql")
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)
queue.report_result(conn, r["id"], "done") # загрузок состояния не было: 0 задач
assert conn.execute("SELECT neighbor FROM issue WHERE key = 'A-1'").fetchone()["neighbor"] == 0
def test_redelivered_refresh_resets_counts():
conn = fresh()
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
queue.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"] # отчёт потерян — довыдача
queue.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)
assert run["started_at"] == started
def test_refresh_failed_by_limit_finishes_sync_run():
conn = fresh()
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
for _ in range(queue.MAX_DELIVERIES):
queue.claim(conn)
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 = fresh()
b = queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
c = queue.claim(conn)
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"