"""Журнал изменений 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"))) SINCE = "2026-10-22T08:45:00Z" CHANGED = "2026-10-22T12:30:00.000+0300" # 09:30 UTC — позже SINCE BASE = "project = AURORA AND labels = AWPCA" class RefreshHistoryTest(Base): def refresh(self, history=None, **params): p = dict({"jql": BASE}, **params) if history is not None: p["history"] = history return self.ex.run({"id": 5, "type": "refresh", "params": p}) def changelog_queries(self): return [q for q in self.fake.queries if q.get("expand") == "changelog"] def sent(self): return [c for b in self.hub.histories for c in b["changes"]] def test_nothing_changed_no_changelog_calls(self): # «Ничего не менялось» for n in range(3): self.fake.add("AURORA-%d" % n) out = self.refresh({"since": SINCE}) self.assertEqual(out["status"], "done") self.assertEqual(out["result"]["history"], {"issues": 0, "changes": 0, "incomplete": []}) self.assertEqual(self.changelog_queries(), []) self.assertEqual(self.hub.histories, []) def test_two_changed_one_search(self): # «Две задачи изменились» self.fake.add("X", updated=CHANGED) self.fake.add("Y", updated=CHANGED) self.fake.add("Z") self.fake.add_history("X", "2026-10-21T12:00:00.000+0300", [_st("status", "a", "b")]) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "b", "c"), _st("resolution", None, "Готово")]) self.fake.add_history("Y", "2026-10-22T12:30:00.000+0300", [_st("description", "старое", "новое")]) out = self.refresh({"since": SINCE}) qs = self.changelog_queries() self.assertEqual(len(qs), 1) self.assertEqual(qs[0]["jql"], "key in (X,Y)") self.assertEqual(qs[0]["fields"], "key") self.assertEqual([(c["key"], c["seq"]) for c in self.sent()], [("X", 0), ("X", 1), ("Y", 0)]) self.assertEqual(self.hub.histories[0]["command_id"], 5) self.assertEqual(out["result"]["history"], {"issues": 2, "changes": 3, "incomplete": []}) self.assertEqual(self.hub.posts, ["/agent/state", "/agent/history"]) # журнал — после состояния def test_backfill_by_jql(self): # «Стартовая загрузка» jql = "(%s) AND sprint in openSprints() AND statusCategory != Done" % BASE keys = ["AURORA-%d" % n for n in range(12)] for k in keys: self.fake.add(k) self.fake.add_history(k, "2020-01-01T00:00:00.000+0300", [_st("status", "a", "b")]) self.fake.jql_results[jql] = keys out = self.refresh({"jql": jql}) self.assertEqual([q["jql"] for q in self.changelog_queries()], [jql]) self.assertEqual(len(self.sent()), 12) self.assertEqual(out["result"]["history"], {"issues": 12, "changes": 12, "incomplete": []}) def test_since_and_jql_dedup(self): # Review Focus 4 jql = "(%s) AND sprint in openSprints() AND statusCategory != Done" % BASE self.fake.add("X", updated=CHANGED) self.fake.add("Y") self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) self.fake.add_history("Y", "2020-01-01T00:00:00.000+0300", [_st("status", "a", "b")]) self.fake.jql_results[jql] = ["X", "Y"] out = self.refresh({"since": SINCE, "jql": jql}) self.assertEqual(sorted(c["key"] for c in self.sent()), ["X", "Y"]) self.assertEqual(out["result"]["history"]["changes"], 2) def test_since_and_jql_keeps_old_history(self): # I-1: задача в обеих выборках — журнал целиком jql = "(%s) AND sprint in openSprints() AND statusCategory != Done" % BASE self.fake.add("X", updated=CHANGED) self.fake.add_history("X", "2020-01-01T00:00:00.000+0300", [_st("status", "a", "b")]) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "b", "c")]) self.fake.jql_results[jql] = ["X"] out = self.refresh({"since": SINCE, "jql": jql}) self.assertEqual(sorted(c["created"][:4] for c in self.sent()), ["2020", "2026"]) self.assertEqual(out["result"]["history"]["changes"], 2) self.assertEqual(len(self.changelog_queries()), 1) # X уже прочитана стартовой — второго search нет def test_backfill_failure_on_issue_leaves_it_to_since(self): # I-1: упавшая на задаче стартовая часть её не «съедает» jql = "(%s) AND sprint in openSprints() AND statusCategory != Done" % BASE self.fake.add("X", updated=CHANGED) self.fake.add_history("X", "2020-01-01T00:00:00.000+0300", [_st("status", "a", "b")]) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "b", "c")]) self.fake.jql_results[jql] = ["X"] self.fake.changelog_cap = 1 # search обрезает — дочитывание GET /issue отдаёт обе записи self.fake.issue_changelog_cap = 2 orig, failed = self.fake.route, [] def route(method, path, q, body): if path == "/rest/api/2/issue/X" and q.get("expand") == "changelog" and not failed: failed.append(1) return 500, {"errorMessages": ["сбой"]} return orig(method, path, q, body) self.fake.route = route out = self.refresh({"since": SINCE, "jql": jql}) self.assertTrue(out["result"]["history"]["backfill_error"].startswith("HTTP 500")) self.assertEqual([c["created"][:4] for c in self.sent()], ["2026"]) def test_neighbor_changed_included(self): # Review Focus 5 self.fake.add("X") self.fake.add("N", in_base=False, updated=CHANGED) self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "X", "target": "N"}) self.fake.add_history("N", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) self.refresh({"since": SINCE}) self.assertEqual([c["key"] for c in self.sent()], ["N"]) def test_jira_500_degrades(self): # «Ошибка Jira при чтении журнала» self.fake.add("X", updated=CHANGED) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) orig = self.fake.route def route(method, path, q, body): if q.get("expand") == "changelog": return 500, {"errorMessages": ["сбой"]} return orig(method, path, q, body) self.fake.route = route out = self.refresh({"since": SINCE}) self.assertEqual(out["status"], "done") self.assertTrue(out["result"]["history"]["error"].startswith("HTTP 500")) self.assertNotIn("retryable", out) self.assertTrue(self.hub.states) self.assertTrue(any("журнал" in l for l in self.logs)) def test_backfill_400(self): # «Ошибка стартовой загрузки» jql = "bad jql" self.fake.add("X", updated=CHANGED) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) orig = self.fake.route def route(method, path, q, body): if q.get("jql") == jql: return 400, {"errorMessages": ["Значение не существует"]} return orig(method, path, q, body) self.fake.route = route out = self.refresh({"since": SINCE, "jql": jql}) h = out["result"]["history"] self.assertEqual(out["status"], "done") self.assertTrue(h["backfill_error"].startswith("HTTP 400")) self.assertNotIn("error", h) self.assertEqual([c["key"] for c in self.sent()], ["X"]) def test_incomplete(self): # «Неполный журнал» self.fake.add("X", updated=CHANGED) for n in range(250): self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) self.fake.changelog_cap = 100 out = self.refresh({"since": SINCE}) self.assertEqual(out["result"]["history"]["incomplete"], ["X"]) self.assertEqual(len(self.sent()), 100) self.assertTrue(any("X: получено 100 из 250" in l for l in self.logs)) self.assertIn(("GET", "/rest/api/2/issue/X"), self.fake.requests) def test_truncated_search_completed_by_issue(self): self.fake.add("X", updated=CHANGED) for n in range(150): self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) self.fake.changelog_cap = 100 self.fake.issue_changelog_cap = 150 out = self.refresh({"since": SINCE}) self.assertEqual(out["result"]["history"], {"issues": 1, "changes": 150, "incomplete": []}) def test_custom_field_lookup_once(self): # «Плановая дата без fieldId» self.fake.add("X", updated=CHANGED) self.fake.add("Y", updated=CHANGED) for k in ("X", "Y"): self.fake.add_history(k, "2026-10-22T12:10:00.000+0300", [_st("Плановое окончание", "2026-10-25", "2026-10-30", fieldtype="custom")]) self.refresh({"since": SINCE}) self.assertEqual([c["field_id"] for c in self.sent()], ["customfield_10201", "customfield_10201"]) self.assertEqual(self.fake.requests.count(("GET", "/rest/api/2/field")), 1) def test_hub_422_degrades(self): # Review Focus 3 self.fake.add("X", updated=CHANGED) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) self.hub.reject_history = (422, {"error": "bad_history", "detail": "changes[0].seq"}) out = self.refresh({"since": SINCE}) self.assertEqual(out["status"], "done") self.assertTrue(out["result"]["history"]["error"].startswith("хаб: ")) self.assertNotIn("retryable", out) def test_hub_down_fails(self): self.fake.add("X", updated=CHANGED) self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) self.hub.reject_history = "drop" out = self.refresh({"since": SINCE}) self.assertEqual(out["status"], "failed") self.assertTrue(out["retryable"]) def test_batches_of_500(self): self.fake.add("X", updated=CHANGED) for n in range(600): self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) out = self.refresh({"since": SINCE}) self.assertEqual([len(b["changes"]) for b in self.hub.histories], [500, 100]) self.assertEqual(out["result"]["history"]["changes"], 600) def test_no_history_param(self): # «Хаб без журнала» self.fake.add("X", updated=CHANGED) out = self.refresh() self.assertNotIn("history", out["result"]) self.assertEqual(self.changelog_queries(), []) def test_version(self): self.assertEqual(agent.AGENT_VERSION, 16) if __name__ == "__main__": unittest.main()