import unittest from unittest import mock import fake_jira from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_jira import FakeJira class ExecutorLinksTest(unittest.TestCase): def setUp(self): self.fake = FakeJira() self.jira = agent.JiraClient(jira_cfg(self.fake)) self.jira.sleep = no_sleep self.ex = agent.Executor(self.jira, dict(SETTINGS)) self.gaps = [] self.ex.sleep = self.gaps.append self.fake.add("AURORA-1") self.fake.add("AURORA-2") def tearDown(self): self.fake.stop() def run_cmd(self, ctype, **params): return self.ex.run({"id": 1, "type": ctype, "params": params}) # --- статус --- def test_status_path(self): out = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"]) self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"from": "Ожидание разработки", "to": "Готово", "steps": 2}) self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Готово") again = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"]) self.assertEqual(again["result"]["steps"], 0) self.assertEqual(again["jira_calls"], 1) def test_status_resumes_from_middle(self): self.fake.issues["AURORA-1"]["status"] = "Разработка" out = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово к тестированию"]) self.assertEqual(out["result"]["steps"], 1) self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Готово к тестированию") def test_status_single_and_by_transition_name(self): # имя перехода принято как шаг, но в result.to — реальный статус после перехода out = self.run_cmd("ensure_status", key="AURORA-1", target="Взять в разработку") self.assertEqual(out["result"]["to"], "Разработка") self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Разработка") def test_status_ambiguous_prefers_target_status(self): # переход «Готово» ведёт в «Анализ», переход «Старт» ведёт в статус «Готово»: выбирается по целевому статусу with mock.patch.dict(fake_jira.WORKFLOW, {"Ожидание разработки": [("Готово", "Анализ"), ("Старт", "Готово")]}): out = self.run_cmd("ensure_status", key="AURORA-1", target="Готово") self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"from": "Ожидание разработки", "to": "Готово", "steps": 1}) self.assertEqual(self.fake.issues["AURORA-1"]["status"], "Готово") def test_status_no_transition_is_failed(self): out = self.run_cmd("ensure_status", key="AURORA-1", target="Тестирование") self.assertEqual(out["status"], "failed") self.assertIn("нет перехода из «Ожидание разработки» в «Тестирование»", out["error"]) self.assertIn("Взять в разработку→Разработка", out["error"]) self.assertEqual(out["result"], {"from": "Ожидание разработки", "reached": "Ожидание разработки"}) def test_status_bad_target(self): cases = (([], "пустой путь"), ("", "пустой путь"), ([""], "пустой путь"), (5, "target должен быть"), (["Готово", 1], "target должен быть"), (None, "target должен быть")) for bad, msg in cases: out = self.run_cmd("ensure_status", key="AURORA-1", target=bad) self.assertEqual(out["status"], "failed", bad) self.assertIn(msg, out["error"]) self.assertEqual(self.fake.writes(), []) def test_status_http_error_is_failed(self): self.fake.fail_next.append((400, {})) out = self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"]) self.assertEqual(out["status"], "failed") self.assertIn("HTTP 400", out["error"]) def test_status_gap_between_steps(self): self.ex.s["pacing"] = {"ensure_status": 7} self.run_cmd("ensure_status", key="AURORA-1", target=["Разработка", "Готово"]) self.assertEqual(self.gaps, [7]) # --- связи --- def test_link_create_and_idempotent(self): out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало") self.assertEqual(out["result"], {"created": True, "removed": []}) link = self.fake.links[0] self.assertEqual((link["source"], link["target"], link["type"]), ("AURORA-1", "AURORA-2", "ASUPR_Блокер_Начала")) again = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало") self.assertEqual(again["result"], {"existed": True, "removed": []}) self.assertEqual(len(self.fake.links), 1) def test_link_symmetric_any_direction(self): self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-2", "target": "AURORA-1"}) out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="связанные") self.assertEqual(out["result"], {"existed": True, "removed": []}) def test_link_conflict_without_replace(self): self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-2", "target": "AURORA-1"}) out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало") self.assertEqual(out["status"], "failed") self.assertIn("обратное направление", out["error"]) self.assertEqual(len(self.fake.links), 1) def test_link_replace_reverse_and_other_type(self): self.ex.s["pacing"] = {"ensure_link": 3} self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-2", "target": "AURORA-1"}) self.fake.links.append({"id": 2, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "AURORA-2"}) out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало", replace=True) self.assertEqual(out["status"], "done") self.assertEqual(sorted(out["result"]["removed"]), ["ASUPR_Блокер_Начала", "ASUPR_Связанные"]) self.assertTrue(out["result"]["created"]) self.assertEqual([(l["source"], l["target"], l["type"]) for l in self.fake.links], [("AURORA-1", "AURORA-2", "ASUPR_Блокер_Начала")]) self.assertEqual(self.gaps, [3, 3]) # между двумя DELETE и перед POST self.assertEqual([r[0] for r in self.fake.writes()], ["DELETE", "DELETE", "POST"]) def test_link_unmanaged_type_ignored(self): self.fake.links.append({"id": 1, "type": "Duplicate", "source": "AURORA-1", "target": "AURORA-2"}) out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="связанные") self.assertEqual(out["result"], {"created": True, "removed": []}) self.assertEqual(len(self.fake.links), 2) def test_link_unknown_type_sent_raw(self): # намеренный fallback (рулинг R-5): неизвестное имя типа уходит в Jira как есть out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="Duplicate") self.assertEqual(out["result"], {"created": True, "removed": []}) self.assertEqual(self.fake.links[-1]["type"], "Duplicate") def test_link_http_error_is_failed(self): self.ex.symmetric_types() # тип уже известен, падает чтение задачи self.fake.fail_next.append((400, {})) out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="блокирует начало") self.assertEqual(out["status"], "failed") self.assertIn("HTTP 400", out["error"]) self.assertEqual(self.fake.links, []) def test_symmetric_not_cached_on_error(self): self.fake.fail_next.append((400, {})) # падает чтение типов связей out = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="связанные") self.assertEqual(out["status"], "failed") self.assertIsNone(self.ex._symmetric) # пустой набор в кэш не попал self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-2", "target": "AURORA-1"}) again = self.run_cmd("ensure_link", source="AURORA-1", target="AURORA-2", link_type="связанные") self.assertEqual(again["result"], {"existed": True, "removed": []}) # --- спринты --- def test_sprint_move_and_idempotent(self): out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 2") self.assertEqual(out["result"], {"sprint": "Спринт 2", "from": None}) self.assertEqual(self.fake.issues["AURORA-1"]["sprint"], "Спринт 2") again = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 2") self.assertEqual(again["result"], {"sprint": "Спринт 2", "changed": False}) self.assertEqual(again["jira_calls"], 1) def test_sprint_pagination(self): for i in range(3, 58): # 57 спринтов: доска отдаёт по 50, нужны две страницы self.fake.sprints.append({"id": i, "name": "Спринт %d" % i, "state": "future"}) out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 57") self.assertEqual(out["status"], "done") self.assertEqual(self.fake.issues["AURORA-1"]["sprint"], "Спринт 57") pages = [r for r in self.fake.requests if r[1].endswith("/board/4216/sprint")] self.assertEqual(len(pages), 2) def test_sprint_backlog(self): self.fake.issues["AURORA-1"]["sprint"] = "Спринт 1" out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="backlog") self.assertEqual(out["result"], {"sprint": None, "from": "Спринт 1"}) self.assertIsNone(self.fake.issues["AURORA-1"]["sprint"]) self.assertEqual(self.run_cmd("ensure_sprint", key="AURORA-1", sprint="backlog")["result"], {"sprint": None, "changed": False}) def test_sprint_backlog_without_board(self): self.ex.s["board_id"] = None self.fake.issues["AURORA-1"]["sprint"] = "Спринт 1" out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="backlog") self.assertEqual(out["status"], "done") self.assertEqual(out["result"], {"sprint": None, "from": "Спринт 1"}) self.assertIsNone(self.fake.issues["AURORA-1"]["sprint"]) def test_sprint_unknown_name(self): out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 9") self.assertEqual(out["status"], "failed") self.assertIn("Спринт 9", out["error"]) self.assertIn("Спринт 2", out["error"]) def test_sprint_without_board(self): self.ex.s["board_id"] = None out = self.run_cmd("ensure_sprint", key="AURORA-1", sprint="Спринт 2") self.assertEqual(out["status"], "failed") self.assertIn("board_id", out["error"]) if __name__ == "__main__": unittest.main()