import unittest from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_hub import FakeHub from fake_jira import FakeJira class RefreshTest(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 def tearDown(self): self.fake.stop() self.hub.stop() def test_to_state_and_links(self): self.fake.add("AURORA-1", assignee="ivanov", customfield_10801="2026-10-20T00:00:00.000+0300", customfield_15302="AURORA-1350", labels=["AWPCA", "kind-dev"], sprint="Спринт 2") self.fake.add("AURORA-2") self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-1", "target": "AURORA-2"}) raw = self.fake.issue_json("AURORA-1") st = self.ex.to_state(raw) self.assertEqual(st["key"], "AURORA-1") self.assertEqual(st["assignee"], "ivanov") self.assertEqual(st["planned_start"], "2026-10-20") self.assertEqual(st["epic_key"], "AURORA-1350") self.assertEqual(st["labels"], ["AWPCA", "kind-dev"]) self.assertEqual(st["issue_type"], "Задача") self.assertEqual(st["jira_updated"], "2026-10-01T10:00:00.000+0300") self.assertIsNone(st["sprint"]) # спринт заполняет do_refresh self.assertEqual(self.ex.links_from(raw), [{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "ASUPR_Блокер_Начала"}]) self.assertEqual(self.ex.links_from(self.fake.issue_json("AURORA-2")), [{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "ASUPR_Блокер_Начала"}]) self.assertIn("customfield_15302", self.ex.state_fields()) def test_full_refresh_with_neighbors_and_sprints(self): for i in range(1, 4): self.fake.add("AURORA-%d" % i, sprint="Спринт 2" if i < 3 else None) self.fake.add("OTHER-9", in_base=False) self.fake.add("AURORA-7", in_base=False) self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "OTHER-9"}) self.fake.links.append({"id": 2, "type": "ASUPR_Связанные", "source": "AURORA-7", "target": "AURORA-2"}) self.hub.known = ["AURORA-7"] out = self.ex.run({"id": 5, "type": "refresh", "params": {"jql": "project = AURORA AND labels = AWPCA"}}) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"issues": 3, "neighbors": 1, "pages": 1}) state = self.hub.states[0] self.assertEqual(state["command_id"], 5) self.assertEqual(sorted(i["key"] for i in state["issues"]), ["AURORA-1", "AURORA-2", "AURORA-3", "OTHER-9"]) self.assertEqual(state["neighbor_keys"], ["OTHER-9"]) self.assertTrue(state["last_page"]) sprints = {i["key"]: i["sprint"] for i in state["issues"]} self.assertEqual(sprints["AURORA-1"], "Спринт 2") self.assertIsNone(sprints["AURORA-3"]) links = {(l["source_key"], l["target_key"]) for l in state["links"]} self.assertEqual(links, {("AURORA-1", "OTHER-9"), ("AURORA-7", "AURORA-2")}) searches = [r for r in self.fake.requests if r[1] == "/rest/api/2/search"] self.assertEqual(len(searches), 2) # базовая выборка + дочитывание соседей def test_refresh_by_keys_pages(self): for i in range(1, 121): self.fake.add("AURORA-%d" % i) keys = ["AURORA-%d" % i for i in range(1, 121)] out = self.ex.run({"id": 6, "type": "refresh", "params": {"keys": keys}}) self.assertEqual(out["result"], {"issues": 120, "neighbors": 0, "pages": 2}) self.assertEqual(len(self.hub.states), 2) self.assertEqual(self.hub.states[0]["page"], 1) self.assertFalse(self.hub.states[0]["last_page"]) self.assertTrue(self.hub.states[1]["last_page"]) self.assertEqual(len(self.hub.states[0]["issues"]) + len(self.hub.states[1]["issues"]), 120) def test_refresh_empty_keys(self): out = self.ex.run({"id": 7, "type": "refresh", "params": {"keys": []}}) self.assertEqual(out["result"], {"issues": 0, "neighbors": 0, "pages": 0}) self.assertEqual(self.fake.requests, []) self.assertEqual(self.hub.states, []) def test_comments_in_state(self): self.fake.add("AURORA-1") self.fake.add("AURORA-2") self.fake.add_comment("AURORA-1", "первый\nвторая строка", name="petrov", display="Петров Пётр", created="2026-10-14T10:00:00.000+0300") self.fake.add_comment("AURORA-1", "второй", name="ivanov", display=None, created="2026-10-14T11:00:00.000+0300", updated="2026-10-14T12:00:00.000+0300") self.fake.add_comment("AURORA-1", None, name=None, display=None) # автор удалён, текста нет out = self.ex.run({"id": 8, "type": "refresh", "params": {"keys": ["AURORA-1", "AURORA-2"]}}) self.assertEqual(out["status"], "done") by_key = {i["key"]: i for i in self.hub.states[0]["issues"]} c = by_key["AURORA-1"]["comments"] self.assertEqual([x["author"] for x in c], ["Петров Пётр", "ivanov", None]) self.assertEqual(c[0], {"id": c[0]["id"], "author": "Петров Пётр", "created": "2026-10-14T10:00:00.000+0300", "updated": "2026-10-14T10:00:00.000+0300", "body": "первый\nвторая строка"}) self.assertIsInstance(c[0]["id"], str) self.assertEqual(c[1]["updated"], "2026-10-14T12:00:00.000+0300") self.assertEqual(c[2]["body"], "") self.assertEqual(by_key["AURORA-2"]["comments"], []) self.assertIn("comment", self.ex.state_fields().split(",")) searches = [r for r in self.fake.requests if r[1] == "/rest/api/2/search"] self.assertEqual(len(searches), 1) # комментарии пришли тем же search self.assertFalse(any(r[1].endswith("/comment") for r in self.fake.requests)) def test_refresh_attachments_in_state(self): # сценарий «Вложения в состоянии» self.fake.add("AURORA-1650") self.fake.add_attachment("AURORA-1650", "скрин 1.png", b"x" * 240000, mime="image/png") self.fake.add_attachment("AURORA-1650", "ОСВ 60.xlsx", b"y" * 10, name="ivanov", display=None) out = self.ex.run({"id": 5, "type": "refresh", "params": {"keys": ["AURORA-1650"]}}) self.assertEqual(out["status"], "done") st = self.hub.states[0]["issues"][0] a = st["attachments"] self.assertEqual([x["filename"] for x in a], ["скрин 1.png", "ОСВ 60.xlsx"]) self.assertEqual(a[0], {"id": a[0]["id"], "filename": "скрин 1.png", "size": 240000, "mime_type": "image/png", "author": "Потапова Анна", "created": "2026-10-20T09:15:00.000+0300"}) self.assertIsInstance(a[0]["id"], str) self.assertEqual(a[1]["author"], "ivanov") self.assertIn("attachment", self.ex.state_fields().split(",")) self.assertEqual(sum(1 for r in self.fake.requests if r[1].endswith("/search")), 1) self.assertFalse(any("/secure/attachment/" in r[1] for r in self.fake.requests)) def test_refresh_attachment_without_id_dropped(self): self.fake.add("AURORA-1") self.fake.add_attachment("AURORA-1", "a.txt", b"a") self.fake.add_attachment("AURORA-1", "b.txt", b"b") self.fake.attachments["AURORA-1"][0]["id"] = None self.ex.run({"id": 5, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) a = self.hub.states[0]["issues"][0]["attachments"] self.assertEqual([x["filename"] for x in a], ["b.txt"]) def test_refresh_no_attachments_empty_list(self): self.fake.add("AURORA-1") self.fake.add("AURORA-2") self.fake.add_attachment("AURORA-2", "a.txt", b"a") self.fake.attachments["AURORA-2"][0]["size"] = None self.ex.run({"id": 5, "type": "refresh", "params": {"keys": ["AURORA-1", "AURORA-2"]}}) by_key = {i["key"]: i for i in self.hub.states[0]["issues"]} self.assertEqual(by_key["AURORA-1"]["attachments"], []) self.assertIsNone(by_key["AURORA-2"]["attachments"][0]["size"]) self.fake.hide_attachment_field = True self.ex.run({"id": 6, "type": "refresh", "params": {"keys": ["AURORA-2"]}}) self.assertEqual(self.hub.states[-1]["issues"][0]["attachments"], []) def test_comments_partial_list_is_read_fully(self): self.fake.add("AURORA-1") for n in range(130): self.fake.add_comment("AURORA-1", "к%d" % n) self.fake.search_comment_cap = 20 out = self.ex.run({"id": 9, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) self.assertEqual(out["status"], "done") c = self.hub.states[0]["issues"][0]["comments"] self.assertEqual(len(c), 130) self.assertEqual([x["body"] for x in c[:2]] + [c[-1]["body"]], ["к0", "к1", "к129"]) reads = [r for r in self.fake.requests if r == ("GET", "/rest/api/2/issue/AURORA-1/comment")] self.assertEqual(len(reads), 2) # 100 + 30 def test_comments_field_missing(self): raw = {"key": "AURORA-1", "fields": {}} # поле comment недоступно self.assertEqual(self.ex.comments_of(raw), []) raw = {"key": "AURORA-1", "fields": {"comment": {"comments": [{"id": "5", "body": "x"}]}}} # total нет self.assertEqual(self.ex.comments_of(raw), [{"id": "5", "author": None, "created": None, "updated": None, "body": "x"}]) self.assertEqual(self.fake.requests, []) # без дочитывания def test_comment_without_id_dropped(self): self.fake.add("AURORA-1") self.fake.add_comment("AURORA-1", "нормальный") self.fake.add_comment("AURORA-1", "второй")["id"] = None self.fake.add_comment("AURORA-1", "третий")["id"] = "" logs = [] orig = agent.log agent.log = logs.append try: out = self.ex.run({"id": 10, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) finally: agent.log = orig self.assertEqual(out["status"], "done") c = self.hub.states[0]["issues"][0]["comments"] self.assertEqual([x["body"] for x in c], ["нормальный"]) self.assertEqual(len([l for l in logs if "AURORA-1" in l and "комментарий без id" in l]), 2, logs) def test_comments_read_denied_keeps_refresh(self): for key in ("AURORA-1", "AURORA-2", "AURORA-3"): self.fake.add(key) for n in range(3): self.fake.add_comment(key, "%s-%d" % (key, n)) self.fake.search_comment_cap = 1 # search отдаёт 1 из 3 — нужно дочитывание self.fake.comment_denied = {"AURORA-1": 403, "AURORA-2": 404} logs = [] orig = agent.log agent.log = logs.append try: out = self.ex.run({"id": 11, "type": "refresh", "params": {"keys": ["AURORA-1", "AURORA-2", "AURORA-3"]}}) finally: agent.log = orig self.assertEqual(out["status"], "done") by_key = {i["key"]: i for i in self.hub.states[0]["issues"]} self.assertNotIn("comments", by_key["AURORA-1"]) self.assertNotIn("comments", by_key["AURORA-2"]) self.assertEqual([x["body"] for x in by_key["AURORA-3"]["comments"]], ["AURORA-3-0", "AURORA-3-1", "AURORA-3-2"]) self.assertTrue(any("AURORA-1" in l and "HTTP 403" in l for l in logs), logs) self.assertTrue(any("AURORA-2" in l and "HTTP 404" in l for l in logs), logs) def test_comments_read_5xx_still_fails_refresh(self): self.fake.add("AURORA-1") for n in range(3): self.fake.add_comment("AURORA-1", "к%d" % n) self.fake.search_comment_cap = 1 self.fake.comment_denied = {"AURORA-1": 500} out = self.ex.run({"id": 12, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) self.assertEqual(out["status"], "failed") self.assertTrue(out["retryable"]) # 5xx — по-прежнему повтор через проверку (R-10) self.assertEqual(self.hub.states, []) def test_refresh_empty_jql_result(self): out = self.ex.run({"id": 8, "type": "refresh", "params": {"jql": "project = AURORA AND labels = AWPCA"}}) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"issues": 0, "neighbors": 0, "pages": 0}) self.assertEqual(self.hub.states, []) def test_keys_with_deleted_key(self): for i in (1, 2, 3, 4): self.fake.add("AURORA-%d" % i) out = self.ex.run({"id": 9, "type": "refresh", "params": {"keys": ["AURORA-1", "AURORA-2", "AURORA-99", "AURORA-3", "AURORA-98"]}}) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"issues": 3, "neighbors": 0, "pages": 1, "missing": ["AURORA-99", "AURORA-98"]}) self.assertEqual(sorted(i["key"] for i in self.hub.states[0]["issues"]), ["AURORA-1", "AURORA-2", "AURORA-3"]) def test_all_keys_400_fails(self): out = self.ex.run({"id": 14, "type": "refresh", "params": {"keys": ["X-1", "X-2", "X-3"]}}) self.assertEqual(out["status"], "failed") self.assertIn("HTTP 400", out["error"]) self.assertEqual(self.hub.states, []) def test_bisect_cost_one_missing_of_100(self): for i in range(1, 100): self.fake.add("AURORA-%d" % i) keys = ["AURORA-%d" % i for i in range(1, 101)] # AURORA-100 отсутствует out = self.ex.run({"id": 15, "type": "refresh", "params": {"keys": keys}}) self.assertEqual(out["result"]["missing"], ["AURORA-100"]) searches = [r for r in self.fake.requests if r[1] == "/rest/api/2/search"] self.assertLessEqual(len(searches), 2 * 7 + 1) def test_missing_deduped_and_truncated(self): self.fake.add("AURORA-1") keys = ["AURORA-1"] + ["GONE-%d" % i for i in range(1, 61)] + ["GONE-1"] out = self.ex.run({"id": 16, "type": "refresh", "params": {"keys": keys}}) self.assertEqual(out["result"]["missing"], ["GONE-%d" % i for i in range(1, 51)]) def test_missing_not_duplicated_when_also_a_neighbor(self): self.fake.add("AURORA-1") self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "GONE-1"}) out = self.ex.run({"id": 17, "type": "refresh", "params": {"keys": ["AURORA-1", "GONE-1"]}}) self.assertEqual(out["status"], "done") self.assertEqual(out["result"]["missing"], ["GONE-1"]) def test_input_keys_deduplicated(self): self.fake.add("AURORA-1") out = self.ex.run({"id": 18, "type": "refresh", "params": {"keys": ["AURORA-1", "AURORA-1", "", "AURORA-1"]}}) self.assertEqual(out["result"], {"issues": 1, "neighbors": 0, "pages": 1}) self.assertEqual(len(self.hub.states[0]["issues"]), 1) self.assertEqual(self.hub.states[0]["agent_version"], agent.AGENT_VERSION) def test_deleted_neighbor(self): self.fake.add("AURORA-1") self.fake.add("AURORA-2", in_base=False) self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "GONE-1"}) self.fake.links.append({"id": 2, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "AURORA-2"}) out = self.ex.run({"id": 10, "type": "refresh", "params": {"jql": "project = AURORA AND labels = AWPCA"}}) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"issues": 1, "neighbors": 1, "pages": 1, "missing": ["GONE-1"]}) self.assertEqual(self.hub.states[0]["neighbor_keys"], ["AURORA-2"]) def test_non_400_error_fails(self): self.fake.add("AURORA-1") def boom(jql): raise agent.HttpError(500, "oops") self.ex.search_all = boom out = self.ex.run({"id": 11, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) self.assertEqual(out["status"], "failed") self.assertIn("500", out["error"]) def test_link_visible_from_both_ends_deduped(self): self.fake.add("AURORA-1") self.fake.add("AURORA-2") self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "AURORA-2"}) self.ex.run({"id": 12, "type": "refresh", "params": {"keys": ["AURORA-1", "AURORA-2"]}}) self.assertEqual(len(self.hub.states[0]["links"]), 1) def test_many_neighbors_several_requests(self): self.fake.add("AURORA-1") for i in range(1, 151): self.fake.add("NB-%d" % i, in_base=False) self.fake.links.append({"id": i, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "NB-%d" % i}) out = self.ex.run({"id": 13, "type": "refresh", "params": {"jql": "project = AURORA AND labels = AWPCA"}}) self.assertEqual(out["result"], {"issues": 1, "neighbors": 150, "pages": 2}) searches = [r for r in self.fake.requests if r[1] == "/rest/api/2/search"] self.assertEqual(len(searches), 1 + 2) # база + два чанка соседей def test_sprint_names_paginated(self): for i in range(1, 121): self.fake.add("AURORA-%d" % i, sprint="Спринт 2") names = self.ex.sprint_names(["AURORA-%d" % i for i in range(1, 101)], [4216]) self.assertEqual(len(names), 100) pages = [r for r in self.fake.requests if r[1] == "/rest/agile/1.0/sprint/2/issue"] self.assertEqual(len(pages), 2) def test_fake_hub_terminal_result_not_overwritten(self): cid = self.hub.enqueue("refresh", {"keys": []}) hub = agent.HubClient(jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)) hub.result(cid, {"status": "done", "result": {"a": 1}}) hub.result(cid, {"status": "failed", "error": "x"}) self.assertEqual(self.hub.commands[cid]["status"], "done") self.assertEqual(self.hub.results[cid]["status"], "done") def test_hub_unreachable_in_refresh_names_hub(self): self.fake.add("AURORA-1") self.ex.hub = agent.HubClient({"hub_url": "http://127.0.0.1:1", "hub_token": "t", "timeout": 2}) out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) self.assertEqual(out["status"], "failed") self.assertTrue(out["error"].startswith("хаб: "), out["error"]) self.assertTrue(out["retryable"]) self.assertEqual(out["peer"], "хабом") def test_hub_4xx_in_refresh_not_retryable(self): self.fake.add("AURORA-1") self.ex.hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": "wrong-token", "timeout": 5}) out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) self.assertEqual(out["status"], "failed") self.assertTrue(out["error"].startswith("хаб: HTTP 401"), out["error"]) self.assertFalse(out["retryable"]) def test_jira_error_has_no_peer(self): self.fake.fail_next.append((500, {})) out = self.ex.run({"id": 1, "type": "refresh", "params": {"jql": "project = AURORA"}}) self.assertTrue(out["retryable"]) self.assertNotIn("peer", out) JQL = "project = AURORA AND labels in (AWPCA, STOCKS_DRRK)" def refresh(self, **params): return self.ex.run({"id": 5, "type": "refresh", "params": params}) def logged(self, fn): logs, orig = [], agent.log agent.log = logs.append try: return fn(), logs finally: agent.log = orig def test_two_boards_one_new(self): # сценарий «Две доски, одна новая» self.fake.boards[5120] = "kanban" self.fake.add("AURORA-1", sprint="Спринт 2") out = self.refresh(jql=self.JQL, boards=[{"id": 4216, "type": "scrum"}, {"id": 5120, "type": None}]) self.assertEqual(out["status"], "done", out) self.assertEqual(out["result"]["boards"], {"5120": "kanban"}) paths = [r[1] for r in self.fake.requests] self.assertEqual(paths.count("/rest/agile/1.0/board/5120"), 1) self.assertNotIn("/rest/agile/1.0/board/5120/sprint", paths) self.assertNotIn("/rest/agile/1.0/board/4216", paths) # тип известен — не спрашиваем self.assertEqual(self.hub.states[0]["issues"][0]["sprint"], "Спринт 2") def test_board_not_found(self): # сценарий «Доска не найдена» self.fake.add("AURORA-1") out, logs = self.logged(lambda: self.refresh(jql=self.JQL, boards=[{"id": 9999, "type": None}])) self.assertEqual(out["status"], "done", out) self.assertEqual(out["result"]["boards"], {"9999": None}) self.assertTrue(any("9999" in l and "HTTP 404" in l for l in logs), logs) self.assertEqual(len(self.hub.states), 1) def test_new_scrum_board_sprints_read(self): self.fake.boards[5120] = "scrum" self.fake.board_sprints[5120] = [{"id": 7, "name": "Спринт K", "state": "active"}] self.fake.add("AURORA-1", sprint="Спринт K") out = self.refresh(jql=self.JQL, boards=[{"id": 5120, "type": None}]) self.assertEqual(out["result"]["boards"], {"5120": "scrum"}) self.assertEqual(self.hub.states[0]["issues"][0]["sprint"], "Спринт K") self.assertNotIn("/rest/agile/1.0/board/4216/sprint", [r[1] for r in self.fake.requests]) def test_old_hub_without_boards(self): # сценарий «Старый хаб» self.fake.add("AURORA-1", sprint="Спринт 2") out = self.refresh(jql=self.JQL) self.assertNotIn("boards", out["result"]) self.assertIn("/rest/agile/1.0/board/4216/sprint", [r[1] for r in self.fake.requests]) self.assertEqual(self.hub.states[0]["issues"][0]["sprint"], "Спринт 2") def test_known_boards_no_type_requests(self): self.fake.add("AURORA-1", sprint="Спринт 2") out = self.refresh(jql=self.JQL, boards=[{"id": 4216, "type": "scrum"}]) self.assertNotIn("boards", out["result"]) self.assertNotIn("/rest/agile/1.0/board/4216", [r[1] for r in self.fake.requests]) self.assertEqual(self.hub.states[0]["issues"][0]["sprint"], "Спринт 2") def test_empty_boards_no_settings_fallback(self): self.fake.add("AURORA-1", sprint="Спринт 2") out = self.refresh(jql=self.JQL, boards=[]) self.assertEqual(out["status"], "done", out) self.assertNotIn("boards", out["result"]) self.assertFalse(any("/rest/agile/1.0/board/" in r[1] for r in self.fake.requests)) self.assertIsNone(self.hub.states[0]["issues"][0]["sprint"]) def test_board_type_5xx_logged_refresh_continues(self): self.fake.add("AURORA-1") self.fake.fail_next.append((500, {})) # первый вызов Jira в refresh — тип доски out, logs = self.logged(lambda: self.refresh(jql=self.JQL, boards=[{"id": 5120, "type": None}])) self.assertEqual(out["status"], "done", out) self.assertEqual(out["result"]["boards"], {"5120": None}) self.assertTrue(any("5120" in l and "HTTP 500" in l for l in logs), logs) self.assertEqual(len(self.hub.states), 1) # M-4: состояние ушло в хаб def test_board_other_type_is_null(self): self.fake.boards[7000] = "simple" out, logs = self.logged(lambda: self.refresh(keys=[], boards=[{"id": 7000, "type": None}])) self.assertEqual(out["result"]["boards"], {"7000": None}) self.assertTrue(any("7000" in l and "simple" in l for l in logs), logs) def test_duplicate_null_board_requested_once(self): self.fake.boards[5120] = "kanban" out = self.refresh(keys=[], boards=[{"id": 5120, "type": None}, {"id": 5120, "type": None}]) self.assertEqual(out["result"]["boards"], {"5120": "kanban"}) self.assertEqual([r[1] for r in self.fake.requests].count("/rest/agile/1.0/board/5120"), 1) def test_empty_keys_still_resolves_boards(self): self.fake.boards[5120] = "kanban" out = self.refresh(keys=[], boards=[{"id": 5120, "type": None}]) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"issues": 0, "neighbors": 0, "pages": 0, "boards": {"5120": "kanban"}}) def test_last_found_sprint_wins_across_boards(self): self.fake.boards[5120] = "scrum" self.fake.board_sprints[5120] = [{"id": 7, "name": "Спринт K", "state": "active"}] self.fake.add("AURORA-1") self.fake.sprint_members = {2: ["AURORA-1"], 7: ["AURORA-1"]} both = [{"id": 4216, "type": "scrum"}, {"id": 5120, "type": "scrum"}] self.refresh(jql=self.JQL, boards=both) self.assertEqual(self.hub.states[-1]["issues"][0]["sprint"], "Спринт K") self.refresh(jql=self.JQL, boards=both[::-1]) self.assertEqual(self.hub.states[-1]["issues"][0]["sprint"], "Спринт 2") def test_scrum_board_turned_kanban_self_heals(self): # I-1, R-36 self.fake.boards[5120] = "kanban" # хаб думает, что scrum self.fake.add("AURORA-1", sprint="Спринт 2") out = self.refresh(jql=self.JQL, boards=[{"id": 4216, "type": "scrum"}, {"id": 5120, "type": "scrum"}]) self.assertEqual(out["status"], "done", out) self.assertEqual(out["result"]["boards"], {"5120": "kanban"}) self.assertEqual(self.hub.states[0]["issues"][0]["sprint"], "Спринт 2") def test_scrum_board_sprints_error_fails_refresh(self): # R-32, I-1: ошибка называет доску self.fake.add("AURORA-1") out = self.refresh(jql=self.JQL, boards=[{"id": 9999, "type": "scrum"}]) # доски нет в Jira self.assertEqual(out["status"], "failed") self.assertIn("HTTP 404", out["error"]) self.assertIn("доска 9999", out["error"]) self.assertFalse(out["retryable"]) self.assertEqual(self.hub.states, []) def test_non_dict_board_item_skipped(self): # M-2 self.fake.boards[5120] = "kanban" out = self.refresh(keys=[], boards=[4216, None, {"id": 5120, "type": None}]) self.assertEqual(out["status"], "done", out) self.assertEqual(out["result"]["boards"], {"5120": "kanban"}) if __name__ == "__main__": unittest.main() class JiraFieldsTest(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 def tearDown(self): self.fake.stop() self.hub.stop() def logged(self, fn): logs, orig = [], agent.log agent.log = logs.append try: return fn(), logs finally: agent.log = orig def test_fields_found_by_schema_and_name(self): want = {"rank": "customfield_10105", "sprint": "customfield_10104", "story_points": "customfield_10106"} self.assertEqual(self.ex.jira_fields(), want) self.ex.run({"id": 2, "type": "nope", "params": {}}) # новая команда — кэш процесса жив self.assertEqual(self.ex.jira_fields(), want) self.assertEqual(self.fake.requests.count(("GET", "/rest/api/2/field")), 1) def test_fields_override_skips_lookup(self): self.ex.s["field_ids"] = dict(SETTINGS["field_ids"], rank="cf_r", sprint="cf_s", story_points="cf_p") self.assertEqual(self.ex.jira_fields(), {"rank": "cf_r", "sprint": "cf_s", "story_points": "cf_p"}) self.assertEqual(self.fake.requests, []) def test_fields_partial_override(self): self.ex.s["field_ids"] = dict(SETTINGS["field_ids"], story_points="customfield_20000") self.assertEqual(self.ex.jira_fields()["story_points"], "customfield_20000") self.assertEqual(self.ex.jira_fields()["rank"], "customfield_10105") def test_story_points_first_by_numeric_id_any_case(self): self.fake.fields += [{"id": "customfield_9999", "name": " story points ", "custom": True}, {"id": "customfield_10050", "name": "Story Points", "custom": True}] self.assertEqual(self.ex.jira_fields()["story_points"], "customfield_9999") def test_field_lookup_error_retried_next_command(self): self.fake.fail_next.append((403, {})) out, logs = self.logged(lambda: (self.ex.jira_fields(), self.ex.jira_fields())) # logged — как в RefreshTest self.assertEqual(out[0], {"rank": None, "sprint": None, "story_points": None}) self.assertTrue(any("HTTP 403" in l for l in logs), logs) self.assertEqual(self.fake.requests.count(("GET", "/rest/api/2/field")), 1) # в той же команде — не повторяем self.ex.run({"id": 2, "type": "nope", "params": {}}) self.assertEqual(self.ex.jira_fields()["rank"], "customfield_10105")