176 lines
10 KiB
Python
176 lines
10 KiB
Python
import pytest
|
||
|
||
from jira_hub.modules.jira 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.core 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.modules.jira 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.modules.jira 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") == []
|