407 lines
23 KiB
Python
407 lines
23 KiB
Python
import pytest
|
||
|
||
from sdlc_hub.core import db
|
||
from sdlc_hub.modules.jira import mirror
|
||
|
||
|
||
@pytest.fixture
|
||
def conn(open_db):
|
||
"""Пустое зеркало в памяти; закрывается после теста."""
|
||
return open_db()
|
||
|
||
|
||
def issue(key, **over):
|
||
base = {"key": key, "id": "1", "issue_type": "Задача", "summary": f"Тема {key}", "description": "h3. Контекст",
|
||
"status": "Ожидание разработки", "priority": "Средний", "assignee": None, "reporter": "me",
|
||
"duedate": "2026-10-23", "planned_start": None, "planned_end": "2026-10-23", "epic_key": "AURORA-1350",
|
||
"labels": ["AWPCA"], "sprint": "Спринт 2", "created": "2026-10-10T12:00:00.000+0300",
|
||
"jira_updated": "2026-10-11T12:00:00.000+0300"}
|
||
base.update(over)
|
||
return base
|
||
|
||
|
||
def test_apply_state_upserts_and_replaces_links(conn):
|
||
links = [{"source_key": "A-1", "target_key": "A-2", "link_type": "ASUPR_Блокер_Начала"}]
|
||
out = mirror.apply_state(conn, [issue("A-1"), issue("A-2")], links)
|
||
assert out == {"issues": 2, "links": 1}
|
||
mirror.apply_state(conn, [issue("A-1", summary="Новая тема")],
|
||
[{"source_key": "A-1", "target_key": "A-2", "link_type": "ASUPR_Связанные"}])
|
||
got = mirror.get_issue(conn, "A-1")
|
||
assert got["summary"] == "Новая тема" and got["labels"] == ["AWPCA"]
|
||
assert got["links_out"] == [{"target_key": "A-2", "link_type": "ASUPR_Связанные", "summary": "Тема A-2",
|
||
"status": "Ожидание разработки"}]
|
||
assert mirror.get_issue(conn, "A-2")["links_in"] == [{"source_key": "A-1", "link_type": "ASUPR_Связанные",
|
||
"summary": "Новая тема", "status": "Ожидание разработки"}]
|
||
|
||
|
||
def test_links_from_other_source_survive(conn):
|
||
mirror.apply_state(conn, [issue("A-1"), issue("A-2")],
|
||
[{"source_key": "A-2", "target_key": "A-1", "link_type": "ASUPR_Связанные"}])
|
||
mirror.apply_state(conn, [issue("A-1")], []) # пришла только A-1 без связей
|
||
assert mirror.get_issue(conn, "A-1")["links_in"] == [{"source_key": "A-2", "link_type": "ASUPR_Связанные",
|
||
"summary": "Тема A-2", "status": "Ожидание разработки"}]
|
||
|
||
|
||
def test_apply_state_marks_neighbors(conn):
|
||
mirror.apply_state(conn, [issue("A-1"), issue("B-9")], [], neighbor_keys={"B-9"})
|
||
assert mirror.get_issue(conn, "B-9")["neighbor"] is True
|
||
assert mirror.get_issue(conn, "A-1")["neighbor"] is False
|
||
mirror.apply_state(conn, [issue("B-9")], []) # вошла в базовую JQL
|
||
assert mirror.get_issue(conn, "B-9")["neighbor"] is False
|
||
assert mirror.known_keys(conn) == ["A-1", "B-9"]
|
||
|
||
|
||
def test_partial_load_never_lowers_neighbor(conn):
|
||
mirror.apply_state(conn, [issue("B-9")], [], neighbor_keys={"B-9"})
|
||
mirror.apply_state(conn, [issue("B-9", summary="новая")], [], authoritative=False) # refresh {keys}
|
||
assert mirror.get_issue(conn, "B-9")["neighbor"] is True
|
||
mirror.apply_state(conn, [issue("B-9")], [], authoritative=True) # полный прогон: вошла в базовую
|
||
assert mirror.get_issue(conn, "B-9")["neighbor"] is False
|
||
|
||
|
||
def test_note_survives_refresh_and_renames(conn):
|
||
mirror.set_note(conn, "new:7", {"fact_status": "В работе", "work_kind": "разработка"}, "import")
|
||
mirror.rename_note(conn, "new:7", "A-5")
|
||
mirror.apply_state(conn, [issue("A-5")], [])
|
||
got = mirror.get_issue(conn, "A-5")
|
||
assert got["note"]["fact_status"] == "В работе" and got["note"]["updated_by"] == "import"
|
||
assert mirror.get_note(conn, "new:7") is None
|
||
with pytest.raises(ValueError):
|
||
mirror.set_note(conn, "A-5", {"summary": "нельзя"}, "x")
|
||
|
||
|
||
def test_list_issues_filters(conn):
|
||
mirror.apply_state(conn, [issue("A-1", sprint="Спринт 2", assignee="ivanov", duedate="2026-10-20"),
|
||
issue("A-2", sprint="Спринт 1", labels=["AWPCA", "fact-done"]),
|
||
issue("A-3", sprint="Спринт 2", duedate=None)], [])
|
||
mirror.set_note(conn, "A-3", {"work_kind": "тестирование"}, "x")
|
||
keys = [i["key"] for i in mirror.list_issues(conn, {"sprint": "Спринт 2"})]
|
||
assert keys == ["A-1", "A-3"] # без срока — последней
|
||
assert [i["key"] for i in mirror.list_issues(conn, {"label": "fact-done"})] == ["A-2"]
|
||
assert [i["key"] for i in mirror.list_issues(conn, {"work_kind": "тестирование"})] == ["A-3"]
|
||
assert [i["key"] for i in mirror.list_issues(conn, {"q": "A-2"})] == ["A-2"]
|
||
assert mirror.list_issues(conn, {"assignee": "ivanov"})[0]["work_kind"] is None
|
||
assert len(mirror.list_issues(conn, None, limit=1)) == 1
|
||
|
||
|
||
def test_rename_note_same_key_keeps_note(conn):
|
||
mirror.apply_state(conn, [issue("A-1")], [])
|
||
mirror.set_note(conn, "A-1", {"fact_status": "В работе"}, "x")
|
||
mirror.rename_note(conn, "A-1", "A-1")
|
||
assert mirror.get_note(conn, "A-1")["fact_status"] == "В работе"
|
||
|
||
|
||
def test_rename_note_missing_is_noop(conn):
|
||
mirror.rename_note(conn, "new:404", "A-9") # заметки нет — не падает
|
||
assert mirror.get_note(conn, "new:404") is None
|
||
assert mirror.get_note(conn, "A-9") is None
|
||
|
||
|
||
def test_set_note_partial_keeps_other_fields(conn):
|
||
mirror.set_note(conn, "A-1", {"work_kind": "разработка", "done_text": "сделано"}, "x")
|
||
got = mirror.set_note(conn, "A-1", {"fact_status": "Готово"}, "y")
|
||
assert got["work_kind"] == "разработка" and got["done_text"] == "сделано"
|
||
assert got["fact_status"] == "Готово" and got["updated_by"] == "y"
|
||
|
||
|
||
def test_rename_note_merges_with_existing(conn):
|
||
mirror.set_note(conn, "A-5", {"owner_note": "спросить ИБ", "work_kind": "Старое"}, "human")
|
||
mirror.set_note(conn, "new:5", {"work_kind": "Аналитика", "fact_status": "В работе"}, "import")
|
||
mirror.rename_note(conn, "new:5", "A-5")
|
||
note = mirror.get_note(conn, "A-5")
|
||
assert note["work_kind"] == "Аналитика" # непустое поле временной заметки побеждает
|
||
assert note["owner_note"] == "спросить ИБ" # пустое у временной — остаётся под ключом
|
||
assert note["fact_status"] == "В работе"
|
||
assert note["updated_by"] == "import" # временная заметка записана позже
|
||
assert mirror.get_note(conn, "new:5") is None
|
||
|
||
|
||
def test_rename_note_keeps_newer_real_key_stamp(conn):
|
||
mirror.set_note(conn, "new:6", {"fact_status": "В работе"}, "import")
|
||
mirror.set_note(conn, "A-6", {"owner_note": "позже"}, "human")
|
||
conn.execute("UPDATE issue_note SET updated_at = '2020-01-01T00:00:00+00:00' WHERE key = 'new:6'")
|
||
conn.execute("UPDATE issue_note SET updated_at = '2030-01-01T00:00:00+00:00' WHERE key = 'A-6'")
|
||
mirror.rename_note(conn, "new:6", "A-6")
|
||
note = mirror.get_note(conn, "A-6")
|
||
assert note["updated_by"] == "human" and note["updated_at"] == "2030-01-01T00:00:00+00:00"
|
||
assert note["fact_status"] == "В работе" and note["owner_note"] == "позже"
|
||
|
||
|
||
def test_rename_note_empty_string_does_not_overwrite(conn):
|
||
mirror.set_note(conn, "A-6", {"rest_text": "доделать отчёт"}, "human")
|
||
mirror.set_note(conn, "new:6", {"rest_text": "", "done_text": "сделано"}, "import")
|
||
mirror.rename_note(conn, "new:6", "A-6")
|
||
note = mirror.get_note(conn, "A-6")
|
||
assert note["rest_text"] == "доделать отчёт" and note["done_text"] == "сделано"
|
||
|
||
|
||
BASE = "project = AURORA AND labels = AWPCA"
|
||
|
||
|
||
def _sync_run(conn, started, finished, kind, query, status="done", count=5):
|
||
"""Строка журнала синхронизаций с командой refresh в нужном статусе."""
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('t', 'test', ?)", (started,))
|
||
batch_id = conn.execute("SELECT last_insert_rowid() AS i").fetchone()["i"]
|
||
conn.execute("INSERT INTO command(batch_id, seq, type, params, status, finished_at) VALUES (?, 1, 'refresh', '{}', ?, ?)",
|
||
(batch_id, status, finished))
|
||
command_id = conn.execute("SELECT last_insert_rowid() AS i").fetchone()["i"]
|
||
conn.execute("INSERT INTO sync_run(started_at, finished_at, kind, query, issues_count, command_id) VALUES (?, ?, ?, ?, ?, ?)",
|
||
(started, finished, kind, query, count, command_id))
|
||
|
||
|
||
def test_board_snapshot_raw_mirror_data(conn):
|
||
mirror.apply_state(conn, [issue("AURORA-1351", issue_type="Фича", epic_key=None),
|
||
issue("AURORA-1620", labels=["kind-dev"], epic_key="AURORA-1351"),
|
||
issue("AURORA-1621", epic_key="AURORA-1351")],
|
||
[{"source_key": "AURORA-1620", "target_key": "AURORA-1621", "link_type": "ASUPR_Блокер_Начала"}],
|
||
neighbor_keys=["AURORA-1621"])
|
||
mirror.set_note(conn, "AURORA-1620", {"work_kind": "разработка"}, "test")
|
||
snap = mirror.board(conn, BASE)
|
||
assert [i["key"] for i in snap["issues"]] == ["AURORA-1351", "AURORA-1620", "AURORA-1621"]
|
||
a = snap["issues"][1]
|
||
assert set(a) == {"key", "summary", "issue_type", "status", "priority", "assignee", "sprint", "duedate",
|
||
"planned_start", "planned_end", "epic_key", "labels", "neighbor", "fact_status", "work_kind"}
|
||
assert a["labels"] == ["kind-dev"] and a["epic_key"] == "AURORA-1351"
|
||
assert a["work_kind"] == "разработка" and a["fact_status"] is None and a["neighbor"] is False
|
||
assert snap["issues"][2]["neighbor"] is True and snap["issues"][2]["work_kind"] is None
|
||
assert snap["links"] == [{"source": "AURORA-1620", "target": "AURORA-1621", "link_type": "ASUPR_Блокер_Начала"}]
|
||
assert snap["last_sync"] is None and snap["generated_at"]
|
||
|
||
|
||
def test_board_last_sync_is_last_successful_full_refresh(conn):
|
||
_sync_run(conn, "2026-10-14T09:59:00+00:00", "2026-10-14T10:00:00+00:00", "jql", BASE, count=198)
|
||
_sync_run(conn, "2026-10-14T10:59:00+00:00", "2026-10-14T11:00:00+00:00", "keys", "AURORA-1620", count=1)
|
||
_sync_run(conn, "2026-10-14T11:30:00+00:00", "2026-10-14T11:31:00+00:00", "jql", "key = AURORA-1", count=1)
|
||
_sync_run(conn, "2026-10-14T12:00:00+00:00", "2026-10-14T12:01:00+00:00", "jql", BASE, status="failed", count=0)
|
||
_sync_run(conn, "2026-10-14T13:00:00+00:00", None, "jql", BASE, status="claimed", count=50)
|
||
assert mirror.board(conn, BASE)["last_sync"] == {"started_at": "2026-10-14T09:59:00+00:00",
|
||
"finished_at": "2026-10-14T10:00:00+00:00", "issues_count": 198}
|
||
|
||
|
||
def test_board_inside_open_transaction_keeps_it_open(conn):
|
||
"""M-7 #20: вызов внутри уже открытой транзакции не открывает и не закрывает свою."""
|
||
mirror.apply_state(conn, [issue("AURORA-1620")], [])
|
||
with db.transaction(conn):
|
||
assert conn.in_transaction
|
||
snap = mirror.board(conn, BASE)
|
||
assert conn.in_transaction
|
||
assert not conn.in_transaction
|
||
assert [i["key"] for i in snap["issues"]] == ["AURORA-1620"]
|
||
|
||
|
||
def comment(cid, created="2026-10-14T12:00:00.000+0300", **over):
|
||
base = {"id": cid, "author": "Петров Пётр", "created": created, "updated": created, "body": f"текст {cid}"}
|
||
base.update(over)
|
||
return base
|
||
|
||
|
||
def test_comments_replaced_and_kept_without_field(conn):
|
||
mirror.apply_state(conn, [issue("A-1", comments=[comment("101"), comment("102")])], [])
|
||
mirror.apply_state(conn, [issue("A-1", comments=[comment("102", body="новый текст")])], [])
|
||
got = mirror.get_issue(conn, "A-1")["comments"]
|
||
assert [(c["comment_id"], c["body"]) for c in got] == [("102", "новый текст")]
|
||
mirror.apply_state(conn, [issue("A-1", summary="Без поля comments")], []) # агент до v10
|
||
assert [c["comment_id"] for c in mirror.get_issue(conn, "A-1")["comments"]] == ["102"]
|
||
mirror.apply_state(conn, [issue("A-1", comments=[])], [])
|
||
assert mirror.get_issue(conn, "A-1")["comments"] == []
|
||
|
||
|
||
def test_comments_newest_first_and_shape(conn):
|
||
mirror.apply_state(conn, [issue("A-1", comments=[
|
||
comment("9", created="2026-10-14T10:00:00.000+0300"),
|
||
comment("10", created="2026-10-15T10:00:00.000+0300", author=None, body=None),
|
||
comment("11", created="2026-10-15T10:00:00.000+0300")])], [])
|
||
got = mirror.get_issue(conn, "A-1")["comments"]
|
||
assert [c["comment_id"] for c in got] == ["11", "10", "9"] # то же время — по номеру, как число
|
||
assert got[1] == {"comment_id": "10", "author": None, "created": "2026-10-15T10:00:00.000+0300",
|
||
"updated": "2026-10-15T10:00:00.000+0300", "body": "", "attachment_ids": []}
|
||
|
||
|
||
def test_issue_epic_summary_and_absent_link(conn):
|
||
mirror.apply_state(conn, [issue("A-1", epic_key="E-1"), issue("E-1", summary="Эпик свода", epic_key=None)],
|
||
[{"source_key": "A-1", "target_key": "Z-9", "link_type": "ASUPR_Связанные"}])
|
||
got = mirror.get_issue(conn, "A-1")
|
||
assert got["epic_summary"] == "Эпик свода"
|
||
assert got["links_out"] == [{"target_key": "Z-9", "link_type": "ASUPR_Связанные", "summary": None, "status": None}]
|
||
assert mirror.get_issue(conn, "E-1")["epic_summary"] is None
|
||
|
||
|
||
@pytest.mark.parametrize("bad, field", [
|
||
("x", "comments"), (["x"], "comments[0]"), ([{"id": "", "body": "т"}], "comments[0].id"),
|
||
([{"id": 5, "body": "т"}], "comments[0].id"), ([{"id": "5", "body": 42}], "comments[0].body"),
|
||
([{"id": "5", "body": "т", "author": 7}], "comments[0].author"),
|
||
([{"id": "5", "body": "т", "created": 1}], "comments[0].created"),
|
||
([{"id": "5", "body": "т", "updated": []}], "comments[0].updated"),
|
||
])
|
||
def test_check_comments_rejects(bad, field):
|
||
with pytest.raises(ValueError) as e:
|
||
mirror.check_comments([issue("A-1"), issue("A-2", comments=bad)])
|
||
assert str(e.value).startswith(f"A-2: {field}")
|
||
|
||
|
||
def test_check_comments_accepts_missing_and_valid():
|
||
mirror.check_comments([issue("A-1"), issue("A-2", comments=[comment("1"), comment("2", author=None, body="")])])
|
||
|
||
|
||
def test_list_issues_hide_closed(conn):
|
||
mirror.apply_state(conn, [issue("A-1", status="В работе"), issue("A-2", status="Готово"),
|
||
issue("A-3", status="Снято"), issue("A-4", status=None)], [])
|
||
closed = ["Готово", "Снято"]
|
||
|
||
def keys(f, c=closed):
|
||
return sorted(i["key"] for i in mirror.list_issues(conn, f, closed_statuses=c))
|
||
assert keys({"hide_closed": "1"}) == ["A-1", "A-4"] # без статуса — не закрыта
|
||
assert keys({"hide_closed": "true"}) == ["A-1", "A-4"]
|
||
assert keys({}) == ["A-1", "A-2", "A-3", "A-4"]
|
||
assert keys({"hide_closed": "0"}) == ["A-1", "A-2", "A-3", "A-4"]
|
||
assert keys({"hide_closed": "1"}, []) == ["A-1", "A-2", "A-3", "A-4"] # пустой набор — ничего не скрыто
|
||
assert keys({"hide_closed": "1", "status": "Готово"}) == [] # отменяет скрытие приложение, не сервер
|
||
|
||
|
||
PROJ = "STOCKS_DRRK"
|
||
|
||
|
||
def _scene(conn): # сценарий data-model «Свои и соседи» + hub-api «Снимок проекта»
|
||
mirror.apply_state(conn, [
|
||
issue("AURORA-1", labels=["STOCKS_DRRK"], epic_key="AURORA-9"),
|
||
issue("AURORA-2", labels=["AWPCA"], epic_key=None),
|
||
issue("AURORA-3", labels=["AWPCA"], epic_key=None),
|
||
issue("AURORA-4", labels=["STOCKS_DRRK", "kind-dev"], epic_key="AURORA-404"), # эпика нет в зеркале
|
||
issue("AURORA-5", labels=["AWPCA"], epic_key=None),
|
||
issue("AURORA-9", labels=["AWPCA"], issue_type="Фича", epic_key=None)],
|
||
[{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "L"},
|
||
{"source_key": "AURORA-2", "target_key": "AURORA-3", "link_type": "L"},
|
||
{"source_key": "AURORA-5", "target_key": "AURORA-4", "link_type": "L"}, # входящая — тоже сосед
|
||
{"source_key": "AURORA-4", "target_key": "AURORA-777", "link_type": "L"}], # конца нет в зеркале
|
||
neighbor_keys=["AURORA-1"]) # признак зеркала не важен в разрезе
|
||
|
||
|
||
def test_board_project_scope(conn):
|
||
_scene(conn)
|
||
snap = mirror.board(conn, BASE, project=PROJ)
|
||
assert [(i["key"], i["neighbor"]) for i in snap["issues"]] == [
|
||
("AURORA-1", False), ("AURORA-2", True), ("AURORA-4", False), ("AURORA-5", True), ("AURORA-9", True)]
|
||
assert {(l["source"], l["target"]) for l in snap["links"]} == {
|
||
("AURORA-1", "AURORA-2"), ("AURORA-5", "AURORA-4"), ("AURORA-4", "AURORA-777")}
|
||
assert len(mirror.board(conn, BASE)["issues"]) == 6 # без проекта — всё зеркало
|
||
|
||
|
||
def test_board_project_last_sync_same(conn):
|
||
_scene(conn)
|
||
assert mirror.board(conn, BASE, project=PROJ)["last_sync"] == mirror.board(conn, BASE)["last_sync"]
|
||
|
||
|
||
def test_list_issues_project(conn): # сценарий «Список проекта»
|
||
_scene(conn)
|
||
keys = lambda **f: [i["key"] for i in mirror.list_issues(conn, {"project": PROJ, **f})]
|
||
assert sorted(keys()) == ["AURORA-1", "AURORA-2", "AURORA-4", "AURORA-5", "AURORA-9"]
|
||
assert sorted(keys(neighbor="1")) == ["AURORA-2", "AURORA-5", "AURORA-9"]
|
||
assert sorted(keys(neighbor="0")) == ["AURORA-1", "AURORA-4"]
|
||
assert keys(label="kind-dev") == ["AURORA-4"]
|
||
assert {i["key"]: i["neighbor"] for i in mirror.list_issues(conn, {"project": PROJ})}["AURORA-1"] is False
|
||
|
||
|
||
def test_list_issues_project_paging_and_filters(conn):
|
||
_scene(conn)
|
||
page = mirror.list_issues(conn, {"project": PROJ}, limit=2, offset=1)
|
||
assert [i["key"] for i in page] == ["AURORA-2", "AURORA-4"]
|
||
assert [i["key"] for i in mirror.list_issues(conn, {"project": PROJ, "q": "AURORA-5"})] == ["AURORA-5"]
|
||
|
||
|
||
def test_labels_of(conn):
|
||
_scene(conn)
|
||
assert mirror.labels_of(conn, "AURORA-4") == ["STOCKS_DRRK", "kind-dev"]
|
||
assert mirror.labels_of(conn, "AURORA-404") is None
|
||
|
||
|
||
def test_apply_state_comment_created_at(conn):
|
||
mirror.apply_state(conn, [{"key": "X", "comments": [
|
||
{"id": "1", "body": "a", "created": "2026-10-22T13:05:11.000+0300"},
|
||
{"id": "2", "body": "b", "created": "мусор"}]}], [])
|
||
rows = {r["comment_id"]: r["created_at"] for r in conn.execute("SELECT * FROM issue_comment")}
|
||
assert rows == {"1": "2026-10-22T10:05:11+00:00", "2": None}
|
||
|
||
|
||
def attachment(aid, **over):
|
||
base = {"id": aid, "filename": f"файл{aid}.png", "size": 1000, "mime_type": "image/png", "author": "Петров Пётр",
|
||
"created": "2026-10-14T12:00:00.000+0300"}
|
||
base.update(over)
|
||
return base
|
||
|
||
|
||
def _atts(conn, key="A-1"):
|
||
return {r["attachment_id"]: dict(r) for r in conn.execute("SELECT * FROM issue_attachment WHERE key = ?", (key,))}
|
||
|
||
|
||
def test_attachments_replace(conn):
|
||
mirror.apply_state(conn, [issue("A-1", attachments=[attachment("1"), attachment("2")])], [])
|
||
mirror.apply_state(conn, [issue("A-1", attachments=[attachment("2", filename="новое.pdf", size=5), attachment("3")])], [])
|
||
got = _atts(conn)
|
||
assert sorted(got) == ["2", "3"]
|
||
assert got["2"]["filename"] == "новое.pdf" and got["2"]["size"] == 5
|
||
assert got["2"]["created_at"] == "2026-10-14T09:00:00+00:00"
|
||
|
||
|
||
def test_attachments_absent_kept(conn):
|
||
mirror.apply_state(conn, [issue("A-1", attachments=[attachment("1")])], [])
|
||
mirror.apply_state(conn, [issue("A-1", summary="Без поля")], [])
|
||
assert sorted(_atts(conn)) == ["1"]
|
||
|
||
|
||
def test_attachment_file_untouched(conn):
|
||
mirror.apply_state(conn, [issue("A-1", attachments=[attachment("1")])], [])
|
||
conn.execute("INSERT INTO attachment_file(attachment_id, file, size, fetched_at) VALUES ('1', 'abc', 1000, '2026-10-22T10:00:00+00:00')")
|
||
mirror.apply_state(conn, [issue("A-1", attachments=[])], [])
|
||
assert _atts(conn) == {}
|
||
assert conn.execute("SELECT file FROM attachment_file WHERE attachment_id = '1'").fetchone()["file"] == "abc"
|
||
|
||
|
||
def test_attachment_moves_to_other_key(conn):
|
||
mirror.apply_state(conn, [issue("A-1", attachments=[attachment("1")])], [])
|
||
mirror.apply_state(conn, [issue("A-2", attachments=[attachment("1")])], [])
|
||
assert _atts(conn, "A-1") == {} and sorted(_atts(conn, "A-2")) == ["1"]
|
||
|
||
|
||
@pytest.mark.parametrize("bad, field", [
|
||
("x", "attachments"), (["x"], "attachments[0]"),
|
||
([{"id": "", "filename": "a"}], "attachments[0].id"), ([{"id": "1a", "filename": "a"}], "attachments[0].id"),
|
||
([{"id": "-5", "filename": "a"}], "attachments[0].id"), ([{"id": 5, "filename": "a"}], "attachments[0].id"),
|
||
([{"id": "5", "filename": 1}], "attachments[0].filename"), ([{"id": "5", "filename": ""}], "attachments[0].filename"),
|
||
([{"id": "5", "filename": "a", "size": -1}], "attachments[0].size"),
|
||
([{"id": "5", "filename": "a", "size": "1"}], "attachments[0].size"),
|
||
([{"id": "5", "filename": "a", "size": True}], "attachments[0].size"),
|
||
([{"id": "5", "filename": "a", "mime_type": 5}], "attachments[0].mime_type"),
|
||
])
|
||
def test_check_attachments_errors(bad, field):
|
||
with pytest.raises(ValueError) as e:
|
||
mirror.check_attachments([issue("A-1"), issue("A-2", attachments=bad)])
|
||
assert str(e.value).startswith(f"A-2: {field}")
|
||
|
||
|
||
def test_check_attachments_accepts_missing_and_valid():
|
||
mirror.check_attachments([issue("A-1"), issue("A-2", attachments=[attachment("1"), attachment("2", size=None, mime_type=None)])])
|
||
|
||
|
||
def test_get_issue_attachments(conn):
|
||
mirror.apply_state(conn, [issue(
|
||
"A-1", description="см. !старый.png!",
|
||
attachments=[attachment("31337", filename="скрин 1.png", created="2026-10-14T12:00:00.000+0300"),
|
||
attachment("31338", filename="ОСВ 60.xlsx", created="2026-10-15T12:00:00.000+0300")],
|
||
comments=[comment("101", body="см. !скрин 1.png|thumbnail! и [^ОСВ 60.xlsx], ещё раз !скрин 1.png!"),
|
||
comment("102", body="без ссылок")])], [])
|
||
got = mirror.get_issue(conn, "A-1")
|
||
assert [a["attachment_id"] for a in got["attachments"]] == ["31338", "31337"]
|
||
assert got["attachments"][0]["state"] == "none" and got["attachments"][0]["error"] is None
|
||
assert got["description_attachment_ids"] == []
|
||
by_id = {c["comment_id"]: c for c in got["comments"]}
|
||
assert by_id["101"]["attachment_ids"] == ["31337", "31338"] and by_id["102"]["attachment_ids"] == []
|
||
|
||
|
||
def test_get_issue_commands_skip_fetch(conn):
|
||
mirror.apply_state(conn, [issue("A-1")], [])
|
||
conn.execute("INSERT INTO batch(id, title, author, created_at) VALUES (1, 't', 'a', ?)", (db.now(),))
|
||
for seq, typ in enumerate(["fetch_attachment", "ensure_fields"]):
|
||
conn.execute("INSERT INTO command(batch_id, seq, type, params) VALUES (1, ?, ?, ?)",
|
||
(seq, typ, '{"key": "A-1", "attachment_id": "1"}'))
|
||
assert [c["type"] for c in mirror.get_issue(conn, "A-1")["commands"]] == ["ensure_fields"]
|