import pytest from jira_hub import history ITEM = {"history_id": "5001", "seq": 0, "key": "X", "created": "2026-10-22T09:00:00.000+0300", "author": "Петров Пётр", "field_id": "status", "field_name": "status", "from": "3", "from_string": "В работе", "to": "10001", "to_string": "Готово"} @pytest.fixture def conn(open_db): return open_db() def test_record_dedup(conn): assert history.record_changes(conn, [ITEM]) == {"accepted": 1, "duplicates": 0} assert history.record_changes(conn, [ITEM, {**ITEM, "seq": 1, "field_id": None, "field_name": "resolution"}]) == \ {"accepted": 1, "duplicates": 1} row = conn.execute("SELECT * FROM issue_change WHERE seq = 0").fetchone() assert row["at"] == "2026-10-22T06:00:00+00:00" and row["created"] == ITEM["created"] assert row["field"] == "status" and row["from_value"] == "3" and row["to_string"] == "Готово" assert conn.execute("SELECT field FROM issue_change WHERE seq = 1").fetchone()["field"] == "resolution" def test_record_keeps_first_version(conn): history.record_changes(conn, [ITEM]) history.record_changes(conn, [{**ITEM, "to_string": "Снято"}]) assert conn.execute("SELECT to_string FROM issue_change").fetchone()["to_string"] == "Готово" def test_record_uses_field_settings(conn): from jira_hub import settings settings.set_(conn, "field_ids", {"planned_end": "customfield_10201"}) history.record_changes(conn, [{**ITEM, "field_id": "customfield_10201", "field_name": "Плановое окончание"}]) assert conn.execute("SELECT field FROM issue_change").fetchone()["field"] == "planned_end" @pytest.mark.parametrize("item, field_ids, epic, want", [ ({"field_id": "customfield_10201", "field_name": "Плановое окончание"}, {"planned_end": "customfield_10201"}, {}, "planned_end"), ({"field_id": "customfield_10200", "field_name": "Плановое начало"}, {"planned_start": "customfield_10200"}, {}, "planned_start"), ({"field_id": None, "field_name": "duedate"}, {}, {}, "duedate"), ({"field_id": "status", "field_name": "Статус"}, {}, {}, "status"), ({"field_id": None, "field_name": "Fix Version"}, {}, {}, "Fix Version"), ({"field_id": "customfield_1", "field_name": "Epic Link"}, {}, {"field_id": "customfield_1"}, "epic_link"), ({"field_id": "customfield_9", "field_name": "Sprint"}, None, None, "Sprint"), ]) def test_logical_field(item, field_ids, epic, want): assert history.logical_field(item, field_ids, epic) == want @pytest.mark.parametrize("bad, field", [({"seq": "0"}, "seq"), ({"seq": -1}, "seq"), ({"seq": True}, "seq"), ({"key": ""}, "key"), ({"history_id": 5001}, "history_id"), ({"created": None}, "created"), ({"field_name": ""}, "field_name"), ({"to_string": 5}, "to_string"), ({"author": []}, "author")]) def test_check_changes_rejects(bad, field): with pytest.raises(ValueError, match=rf"changes\[1\]\.{field}"): history.check_changes([ITEM, {**ITEM, **bad}]) def test_check_changes_shape(): with pytest.raises(ValueError, match="changes — список"): history.check_changes({"a": 1}) with pytest.raises(ValueError, match=r"changes\[0\]"): history.check_changes(["x"]) history.check_changes([ITEM, {**ITEM, "author": None, "field_id": None, "from": None, "to_string": None}]) # --- лента и история задачи (Task 4) ---------------------------------------------------------------------------- def _issue(conn, key, labels, summary="Тема", status="В работе"): import json conn.execute("INSERT INTO issue(key, summary, status, labels, fetched_at) VALUES (?, ?, ?, ?, '2026-10-22T00:00:00+00:00')", (key, summary, status, json.dumps(labels))) def _change(conn, key, hid, created, field_name, frm=None, to=None, seq=0, field_id=None, from_value=None, to_value=None): history.record_changes(conn, [{"history_id": hid, "seq": seq, "key": key, "created": created, "author": "Петров", "field_id": field_id, "field_name": field_name, "from": from_value, "from_string": frm, "to": to_value, "to_string": to}]) def test_feed_closed_in_two_hours(conn): # сценарий «Что закрылось за два часа» _issue(conn, "X", ["AWPCA"], status="Готово") _issue(conn, "Y", ["STOCKS_DRRK"]) _change(conn, "X", "1", "2026-10-22T14:15:00.000+0300", "status", "В работе", "Готово") _change(conn, "Y", "2", "2026-10-22T14:20:00.000+0300", "status", "Открыта", "В работе") got = history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", ("status",), project="AWPCA") assert got["truncated"] is False assert got["changes"] == [{"at": "2026-10-22T11:15:00+00:00", "key": "X", "summary": "Тема", "status": "Готово", "kind": "status", "field": "status", "author": "Петров", "from": "В работе", "to": "Готово", "from_value": None, "to_value": None}] def test_feed_comment(conn): # «Комментарий в ленте» from jira_hub import mirror mirror.apply_state(conn, [{"key": "X", "labels": ["AWPCA"], "comments": [ {"id": "7", "author": "Иванов", "body": "Готово к проверке", "created": "2026-10-22T13:30:00.000+0300"}]}], []) got = history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", history.DEFAULT_KINDS) assert got["changes"] == [{"at": "2026-10-22T10:30:00+00:00", "key": "X", "summary": None, "status": None, "kind": "comment", "field": "comment", "author": "Иванов", "from": None, "to": "Готово к проверке", "from_value": None, "to_value": None}] def test_feed_other_only_on_request(conn): # «Прочие поля по запросу» _change(conn, "X", "1", "2026-10-22T11:00:00Z", "assignee", "a", "b", field_id="assignee") window = ("2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00") assert history.feed(conn, *window, history.DEFAULT_KINDS)["changes"] == [] got = history.feed(conn, *window, history.parse_kinds("all"))["changes"] assert [(e["kind"], e["field"]) for e in got] == [("other", "assignee")] def test_feed_neighbor_not_in_project(conn): # «Сосед проекта не в ленте» _issue(conn, "A", ["AWPCA"]) _issue(conn, "Z", ["OTHER"]) conn.execute("INSERT INTO issue_link(source_key, target_key, link_type) VALUES ('Z', 'A', 'блок')") _change(conn, "Z", "1", "2026-10-22T11:00:00Z", "status", "a", "b") window = ("2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00") assert history.feed(conn, *window, ("status",), project="AWPCA")["changes"] == [] assert len(history.feed(conn, *window, ("status",), key="Z")["changes"]) == 1 def test_feed_issue_not_in_mirror(conn): # Review Focus 4 _change(conn, "Q", "1", "2026-10-22T11:00:00Z", "status", "a", "b") got = history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", ("status",), key="Q")["changes"] assert got[0]["key"] == "Q" and got[0]["summary"] is None and got[0]["status"] is None assert history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", ("status",))["changes"] def test_feed_truncated_and_order(conn): for n, t in enumerate(("10:10", "10:30", "10:20")): _change(conn, "X", str(n), f"2026-10-22T{t}:00Z", "status", "a", "b") got = history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", ("status",), limit=2) assert got["truncated"] is True assert [e["at"][11:16] for e in got["changes"]] == ["10:30", "10:20"] def test_feed_window_bounds(conn): _change(conn, "X", "1", "2026-10-22T10:00:00Z", "status", "a", "b") _change(conn, "X", "2", "2026-10-22T12:00:01Z", "status", "b", "c") got = history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", ("status",))["changes"] assert [e["to"] for e in got] == ["b"] def test_feed_dates_kind_value(conn): _change(conn, "X", "1", "2026-10-22T11:00:00Z", "duedate", "25/окт/26", "30/окт/26", field_id="duedate", from_value="2026-10-25", to_value="2026-10-30") _change(conn, "X", "2", "2026-10-22T11:05:00Z", "description", None, None, field_id="description", from_value="старое", to_value="новое") got = history.feed(conn, "2026-10-22T10:00:00+00:00", "2026-10-22T12:00:00+00:00", ("dates", "description"))["changes"] assert [(e["kind"], e["from"], e["from_value"]) for e in got] == [ ("description", "старое", "старое"), ("dates", "25/окт/26", "2026-10-25")] @pytest.mark.parametrize("value, want", [(None, history.DEFAULT_KINDS), ("", history.DEFAULT_KINDS), ("all", history.KINDS), ("status, comment", ("status", "comment"))]) def test_parse_kinds(value, want): assert history.parse_kinds(value) == want def test_parse_kinds_unknown(): with pytest.raises(ValueError, match="kinds"): history.parse_kinds("status,foo") def test_issue_changes(conn): # «Задача с историей» from jira_hub import mirror _issue(conn, "X", ["AWPCA"]) _change(conn, "X", "1", "2026-10-21T11:00:00Z", "status", "a", "b") _change(conn, "X", "2", "2026-10-22T11:00:00Z", "description", "c", "d", field_id="description") _change(conn, "X", "3", "мусор", "assignee", "e", "f", field_id="assignee") got = mirror.get_issue(conn, "X")["changes"] assert [(e["kind"], e["at"]) for e in got] == [("description", "2026-10-22T11:00:00+00:00"), ("status", "2026-10-21T11:00:00+00:00"), ("other", None)] assert "summary" not in got[0] and "status" not in got[0] assert mirror.get_issue(conn, "X")["changes"] and history.issue_changes(conn, "NOPE") == []