import csv import pytest from jira_hub import importer, mirror, settings @pytest.fixture def conn(open_db): """База в памяти с настройками по умолчанию; закрывается после теста.""" c = open_db() settings.ensure_defaults(c) return c def write_csv(path, header, rows): with open(path, "w", encoding="utf-8-sig", newline="") as f: w = csv.writer(f) w.writerow(header) w.writerows(rows) def make_package(folder): write_csv(folder / "tasks.csv", ["Код", "Тема", "Вид работы", "Статус по факту", "Комментарий актуализации", "Что сделано по факту (внутр.)", "Что осталось (внутр.)", "Спринт"], [["AURORA-1533", "Тест ОСВ", "тестирование", "Ожидание разработки", "Актуализация 10.10.2026. Порядок", "", "Прогнать на стенде", "Спринт 2"], ["AURORA-1416", "Карточка", "UI", "В работе", "", "Сделан список", "Модалка", ""]]) write_csv(folder / "new_tasks.csv", ["Ключ", "Тема", "Вид работы", "Статус по факту", "Описание", "Основание (внутр.)", "Спринт"], [["AURORA-1586", "Справочники: тестирование", "тестирование", "Ожидание разработки", "h3. Контекст", "решение 11.10", "Спринт 2"], ["", "Ещё не создана", "разработка", "Ожидание разработки", "h3. Контекст", "идея", ""]]) write_csv(folder / "epics.csv", ["Код", "Ключ", "Эпик", "Комментарий"], [["AURORA-1350", "AURORA-1350", "Загрузка", "ядро"], ["NEW-1", "AURORA-1529", "Роли", "новый"]]) write_csv(folder / "links.csv", ["Откуда", "Тип", "Куда", "Пояснение (внутр.)"], [["AURORA-1533", "блокирует начало", "AURORA-1416", "x"], ["new:2", "связанные", "AURORA-1533", "y"], ["new:3", "связанные", "AURORA-1533", "z"]]) def test_import_notes_and_report(conn, tmp_path): make_package(tmp_path) settings.set_(conn, "link_types", {"блокирует начало": "ASUPR_Блокер_Начала", "связанные": "ASUPR_Связанные"}) mirror.apply_state(conn, [{"key": "AURORA-1533", "summary": "т", "labels": []}, {"key": "AURORA-1586", "summary": "т", "labels": []}], [{"source_key": "AURORA-1586", "target_key": "AURORA-1533", "link_type": "ASUPR_Связанные"}]) rep = importer.import_package(conn, tmp_path) assert rep["notes"] == 6 n = mirror.get_note(conn, "AURORA-1533") assert n["fact_status"] == "Ожидание разработки" and n["work_kind"] == "тестирование" assert n["rest_text"] == "Прогнать на стенде" and n["owner_note"] == "Актуализация 10.10.2026. Порядок" and n["updated_by"] == "import" assert mirror.get_note(conn, "AURORA-1416")["done_text"] == "Сделан список" assert mirror.get_note(conn, "AURORA-1586")["rest_text"] == "решение 11.10" assert mirror.get_note(conn, "new:3")["owner_note"] == "Ещё не создана" assert mirror.get_note(conn, "AURORA-1529")["work_kind"] == "эпик" assert rep["missing_keys"] == ["AURORA-1350", "AURORA-1416", "AURORA-1529"] assert rep["expected_links"] == 2 # new:3 без ключа — не сверяется assert rep["missing_links"] == [{"source": "AURORA-1533", "target": "AURORA-1416", "link_type": "ASUPR_Блокер_Начала"}] assert rep["reversed_links"] == [] assert importer.import_package(conn, tmp_path)["notes"] == 6 assert conn.execute("SELECT COUNT(*) AS n FROM issue_note").fetchone()["n"] == 6 def write_tasks(folder, comment): write_csv(folder / "tasks.csv", ["Код", "Тема", "Вид работы", "Статус по факту", "Комментарий актуализации", "Что сделано по факту (внутр.)", "Что осталось (внутр.)", "Спринт"], [["AURORA-1533", "Тест ОСВ", "тестирование", "Ожидание разработки", comment, "", "Прогнать", "Спринт 2"]]) def test_reimport_keeps_hub_edit_when_cell_empty(conn, tmp_path): write_tasks(tmp_path, "Актуализация 10.10.2026") importer.import_package(conn, tmp_path) mirror.set_note(conn, "AURORA-1533", {"owner_note": "правка в хабе"}, "hub") write_tasks(tmp_path, "") # пустая ячейка в CSV importer.import_package(conn, tmp_path) n = mirror.get_note(conn, "AURORA-1533") assert n["owner_note"] == "правка в хабе" assert n["rest_text"] == "Прогнать" def test_reimport_updates_nonempty_cell(conn, tmp_path): write_tasks(tmp_path, "Первая версия") importer.import_package(conn, tmp_path) write_tasks(tmp_path, "Вторая версия") importer.import_package(conn, tmp_path) assert mirror.get_note(conn, "AURORA-1533")["owner_note"] == "Вторая версия" def test_import_does_not_touch_issue_tables(conn, tmp_path): make_package(tmp_path) mirror.apply_state(conn, [{"key": "AURORA-1533", "summary": "т", "labels": []}], [{"source_key": "AURORA-1533", "target_key": "AURORA-1416", "link_type": "x"}]) count = lambda t: conn.execute(f"SELECT COUNT(*) AS n FROM {t}").fetchone()["n"] before = (count("issue"), count("issue_link")) importer.import_package(conn, tmp_path) assert (count("issue"), count("issue_link")) == before def test_missing_csv_files_give_empty_report(conn, tmp_path): rep = importer.import_package(conn, tmp_path) assert rep == {"notes": 0, "expected_links": 0, "missing_keys": [], "missing_links": [], "reversed_links": []} def test_reversed_link_reported_separately(conn, tmp_path): make_package(tmp_path) write_csv(tmp_path / "links.csv", ["Откуда", "Тип", "Куда", "Пояснение (внутр.)"], [["AURORA-1533", "связанные", "AURORA-1586", "в Jira стоит наоборот"], ["AURORA-1533", "блокирует начало", "AURORA-1416", "нет совсем"]]) settings.set_(conn, "link_types", {"блокирует начало": "ASUPR_Блокер_Начала", "связанные": "ASUPR_Связанные"}) mirror.apply_state(conn, [{"key": "AURORA-1533", "summary": "т", "labels": []}, {"key": "AURORA-1586", "summary": "т", "labels": []}], [{"source_key": "AURORA-1586", "target_key": "AURORA-1533", "link_type": "ASUPR_Связанные"}]) rep = importer.import_package(conn, tmp_path) assert rep["reversed_links"] == [{"source": "AURORA-1533", "target": "AURORA-1586", "link_type": "ASUPR_Связанные"}] assert rep["missing_links"] == [{"source": "AURORA-1533", "target": "AURORA-1416", "link_type": "ASUPR_Блокер_Начала"}] def test_epic_without_key_is_skipped(tmp_path, conn): """M-8 #16 (T7.3 #4), спека migration «Эпик без ключа»: строка без «Ключ» пропущена, заметки new:* нет.""" write_csv(tmp_path / "epics.csv", ["Код", "Ключ", "Эпик", "Комментарий"], [["NEW-2", "", "Без ключа", "ещё не создан"], ["AURORA-1350", "AURORA-1350", "Загрузка", "ядро"]]) rep = importer.import_package(conn, tmp_path) assert rep["notes"] == 1 and rep["missing_keys"] == ["AURORA-1350"] assert [r["key"] for r in conn.execute("SELECT key FROM issue_note ORDER BY key")] == ["AURORA-1350"]