66 lines
3.4 KiB
Python
66 lines
3.4 KiB
Python
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}])
|