import pytest from jira_hub import db, 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": ""} 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="")])])