Команда fetch_attachment: постановка, выдача, приём результата (#83)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
1def5497e8
commit
90749012f0
@ -14,7 +14,7 @@
|
||||
| `mirror` | зеркало задач, связей, комментариев (время комментария в UTC — `created_at`) и списка вложений `issue_attachment` (схема v6; `check_attachments`, пишет `apply_state`; `attachment_file` ему не принадлежит), разрез по проекту (свои, соседи по связям, эпики своих), `apply_state` (с проверками `check_comments`, `check_attachments`), заметки, фильтры, снимок доски `board`, карточка `get_issue` (с историей задачи из `history`) | `db`, `history`, `attachments` |
|
||||
| `history` | журнал изменений задач `issue_change` (схема v5): проверка и запись пачек агента (`check_changes`, `record_changes`, логическое имя поля `logical_field`), параметр `history` для выдачи `refresh` (`delivery_param`: отметка — последний полный прогон с `sync_run.history_at`, минус 15 минут; стартовая JQL `backfill_jql`), признак принятого журнала `accepted`, лента `feed` (журнал + новые комментарии по `issue_comment.created_at`) и история задачи `issue_changes` | `db`, `settings`, `projects` |
|
||||
| `commands` | контракт типов команд, валидация, временные ключи, оценка, подтверждение по состоянию | `mirror` |
|
||||
| `queue` | пакеты, проверки постановки (`_check_setup`: `bug` без настройки, файл вложения; `_check_project`: проверки пакета проекта — метка, чужие метки, задачи вне проекта; `_assign_boards`: `board_id` проекта у `ensure_sprint`), выдача (`claim`, `size` у `ensure_attachment` и `boards` у `refresh` — `deliver_params`), отчёты, отмена, повтор, автообновление | `commands`, `mirror`, `settings`, `files`, `projects`, `db` |
|
||||
| `queue` | пакеты, проверки постановки (`_check_setup`: `bug` без настройки, файл вложения; `_check_project`: проверки пакета проекта — метка, чужие метки, задачи вне проекта; `_assign_boards`: `board_id` проекта у `ensure_sprint`), выдача (`claim`, `size` у `ensure_attachment` и `boards` у `refresh` — `deliver_params`), отчёты (для `fetch_attachment` в `done` — проверка файла в хранилище и запись `attachment_file`), `request_fetch` (постановка скачивания вложения в одной транзакции, `TooLarge`, `AgentTooOld`; пакет без хвостового `refresh` и без резервной копии), отмена, повтор, автообновление | `commands`, `mirror`, `settings`, `files`, `projects`, `attachments`, `db` |
|
||||
| `backup` | копии файла базы, ротация | `paths` |
|
||||
| `files` | хранилище файлов вложений `<данные>/files/<sha256>` (каталог — параметр, `Hub.files_dir`): `put` (атомарно, одна копия на содержимое, mtime при повторе, предел 10 МБ, пустой — отказ), `path`/`exists`/`size`/`read` (только sha из 64 hex нижнего регистра), `sweep` (старше 30 дней, кроме файлов `ensure_attachment` в `queued`/`claimed`) | `paths` |
|
||||
| `api` | HTTP-сервер, маршруты, авторизация по токенам, фоновый `tick` (автообновление и уборка `files/` раз в сутки); маршруты файлов `POST /files`, `GET /agent/files/{sha256}` (каталог `Hub.files_dir`, 0700 при старте); раздача веб-приложения `/app…` (каталог `web/app` пакета, без токена) и перенаправление `/`, `/links`; снимок `GET /board`; маршруты `/projects` | `queue`, `mirror`, `projects`, `settings`, `backup`, `files`, `relay`, `release` |
|
||||
@ -48,6 +48,7 @@
|
||||
- Вложение: файл → `POST /files` → `files/<sha256>` → `ensure_attachment {key, file, filename}` в пакете (`queue._check_setup` проверяет, что файл есть) → `claim` (хаб дописывает `size`, в базе `params` без него) → агент `GET /agent/files/{sha256}` → Jira; уборка `files/` — из `tick` раз в сутки.
|
||||
- Создание проекта: `POST /projects` (веб или `project_create`) → `projects.create` и пакет «Новый проект <метка>» (`refresh` по общей JQL, `batch.project = NULL`) в одной транзакции → `claim` (хаб дописывает в `refresh` доски проектов, `boards`) → агент определяет тип доски → отчёт `refresh` пишет `board_type` → `ensure_sprint` проекта получает `board_id`.
|
||||
- Чтение Jira: команда `refresh` → агент → `POST /agent/state` → `mirror.apply_state`.
|
||||
- Скачивание вложения: `queue.request_fetch` (одна транзакция: состояние `attachments.state` + пакет `fetch_attachment {key, attachment_id}`) → `claim` (без копии базы) → агент → `POST /agent/files` → результат `{file, size}` → `report_result` проверяет файл в `files/` и пишет `attachment_file`; нет файла или чужой размер — команда `failed`.
|
||||
- Журнал изменений: выдача `refresh` (`queue.deliver_params` дописывает `history` — `since` и/или стартовую `jql`) →
|
||||
агент v13 после состояния читает changelog Jira → `POST /agent/history` → `history.record_changes` (повторы
|
||||
отбрасываются по `(history_id, seq)`) → отчёт `refresh` с `result.history` → `sync_run.history_at` у полного прогона
|
||||
|
||||
@ -187,7 +187,7 @@ def h_agent_command(hub, conn, m, q, body):
|
||||
def h_agent_result(hub, conn, m, q, body):
|
||||
try:
|
||||
cmd = queue.report_result(conn, _int(m), body.get("status"), body.get("result"), body.get("error"),
|
||||
body.get("jira_calls", 0), body.get("rate_limited", 0))
|
||||
body.get("jira_calls", 0), body.get("rate_limited", 0), files_dir=hub.files_dir)
|
||||
except KeyError:
|
||||
raise ApiError(404, "not_found", "command")
|
||||
except queue.NotClaimedError as e:
|
||||
|
||||
@ -34,10 +34,12 @@ SPEC = {
|
||||
"ensure_attachment": {"required": ["key", "file", "filename"], "optional": [], "keys": ["key"],
|
||||
"hints": {"file": "sha256 из POST /files; в MCP — path к локальному файлу",
|
||||
"filename": "имя вложения"}},
|
||||
"fetch_attachment": {"required": ["key", "attachment_id"], "optional": [], "keys": ["key"],
|
||||
"hints": {"attachment_id": "id вложения из issue (цифры)"}},
|
||||
"refresh": {"required": [], "optional": ["keys", "jql"], "keys": ["keys"], "hints": {"keys": "список"},
|
||||
"note": "keys или jql"},
|
||||
}
|
||||
WRITE_TYPES = [t for t in SPEC if t != "refresh"]
|
||||
WRITE_TYPES = [t for t in SPEC if t not in ("refresh", "fetch_attachment")]
|
||||
LIST_PARAMS = {"labels", "labels_add", "labels_remove", "keys"}
|
||||
BOOL_PARAMS = {"replace"}
|
||||
|
||||
@ -190,7 +192,7 @@ def check_confirmed(conn, cmd, link_types: dict) -> tuple[bool, dict | None]:
|
||||
(bridge-reliability, решение 4); совпадение темы с задачей неизвестного автора не доказательство; вложений в
|
||||
зеркале нет (bug-and-attachments)."""
|
||||
ctype, p = cmd["type"], cmd["params"]
|
||||
if ctype in ("ensure_issue", "ensure_comment", "ensure_attachment", "refresh"):
|
||||
if ctype in ("ensure_issue", "ensure_comment", "ensure_attachment", "fetch_attachment", "refresh"):
|
||||
return False, None
|
||||
issue = mirror.get_issue(conn, p.get("key") or p.get("source"))
|
||||
if issue is None:
|
||||
|
||||
@ -7,7 +7,7 @@ from datetime import timedelta
|
||||
|
||||
from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim)
|
||||
from . import commands as C
|
||||
from . import db, files, history, mirror, projects, settings
|
||||
from . import attachments, db, files, history, mirror, projects, settings
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@ -15,6 +15,7 @@ TERMINAL = {"done", "failed", "skipped", "cancelled", "confirmed_by_state"}
|
||||
MAX_DELIVERIES = 3 # выдач без отчёта, после которых команда — failed (bridge-reliability, решение 2)
|
||||
ERROR_LIMIT = 1500 # символов ошибки в command.error (spec commands)
|
||||
ONLINE_MINUTES = 5
|
||||
READ_TYPES = {"refresh", "fetch_attachment"} # не пишут в Jira: без хвостового refresh и без резервной копии
|
||||
|
||||
|
||||
class ClaimedError(Exception):
|
||||
@ -23,6 +24,17 @@ class ClaimedError(Exception):
|
||||
self.claimed_at = claimed_at
|
||||
|
||||
|
||||
class TooLarge(C.CommandError):
|
||||
"""Вложение больше предела хаба (fetch_attachment) — API отвечает 413 too_large; для POST /batches это 422."""
|
||||
|
||||
|
||||
class AgentTooOld(ValueError):
|
||||
"""Агент ещё не умеет fetch_attachment — API отвечает 409 agent_too_old."""
|
||||
|
||||
|
||||
FETCH_AGENT_VERSION = 14 # первая версия агента с fetch_attachment
|
||||
|
||||
|
||||
class NotClaimedError(Exception):
|
||||
"""Отчёт по команде, которую агенту не выдавали (status = queued) — API отвечает 409."""
|
||||
|
||||
@ -67,6 +79,15 @@ def _check_setup(conn, cmds, files_dir) -> None:
|
||||
if '"' in p["filename"] or any(ord(ch) < 32 or ord(ch) == 127 for ch in p["filename"]):
|
||||
# кавычка, CR, LF ломают заголовок части multipart/form-data у агента (R-2)
|
||||
raise C.CommandError(f"команда {i}: filename без кавычек и управляющих символов (CR, LF и др.)")
|
||||
if c["type"] == "fetch_attachment":
|
||||
if not p["attachment_id"].isdigit():
|
||||
raise C.CommandError(f"команда {i}: attachment_id — цифры id вложения из issue")
|
||||
row = conn.execute("SELECT size FROM issue_attachment WHERE attachment_id = ? AND key = ?",
|
||||
(p["attachment_id"], p["key"])).fetchone()
|
||||
if row is None:
|
||||
raise C.CommandError(f"у {p['key']} нет вложения #{p['attachment_id']} в зеркале — обновите задачу")
|
||||
if row["size"] is not None and row["size"] > files.MAX_SIZE: # size NULL — размер неизвестен, ставим
|
||||
raise TooLarge(f"файл больше 10 МБ ({attachments.fmt_size(row['size'])}) — скачайте в Jira")
|
||||
_assign_boards(conn, cmds)
|
||||
|
||||
|
||||
@ -123,7 +144,7 @@ def _check_project(conn, cmds, project) -> list[dict]:
|
||||
elif t == "ensure_link":
|
||||
if not (own(p["source"]) or own(p["target"])):
|
||||
raise C.CommandError(f"команда {i}: связь {p['source']} → {p['target']} не касается проекта {project}")
|
||||
elif t != "refresh":
|
||||
elif t not in ("refresh", "fetch_attachment"): # fetch_attachment — не запись в Jira, проекта не касается
|
||||
if t == "ensure_fields":
|
||||
if project.casefold() in {l.casefold() for l in p.get("labels_remove") or []}:
|
||||
raise C.CommandError(f"команда {i}: снять метку своего проекта {project} нельзя")
|
||||
@ -150,8 +171,13 @@ def create_batch(conn, title, author, cmds, files_dir=None, project=None) -> dic
|
||||
keys = []
|
||||
for c in cmds:
|
||||
keys += [k for k in C.referenced_keys(c) if k not in keys]
|
||||
if any(c["type"] != "refresh" for c in cmds) and keys:
|
||||
if any(c["type"] not in READ_TYPES for c in cmds) and keys:
|
||||
cmds.append({"type": "refresh", "params": {"keys": keys}})
|
||||
return _insert_batch(conn, title, author, cmds, project)
|
||||
|
||||
|
||||
def _insert_batch(conn, title, author, cmds, project=None) -> dict:
|
||||
"""Вставка пакета без проверок и хвоста; транзакция своя (вложенная идёт через SAVEPOINT)."""
|
||||
est = C.estimate([c["type"] for c in cmds], settings.pacing(conn))
|
||||
with db.transaction(conn):
|
||||
cur = conn.execute("INSERT INTO batch(title, author, created_at, estimate_seconds, project) VALUES (?, ?, ?, ?, ?)",
|
||||
@ -167,6 +193,27 @@ def create_batch(conn, title, author, cmds, files_dir=None, project=None) -> dic
|
||||
return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)}
|
||||
|
||||
|
||||
def request_fetch(conn, files_dir, attachment_id: str, author: str | None) -> dict:
|
||||
"""Поставить скачивание вложения. Проверка «уже качается/скачано» и вставка пакета — в одной транзакции с
|
||||
блокировкой записи (BEGIN IMMEDIATE): два одновременных запроса дают одну команду (решение 7).
|
||||
Ответ — как attachments.get плюс command_id, batch_id. KeyError — вложения нет в зеркале; TooLarge, AgentTooOld."""
|
||||
attachment_id = str(attachment_id)
|
||||
with db.transaction(conn):
|
||||
info = attachments.get(conn, files_dir, attachment_id)
|
||||
if info is None:
|
||||
raise KeyError(attachment_id)
|
||||
if info["state"] in ("ready", "fetching"):
|
||||
return info
|
||||
agent = conn.execute("SELECT agent_version FROM agent_status WHERE id = 1").fetchone()
|
||||
v = agent["agent_version"] if agent else None
|
||||
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 = C.validate([{"type": "fetch_attachment", "params": {"key": info["key"], "attachment_id": attachment_id}}])
|
||||
_check_setup(conn, cmds, files_dir)
|
||||
_insert_batch(conn, f"Скачать вложение {info['filename']} ({info['key']})", author, cmds)
|
||||
return attachments.get(conn, files_dir, attachment_id)
|
||||
|
||||
|
||||
def _status_of(cmds, cancelled_at=None) -> str:
|
||||
sts = [c["status"] for c in cmds]
|
||||
if not sts:
|
||||
@ -304,7 +351,7 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non
|
||||
pause = int(settings.pacing(conn).get(cmd["type"], 0))
|
||||
first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'",
|
||||
(cmd["batch_id"],)).fetchone()["n"] == 0
|
||||
has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type != 'refresh' LIMIT 1",
|
||||
has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type NOT IN ('refresh', 'fetch_attachment') LIMIT 1",
|
||||
(cmd["batch_id"],)).fetchone() is not None
|
||||
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, pause_seconds = ?, params = ?, deliveries = 1 "
|
||||
"WHERE id = ?", (now, pause, _dump(cmd["params"]), cmd["id"]))
|
||||
@ -340,7 +387,7 @@ def record_state(conn, command_id, issues_count, links_count) -> None:
|
||||
(int(issues_count), int(links_count), command_id))
|
||||
|
||||
|
||||
def report_result(conn, command_id, status, result=None, error=None, jira_calls=0, rate_limited=0) -> dict:
|
||||
def report_result(conn, command_id, status, result=None, error=None, jira_calls=0, rate_limited=0, files_dir=None) -> dict:
|
||||
if status not in ("done", "failed"):
|
||||
raise ValueError("status — done или failed")
|
||||
error = None if error is None else str(error)[:ERROR_LIMIT]
|
||||
@ -363,12 +410,24 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls=
|
||||
return get_command(conn, command_id)
|
||||
return cmd # идемпотентно; терминальный статус (cancelled/failed) не затирается
|
||||
now = db.now()
|
||||
fetched = None
|
||||
if cmd["type"] == "fetch_attachment" and status == "done":
|
||||
# файл должен лежать в хранилище с заявленным размером; иначе отчёт не принимаем (уборка успела раньше, чужой размер)
|
||||
sha = result.get("file") if isinstance(result, dict) else None
|
||||
real = files.size(files_dir, sha) if files_dir is not None else None
|
||||
if real is None or real != result.get("size"):
|
||||
status, error = "failed", f"хаб: файл {sha} не найден в хранилище"[:ERROR_LIMIT]
|
||||
else:
|
||||
fetched = (cmd["params"]["attachment_id"], sha, real)
|
||||
conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?",
|
||||
(status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id))
|
||||
conn.execute("UPDATE agent_status SET current_command_id = NULL, last_seen_at = ? WHERE id = 1 AND current_command_id = ?",
|
||||
(now, command_id))
|
||||
if cmd["type"] == "ensure_issue" and status == "done" and result and result.get("key"):
|
||||
mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"])
|
||||
if fetched:
|
||||
conn.execute("INSERT OR REPLACE INTO attachment_file(attachment_id, file, size, fetched_at) VALUES (?, ?, ?, ?)",
|
||||
(*fetched, now))
|
||||
if cmd["type"] == "refresh":
|
||||
conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?",
|
||||
(now, int(jira_calls or 0), int(rate_limited or 0), command_id))
|
||||
|
||||
@ -5,7 +5,7 @@ import os
|
||||
import secrets
|
||||
|
||||
DEFAULT_PACING = {"ensure_issue": 120, "ensure_fields": 120, "ensure_comment": 30, "ensure_status": 30,
|
||||
"ensure_link": 30, "ensure_sprint": 30, "ensure_attachment": 30, "refresh": 2}
|
||||
"ensure_link": 30, "ensure_sprint": 30, "ensure_attachment": 30, "fetch_attachment": 2, "refresh": 2}
|
||||
|
||||
DEFAULTS = {
|
||||
"project_key": "AURORA",
|
||||
@ -136,7 +136,7 @@ def pacing_from_config(p: dict) -> dict:
|
||||
"""Старая форма pacing (issue_seconds / comment_seconds) → паузы по типам команд."""
|
||||
issue, comment = p.get("issue_seconds", 120), p.get("comment_seconds", 30)
|
||||
return {"ensure_issue": issue, "ensure_fields": issue, "ensure_comment": comment, "ensure_status": comment,
|
||||
"ensure_link": comment, "ensure_sprint": comment, "ensure_attachment": comment, "refresh": 2}
|
||||
"ensure_link": comment, "ensure_sprint": comment, "ensure_attachment": comment, "fetch_attachment": 2, "refresh": 2}
|
||||
|
||||
|
||||
def _issue_types(cfg: dict) -> dict:
|
||||
|
||||
@ -260,3 +260,10 @@ def test_validate_empty_list_params():
|
||||
commands.validate([{"type": "ensure_status", "params": {"key": "A-1", "target": []}}])
|
||||
with pytest.raises(CommandError, match="refresh без keys и jql"):
|
||||
commands.validate([{"type": "refresh", "params": {"keys": []}}])
|
||||
|
||||
|
||||
def test_fetch_not_confirmed_by_state(conn):
|
||||
ok, result = commands.check_confirmed(conn, {"type": "fetch_attachment", "params": {"key": "A-1", "attachment_id": "5"}}, LT)
|
||||
assert (ok, result) == (False, None)
|
||||
assert "fetch_attachment" not in commands.WRITE_TYPES
|
||||
assert commands.SPEC["fetch_attachment"]["required"] == ["key", "attachment_id"]
|
||||
|
||||
@ -2,7 +2,7 @@ import hashlib
|
||||
|
||||
import pytest
|
||||
|
||||
from jira_hub import db, files, mirror, projects, queue, settings
|
||||
from jira_hub import attachments, db, files, mirror, projects, queue, settings
|
||||
from jira_hub.commands import CommandError
|
||||
from jira_hub.queue import ClaimedError
|
||||
|
||||
@ -1028,3 +1028,158 @@ def test_report_sets_history_at(conn, add_project, result, status, marked):
|
||||
c = _full_run(conn, result, status=status)
|
||||
run = conn.execute("SELECT history_at FROM sync_run WHERE command_id = ?", (c["id"],)).fetchone()
|
||||
assert (run["history_at"] is not None) == marked
|
||||
|
||||
|
||||
# --- fetch_attachment (#83) ---------------------------------------------------
|
||||
|
||||
import threading
|
||||
|
||||
FSHA = "cd" * 32
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fconn(conn):
|
||||
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "labels": [], "attachments": [
|
||||
{"id": "31337", "filename": "ОСВ 60.xlsx", "size": 3, "created": "2026-10-20T10:00:00.000+0300"},
|
||||
{"id": "31338", "filename": "big.bin", "size": 25_000_000, "created": "2026-10-20T10:00:00.000+0300"},
|
||||
{"id": "31339", "filename": "unk.bin", "created": "2026-10-20T10:00:00.000+0300"}]},
|
||||
{"key": "B-1", "summary": "т", "labels": []}], [])
|
||||
return conn
|
||||
|
||||
|
||||
def fetch(att="31337", key="A-1"):
|
||||
return {"type": "fetch_attachment", "params": {"key": key, "attachment_id": att}}
|
||||
|
||||
|
||||
def test_fetch_batch_no_tail_refresh(fconn):
|
||||
b = queue.create_batch(fconn, "x", "claude", [fetch()])
|
||||
assert b["count"] == 1
|
||||
assert [c["type"] for c in queue.batch_view(fconn, b["id"])["commands"]] == ["fetch_attachment"]
|
||||
assert b["estimate_seconds"] == 2
|
||||
|
||||
|
||||
def test_fetch_checks(fconn):
|
||||
with pytest.raises(CommandError):
|
||||
queue.create_batch(fconn, "x", "claude", [fetch("1a")])
|
||||
with pytest.raises(CommandError) as e:
|
||||
queue.create_batch(fconn, "x", "claude", [fetch(key="B-1")])
|
||||
assert str(e.value) == "у B-1 нет вложения #31337 в зеркале — обновите задачу"
|
||||
with pytest.raises(queue.TooLarge) as e:
|
||||
queue.create_batch(fconn, "x", "claude", [fetch("31338")])
|
||||
assert str(e.value) == "файл больше 10 МБ (23,8 МБ) — скачайте в Jira"
|
||||
assert queue.create_batch(fconn, "x", "claude", [fetch("31339")])["count"] == 1 # size NULL — ставится
|
||||
|
||||
|
||||
def test_fetch_no_backup(fconn):
|
||||
queue.create_batch(fconn, "x", "claude", [fetch()])
|
||||
calls = []
|
||||
cmd = queue.claim(fconn, backup=calls.append)
|
||||
assert cmd["type"] == "fetch_attachment" and cmd["pause_seconds"] == 2 and calls == []
|
||||
|
||||
|
||||
def test_fetch_project_batch_no_checks(fconn, add_project):
|
||||
add_project(fconn, "AWPCA")
|
||||
b = queue.create_batch(fconn, "x", "claude", [fetch(key="A-1")], project="AWPCA") # A-1 не в проекте
|
||||
assert b["count"] == 1
|
||||
|
||||
|
||||
def test_request_fetch_once(open_db, tmp_path):
|
||||
path = tmp_path / "hub.sqlite"
|
||||
c1 = open_db(path)
|
||||
settings.ensure_defaults(c1)
|
||||
mirror.apply_state(c1, [{"key": "A-1", "summary": "т", "labels": [], "attachments": [
|
||||
{"id": "31337", "filename": "a.png", "size": 3, "created": "2026-10-20T10:00:00.000+0300"}]}], [])
|
||||
c2 = open_db(path)
|
||||
res, errs = [], []
|
||||
barrier = threading.Barrier(2)
|
||||
|
||||
def run(c):
|
||||
try:
|
||||
barrier.wait()
|
||||
res.append(queue.request_fetch(c, tmp_path / "files", "31337", "x"))
|
||||
except Exception as e: # noqa: BLE001
|
||||
errs.append(e)
|
||||
|
||||
ts = [threading.Thread(target=run, args=(c,)) for c in (c1, c2)]
|
||||
[t.start() for t in ts]
|
||||
[t.join() for t in ts]
|
||||
assert not errs
|
||||
assert res[0]["command_id"] == res[1]["command_id"] and res[0]["batch_id"] == res[1]["batch_id"]
|
||||
assert {r["state"] for r in res} <= {"fetching"}
|
||||
assert c1.execute("SELECT COUNT(*) AS n FROM command WHERE type = 'fetch_attachment'").fetchone()["n"] == 1
|
||||
|
||||
|
||||
def put_file(tmp_path, data=b"abc"):
|
||||
root = tmp_path / "files"
|
||||
sha, _ = files.put(root, data)
|
||||
return root, sha
|
||||
|
||||
|
||||
def test_request_fetch_ready(fconn, tmp_path):
|
||||
root, sha = put_file(tmp_path)
|
||||
fconn.execute("INSERT INTO attachment_file(attachment_id, file, size, fetched_at) VALUES ('31337', ?, 3, ?)", (sha, db.now()))
|
||||
r = queue.request_fetch(fconn, root, "31337", "x")
|
||||
assert r["state"] == "ready" and r["filename"] == "ОСВ 60.xlsx"
|
||||
assert fconn.execute("SELECT COUNT(*) AS n FROM command").fetchone()["n"] == 0
|
||||
with pytest.raises(KeyError):
|
||||
queue.request_fetch(fconn, root, "999", "x")
|
||||
|
||||
|
||||
def test_request_fetch_agent_too_old(fconn, tmp_path):
|
||||
fconn.execute("UPDATE agent_status SET agent_version = '13' WHERE id = 1")
|
||||
with pytest.raises(queue.AgentTooOld) as e:
|
||||
queue.request_fetch(fconn, tmp_path / "files", "31337", "x")
|
||||
assert str(e.value) == "агент v13 не умеет скачивать вложения — нужен v14"
|
||||
fconn.execute("UPDATE agent_status SET agent_version = '14' WHERE id = 1")
|
||||
assert queue.request_fetch(fconn, tmp_path / "files", "31337", "x")["state"] == "fetching"
|
||||
fconn.execute("UPDATE agent_status SET agent_version = NULL WHERE id = 1")
|
||||
assert queue.request_fetch(fconn, tmp_path / "files", "31337", "x")["state"] == "fetching"
|
||||
|
||||
|
||||
def test_request_fetch_after_failed(fconn, tmp_path):
|
||||
root = tmp_path / "files"
|
||||
first = queue.request_fetch(fconn, root, "31337", "x")
|
||||
cmd = queue.claim(fconn)
|
||||
queue.report_result(fconn, cmd["id"], "failed", error="нет сети", files_dir=root)
|
||||
again = queue.request_fetch(fconn, root, "31337", "x")
|
||||
assert again["state"] == "fetching" and again["batch_id"] != first["batch_id"]
|
||||
|
||||
|
||||
def report_done(conn, root, result):
|
||||
queue.create_batch(conn, "x", "claude", [fetch()])
|
||||
cmd = queue.claim(conn)
|
||||
return queue.report_result(conn, cmd["id"], "done", result=result, files_dir=root)
|
||||
|
||||
|
||||
def test_report_fetch_done_records_file(fconn, tmp_path):
|
||||
root, sha = put_file(tmp_path)
|
||||
cmd = report_done(fconn, root, {"file": sha, "size": 3})
|
||||
assert cmd["status"] == "done"
|
||||
row = fconn.execute("SELECT * FROM attachment_file WHERE attachment_id = '31337'").fetchone()
|
||||
assert row["file"] == sha and row["size"] == 3
|
||||
assert attachments.state(fconn, root, "31337")["state"] == "ready"
|
||||
|
||||
|
||||
def test_report_fetch_done_missing_file(fconn, tmp_path):
|
||||
root = tmp_path / "files"
|
||||
cmd = report_done(fconn, root, {"file": FSHA, "size": 3})
|
||||
assert cmd["status"] == "failed" and cmd["error"] == f"хаб: файл {FSHA} не найден в хранилище"
|
||||
assert fconn.execute("SELECT COUNT(*) AS n FROM attachment_file").fetchone()["n"] == 0
|
||||
# после уборки файла (done → файла нет) и повторной постановки — новый пакет
|
||||
assert queue.request_fetch(fconn, root, "31337", "x")["state"] == "fetching"
|
||||
|
||||
|
||||
def test_report_fetch_wrong_size(fconn, tmp_path):
|
||||
root, sha = put_file(tmp_path)
|
||||
cmd = report_done(fconn, root, {"file": sha, "size": 99})
|
||||
assert cmd["status"] == "failed" and cmd["error"] == f"хаб: файл {sha} не найден в хранилище"
|
||||
assert fconn.execute("SELECT COUNT(*) AS n FROM attachment_file").fetchone()["n"] == 0
|
||||
|
||||
|
||||
def test_request_fetch_after_done_file_swept(fconn, tmp_path):
|
||||
root, sha = put_file(tmp_path)
|
||||
report_done(fconn, root, {"file": sha, "size": 3})
|
||||
files.path(root, sha).unlink() # уборка файла
|
||||
r = queue.request_fetch(fconn, root, "31337", "x")
|
||||
assert r["state"] == "fetching"
|
||||
assert fconn.execute("SELECT COUNT(*) AS n FROM command WHERE type = 'fetch_attachment'").fetchone()["n"] == 2
|
||||
|
||||
@ -119,3 +119,9 @@ def test_closed_statuses_must_be_list():
|
||||
with pytest.raises(ValueError):
|
||||
settings.validate("closed_statuses", "Готово")
|
||||
settings.validate("closed_statuses", ["Готово"])
|
||||
|
||||
|
||||
def test_pacing_fallback_for_new_type(conn):
|
||||
settings.set_(conn, "pacing", {"ensure_fields": 5})
|
||||
assert settings.pacing(conn)["fetch_attachment"] == 2
|
||||
settings.validate("pacing", {"fetch_attachment": 3})
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user