112 lines
5.9 KiB
Python
112 lines
5.9 KiB
Python
"""Журнал изменений Jira в refresh (агент v13, spec agent «Журнал изменений»)."""
|
||
import unittest
|
||
from datetime import datetime, timezone
|
||
|
||
from _common import SETTINGS, agent, jira_cfg, no_sleep
|
||
from fake_hub import FakeHub
|
||
from fake_jira import FakeJira
|
||
|
||
|
||
def _st(field, frm, to, fieldtype="jira", **kw):
|
||
return dict({"field": field, "fieldtype": fieldtype, "from": frm, "fromString": frm, "to": to, "toString": to}, **kw)
|
||
|
||
|
||
class Base(unittest.TestCase):
|
||
def setUp(self):
|
||
self.fake = FakeJira()
|
||
self.hub = FakeHub(SETTINGS)
|
||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
self.jira = agent.JiraClient(cfg)
|
||
self.jira.sleep = no_sleep
|
||
self.ex = agent.Executor(self.jira, dict(SETTINGS), agent.HubClient(cfg))
|
||
self.ex.sleep = no_sleep
|
||
self.logs = []
|
||
self._log = agent.log
|
||
agent.log = self.logs.append
|
||
|
||
def tearDown(self):
|
||
agent.log = self._log
|
||
self.fake.stop()
|
||
self.hub.stop()
|
||
|
||
|
||
class ChangelogItemsTest(Base):
|
||
def test_jira_time_forms(self):
|
||
want = datetime(2026, 10, 22, 10, 5, 11, tzinfo=timezone.utc)
|
||
for raw in ("2026-10-22T13:05:11.000+0300", "2026-10-22T13:05:11+03:00", "2026-10-22T10:05:11Z",
|
||
"2026-10-22T10:05:11.000Z", "2026-10-22T05:05:11.000-0500"):
|
||
self.assertEqual(agent._jira_time(raw), want, raw)
|
||
for raw in ("мусор", "", None, 42):
|
||
self.assertIsNone(agent._jira_time(raw))
|
||
|
||
def test_changelog_items_since_and_shape(self):
|
||
histories = [
|
||
{"id": "1", "author": {"name": "petrov", "displayName": "Петров Пётр"}, "created": "2026-10-21T10:00:00.000+0300",
|
||
"items": [_st("status", "Открыта", "В работе")]},
|
||
{"id": 2, "author": {"name": "ivanov"}, "created": "2026-10-22T12:10:00.000+0300",
|
||
"items": [dict(_st("status", "В работе", "Готово"), fieldId="status", **{"from": "3", "to": "10001"}),
|
||
_st("resolution", None, "Готово")]}]
|
||
items = self.ex.changelog_items("X", histories, agent._jira_time("2026-10-22T08:45:00Z"), lambda n: None)
|
||
self.assertEqual([(i["history_id"], i["seq"]) for i in items], [("2", 0), ("2", 1)])
|
||
self.assertEqual(items[0], {"history_id": "2", "seq": 0, "key": "X", "created": "2026-10-22T12:10:00.000+0300",
|
||
"author": "ivanov", "field_id": "status", "field_name": "status", "from": "3",
|
||
"from_string": "В работе", "to": "10001", "to_string": "Готово"})
|
||
self.assertEqual(items[1]["field_id"], "resolution")
|
||
self.assertIsNone(items[1]["from"])
|
||
|
||
def test_changelog_items_since_boundary_included(self):
|
||
h = [{"id": "1", "author": None, "created": "2026-10-22T11:45:00.000+0300", "items": [_st("status", "a", "b")]}]
|
||
items = self.ex.changelog_items("X", h, agent._jira_time("2026-10-22T08:45:00Z"), lambda n: None)
|
||
self.assertEqual(len(items), 1)
|
||
self.assertIsNone(items[0]["author"])
|
||
|
||
def test_changelog_items_custom_field_lookup(self):
|
||
asked = []
|
||
|
||
def field_id_of(name):
|
||
asked.append(name)
|
||
return "customfield_10201"
|
||
h = [{"id": "1", "author": {"name": "p"}, "created": "2026-10-22T12:00:00.000+0300",
|
||
"items": [_st("Плановое окончание", "2026-10-25", "2026-10-30", fieldtype="custom"),
|
||
_st("Sprint", "1", "2", fieldtype="custom", fieldId="customfield_10104"),
|
||
_st("duedate", "2026-10-25", "2026-10-30")]}]
|
||
items = self.ex.changelog_items("X", h, None, field_id_of)
|
||
self.assertEqual([i["field_id"] for i in items], ["customfield_10201", "customfield_10104", "duedate"])
|
||
self.assertEqual(asked, ["Плановое окончание"])
|
||
|
||
def test_changelog_items_drops_bad(self):
|
||
h = [{"id": None, "author": None, "created": "2026-10-22T12:00:00.000+0300", "items": [_st("status", "a", "b")]},
|
||
{"id": "2", "author": None, "created": "", "items": [_st("status", "a", "b")]},
|
||
{"id": "3", "author": None, "created": "2026-10-22T12:00:00.000+0300",
|
||
"items": [_st("", "a", "b"), _st("status", 1, 2)]}]
|
||
items = self.ex.changelog_items("X", h, None, lambda n: None)
|
||
self.assertEqual([(i["history_id"], i["seq"], i["from"], i["to"]) for i in items], [("3", 1, "1", "2")])
|
||
self.assertEqual(len([l for l in self.logs if "X" in l]), 3)
|
||
|
||
def test_changelog_items_no_since(self):
|
||
h = [{"id": str(n), "author": None, "created": "2020-01-01T00:00:00.000+0300", "items": [_st("status", "a", "b")]}
|
||
for n in range(3)]
|
||
self.assertEqual(len(self.ex.changelog_items("X", h, None, lambda n: None)), 3)
|
||
|
||
|
||
class FakeChangelogTest(Base):
|
||
def test_fake_search_expands_changelog(self):
|
||
self.fake.add("AURORA-1")
|
||
for n in range(3):
|
||
self.fake.add_history("AURORA-1", "2026-10-22T12:00:00.000+0300", [_st("status", "a", "b")])
|
||
self.fake.changelog_cap = 2
|
||
page = self.jira.get(agent.API + "/search", jql="key in (AURORA-1)", fields="key", expand="changelog")
|
||
cl = page["issues"][0]["changelog"]
|
||
self.assertEqual((cl["total"], len(cl["histories"])), (3, 2))
|
||
self.fake.issue_changelog_cap = 3
|
||
full = self.jira.get(agent.API + "/issue/AURORA-1", fields="key", expand="changelog")
|
||
self.assertEqual(len(full["changelog"]["histories"]), 3)
|
||
self.assertNotIn("changelog", self.jira.get(agent.API + "/issue/AURORA-1"))
|
||
self.fake.jql_results["x AND sprint in openSprints()"] = ["AURORA-1"]
|
||
self.assertEqual(self.fake.search_keys("x AND sprint in openSprints()"), ["AURORA-1"])
|
||
self.assertTrue(any(f["name"] == "Плановое окончание" for f in self.jira.get(agent.API + "/field")))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|