96 lines
4.3 KiB
Python
96 lines
4.3 KiB
Python
import json
|
||
|
||
import pytest
|
||
|
||
from sdlc_hub.core import db, files
|
||
from sdlc_hub.modules.jira import attachments, 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) == "размер неизвестен"
|
||
# M-1: десятые считаются в целых, округление half up; единица — по округлённому значению
|
||
assert attachments.fmt_size(1280) == "1,3 КБ"
|
||
assert attachments.fmt_size(1048575) == "1,0 МБ"
|
||
assert attachments.fmt_size(1310720) == "1,3 МБ"
|
||
|
||
|
||
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_ref_names_lone_exclamation():
|
||
"""I-1: одиночный «!» перед настоящей ссылкой не съедает её открывающий «!»."""
|
||
assert "скрин.png" in attachments.ref_names("Внимание! см. !скрин.png!")
|
||
names = attachments.ref_names("Важно! [^b.xlsx] и !скрин.png!")
|
||
assert names.index("b.xlsx") < names.index("скрин.png")
|
||
assert attachments.ref_names("{code}!a.png!{code} !b.png!")[-1] == "b.png"
|
||
assert attachments.ref_names("!http://x/y.png! !c.png!")[-1] == "c.png"
|
||
|
||
|
||
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
|