b5adfdf9b5
Модуль attachments (fmt_size, ref_names, resolve, state, get); mirror.get_issue отдаёт attachments, description_attachment_ids, attachment_ids комментариев и скрывает fetch_attachment из commands. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
82 lines
3.4 KiB
Python
82 lines
3.4 KiB
Python
import json
|
||
|
||
import pytest
|
||
|
||
from jira_hub import attachments, db, files, mirror
|
||
|
||
|
||
def test_fmt_size():
|
||
assert attachments.fmt_size(512) == "512 Б"
|
||
assert attachments.fmt_size(240000) == "234,4 КБ"
|
||
assert attachments.fmt_size(25_000_000) == "23,8 МБ"
|
||
assert attachments.fmt_size(None) == "размер неизвестен"
|
||
|
||
|
||
def test_ref_names():
|
||
text = "см. !скрин 1.png|thumbnail! и [^ОСВ 60.xlsx], ещё раз !скрин 1.png!"
|
||
assert attachments.ref_names(text) == ["скрин 1.png", "ОСВ 60.xlsx"]
|
||
assert attachments.ref_names("[отчёт|^a.pdf]") == ["a.pdf"]
|
||
assert attachments.ref_names("!http://x/y.png!") == []
|
||
assert attachments.ref_names(None) == []
|
||
|
||
|
||
def test_resolve_newest():
|
||
items = [{"attachment_id": "10", "filename": "скрин.png", "created_at": "2026-10-14T10:00:00+00:00"},
|
||
{"attachment_id": "20", "filename": "скрин.png", "created_at": "2026-10-20T10:00:00+00:00"}]
|
||
assert attachments.resolve(["скрин.png", "нет.png", "скрин.png"], items) == ["20"]
|
||
|
||
|
||
@pytest.fixture
|
||
def conn(open_db):
|
||
c = open_db()
|
||
mirror.apply_state(c, [{"key": "A-1", "summary": "т", "labels": [], "attachments": [
|
||
{"id": "31337", "filename": "a.png", "size": 3, "created": "2026-10-20T10:00:00.000+0300"}]}], [])
|
||
return c
|
||
|
||
|
||
def add_cmd(conn, status, error=None, att="31337"):
|
||
conn.execute("INSERT OR IGNORE INTO batch(id, title, author, created_at) VALUES (1, 't', 'a', ?)", (db.now(),))
|
||
cur = conn.execute("INSERT INTO command(batch_id, seq, type, params, status, error) VALUES (1, (SELECT COUNT(*) FROM command), "
|
||
"'fetch_attachment', ?, ?, ?)", (json.dumps({"key": "A-1", "attachment_id": att}), status, error))
|
||
return cur.lastrowid
|
||
|
||
|
||
def test_state_none(conn, tmp_path):
|
||
assert attachments.state(conn, tmp_path, "31337") == {"state": "none", "error": None, "command_id": None, "batch_id": None}
|
||
|
||
|
||
@pytest.mark.parametrize("status", ["queued", "claimed"])
|
||
def test_state_fetching(conn, tmp_path, status):
|
||
cid = add_cmd(conn, status)
|
||
assert attachments.state(conn, tmp_path, "31337") == {"state": "fetching", "error": None, "command_id": cid, "batch_id": 1}
|
||
|
||
|
||
def put_ready(conn, root):
|
||
sha, size = files.put(root, b"abc")
|
||
conn.execute("INSERT INTO attachment_file(attachment_id, file, size, fetched_at) VALUES ('31337', ?, ?, ?)",
|
||
(sha, size, db.now()))
|
||
return sha
|
||
|
||
|
||
def test_state_ready_and_removed(conn, tmp_path):
|
||
sha = put_ready(conn, tmp_path)
|
||
add_cmd(conn, "done")
|
||
assert attachments.state(conn, tmp_path, "31337")["state"] == "ready"
|
||
assert attachments.get(conn, tmp_path, "31337")["filename"] == "a.png"
|
||
(tmp_path / sha).unlink()
|
||
assert attachments.state(conn, tmp_path, "31337")["state"] == "none"
|
||
assert attachments.state(conn, None, "31337")["state"] == "none"
|
||
|
||
|
||
def test_state_failed_and_cancelled(conn, tmp_path):
|
||
add_cmd(conn, "failed", "нет доступа")
|
||
assert attachments.state(conn, tmp_path, "31337")["state"] == "failed"
|
||
assert attachments.state(conn, tmp_path, "31337")["error"] == "нет доступа"
|
||
add_cmd(conn, "cancelled", "отменена")
|
||
got = attachments.state(conn, tmp_path, "31337")
|
||
assert got["state"] == "failed" and got["error"] == "отменена"
|
||
|
||
|
||
def test_get_missing(conn, tmp_path):
|
||
assert attachments.get(conn, tmp_path, "999") is None
|