import unittest from datetime import datetime from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_jira import FakeJira B = {"board_id": 4216} class ExecutorSprintsTest(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") def tearDown(self): self.fake.stop() def run_cmd(self, ctype, **params): return self.ex.run({"id": 1, "type": ctype, "params": params}) # --- ensure_sprint_meta --- def test_meta_create_then_repeat(self): # сценарий «Создание спринта повтором не дублируется» self.fake.board_sprints[4216] = [{"id": 3, "name": "AWPCA_Sprint_3_Corrections", "state": "active"}] params = dict(B, sprint="AWPCA_Sprint_4", start_date="2026-10-28", end_date="2026-11-11", goal="свод ВГО") out = self.run_cmd("ensure_sprint_meta", **params) self.assertEqual(out["status"], "done", out) sid = out["result"]["sprint_id"] self.assertEqual(out["result"], {"sprint_id": sid, "created": True}) sp = self.fake.board_sprints[4216][-1] off = datetime.now().astimezone().strftime("%z") self.assertEqual((sp["id"], sp["name"], sp["state"], sp["originBoardId"], sp["goal"]), (sid, "AWPCA_Sprint_4", "future", 4216, "свод ВГО")) self.assertEqual((sp["startDate"], sp["endDate"]), ("2026-10-28T09:00:00.000" + off, "2026-11-11T18:00:00.000" + off)) again = self.run_cmd("ensure_sprint_meta", **params) self.assertEqual(again["result"], {"sprint_id": sid, "created": False}) self.assertEqual(len(self.fake.writes()), 1) def test_meta_updates_only_diff(self): # даты совпадают по дню, отличается цель self.fake.board_sprints[4216] = [{"id": 9, "name": "S", "state": "future", "goal": "старая", "startDate": "2026-10-28T10:00:00.000+03:00", "endDate": "2026-11-11T19:00:00.000+03:00"}] out = self.run_cmd("ensure_sprint_meta", sprint="S", start_date="2026-10-28", end_date="2026-11-11", goal="новая", **B) self.assertEqual(out["result"], {"sprint_id": 9, "created": False}) self.assertEqual(self.fake.sprint_posts, [("/rest/agile/1.0/sprint/9", {"goal": "новая"})]) def test_meta_rename_and_name_taken(self): # S → T: POST {"name": "T"}; есть S и T — «имя занято», без записи self.fake.board_sprints[4216] = [{"id": 9, "name": "S", "state": "future"}] out = self.run_cmd("ensure_sprint_meta", sprint="S", rename_to="T", **B) self.assertEqual(out["result"], {"sprint_id": 9, "created": False}) self.assertEqual(self.fake.sprint_posts, [("/rest/agile/1.0/sprint/9", {"name": "T"})]) self.fake.sprint_posts, self.fake.requests = [], [] self.fake.board_sprints[4216] = [{"id": 9, "name": "S", "state": "future"}, {"id": 10, "name": "T", "state": "future"}] out = self.run_cmd("ensure_sprint_meta", sprint="S", rename_to="T", **B) self.assertEqual(out["status"], "failed") self.assertEqual(out["error"], "имя занято") self.assertEqual(self.fake.writes(), []) def test_meta_repeat_after_rename(self): # Review Focus 3, R-69 self.fake.board_sprints[4216] = [{"id": 9, "name": "T", "state": "future"}] out = self.run_cmd("ensure_sprint_meta", sprint="S", rename_to="T", **B) self.assertEqual(out["result"], {"sprint_id": 9, "created": False}) self.assertEqual(self.fake.writes(), []) def test_meta_closed_and_duplicates(self): self.fake.board_sprints[4216] = [{"id": 1, "name": "C", "state": "closed"}, {"id": 5, "name": "D", "state": "future"}, {"id": 6, "name": "D", "state": "active"}] self.assertEqual(self.run_cmd("ensure_sprint_meta", sprint="C", goal="x", **B)["error"], "спринт закрыт") self.assertEqual(self.run_cmd("ensure_sprint_meta", sprint="D", **B)["error"], "несколько спринтов с именем «D»: 5, 6") self.assertEqual(self.fake.writes(), []) def test_meta_bad_date(self): # Review Focus 5 for bad in ("28.10.2026", "2026-13-01"): out = self.run_cmd("ensure_sprint_meta", sprint="N", start_date=bad, **B) self.assertEqual(out["status"], "failed") self.assertIn("ГГГГ-ММ-ДД", out["error"]) self.assertEqual(self.fake.writes(), []) def test_meta_without_board(self): # R-70 out = self.run_cmd("ensure_sprint_meta", sprint="N") self.assertEqual((out["status"], out["error"]), ("failed", "board_id нет в команде — спринты недоступны")) self.assertEqual(self.fake.requests, []) # --- ensure_sprint_state --- def sprint_with(self, n_done, n_open): # спринт 3 активный, спринт 4 будущий, задачи по имени спринта self.fake.board_sprints[4216] = [{"id": 3, "name": "AWPCA_Sprint_3_Corrections", "state": "active"}, {"id": 4, "name": "AWPCA_Sprint_4", "state": "future"}] open_keys = [] for i in range(n_done + n_open): key = "AURORA-%d" % (2000 + i) self.fake.add(key, sprint="AWPCA_Sprint_3_Corrections", status="Готово" if i < n_done else "Разработка") if i >= n_done: open_keys.append(key) return open_keys def test_close_with_move(self): # сценарий «Завершение спринта с переносом» open_keys = self.sprint_with(25, 12) params = dict(B, sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to="AWPCA_Sprint_4") out = self.run_cmd("ensure_sprint_state", **params) self.assertEqual(out["result"], {"sprint_id": 3, "moved": open_keys}) self.assertEqual({self.fake.issues[k]["sprint"] for k in open_keys}, {"AWPCA_Sprint_4"}) self.assertEqual(self.fake.board_sprints[4216][0]["state"], "closed") writes = len(self.fake.writes()) again = self.run_cmd("ensure_sprint_state", **params) self.assertEqual(again["result"], {"sprint_id": 3, "moved": []}) self.assertEqual(len(self.fake.writes()), writes) def test_close_chunks_and_gap(self): self.sprint_with(0, 60) self.ex.s["pacing"] = {"ensure_sprint_state": 30} self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to="backlog", **B) self.assertEqual([p for _, p in self.fake.writes()], ["/rest/agile/1.0/backlog/issue"] * 2 + ["/rest/agile/1.0/sprint/3"]) self.assertEqual(self.gaps, [30, 30]) self.assertEqual(self.fake.requests.count(("GET", "/rest/agile/1.0/sprint/3/issue")), 2) def test_activate_future(self): self.fake.board_sprints[4216] = [{"id": 4, "name": "S4", "state": "future", "startDate": "2026-10-28T09:00:00.000+03:00", "endDate": "2026-11-11T18:00:00.000+03:00"}] out = self.run_cmd("ensure_sprint_state", sprint="S4", state="active", **B) self.assertEqual(out["result"], {"sprint_id": 4, "moved": []}) self.assertEqual(self.fake.sprint_posts, [("/rest/agile/1.0/sprint/4", { "state": "active", "startDate": "2026-10-28T09:00:00.000+03:00", "endDate": "2026-11-11T18:00:00.000+03:00"})]) self.assertEqual(self.fake.board_sprints[4216][0]["state"], "active") writes = len(self.fake.writes()) again = self.run_cmd("ensure_sprint_state", sprint="S4", state="active", **B) self.assertEqual(again["result"], {"sprint_id": 4, "moved": []}) self.assertEqual(len(self.fake.writes()), writes) self.fake.board_sprints[4216] = [{"id": 5, "name": "S5", "state": "future"}] self.fake.sprint_posts, self.fake.requests = [], [] out = self.run_cmd("ensure_sprint_state", sprint="S5", state="active", **B) self.assertEqual((out["status"], out["error"]), ("failed", "у спринта нет дат: задайте ensure_sprint_meta")) self.assertEqual(self.fake.writes(), []) def test_state_wrong_transitions(self): self.fake.board_sprints[4216] = [{"id": 4, "name": "F", "state": "future"}, {"id": 1, "name": "C", "state": "closed"}] out = self.run_cmd("ensure_sprint_state", sprint="F", state="closed", move_open_to="backlog", **B) self.assertEqual((out["status"], out["error"]), ("failed", "спринт не начат")) out = self.run_cmd("ensure_sprint_state", sprint="C", state="active", **B) self.assertEqual((out["status"], out["error"]), ("failed", "спринт закрыт")) self.assertEqual(self.fake.writes(), []) def test_state_validation(self): self.sprint_with(0, 1) out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="future", **B) self.assertEqual((out["status"], out["error"]), ("failed", "state — active или closed")) out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", **B) self.assertEqual((out["status"], out["error"]), ("failed", "move_open_to обязателен при state = closed")) out = self.run_cmd("ensure_sprint_state", sprint="Нет", state="active", **B) self.assertEqual((out["status"], out["error"]), ("failed", "нет спринта «Нет» на доске 4216")) self.fake.board_sprints[4216] += [{"id": 7, "name": "Д", "state": "future"}, {"id": 8, "name": "Д", "state": "future"}] out = self.run_cmd("ensure_sprint_state", sprint="Д", state="active", **B) self.assertEqual(out["error"], "несколько спринтов с именем «Д»: 7, 8") self.assertEqual(self.fake.writes(), []) def test_state_bad_move_target_no_writes(self): # Review Focus 4 self.sprint_with(1, 1) self.fake.board_sprints[4216].append({"id": 1, "name": "Старый", "state": "closed"}) for target in ("Нет такого", "Старый", "AWPCA_Sprint_3_Corrections"): out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to=target, **B) self.assertEqual(out["status"], "failed", target) self.assertEqual(self.fake.writes(), []) def test_close_columns_unreadable(self): self.sprint_with(1, 1) self.fake.fail_paths["/rest/agile/1.0/board/4216/configuration"] = [403] out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to="backlog", **B) self.assertEqual((out["status"], out["error"]), ("failed", "не удалось прочитать колонки доски")) self.assertEqual(self.fake.writes(), []) def test_close_last_column_unknown_status_no_writes(self): # C-1: пустой набор id последней колонки self.sprint_with(2, 1) self.fake.board_columns[4216] = [("В работе", ["10003"]), ("Готово", [])] out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to="backlog", **B) self.assertEqual((out["status"], out["error"]), ("failed", "не удалось прочитать колонки доски")) self.assertEqual(self.fake.writes(), []) def test_close_status_added_after_cache(self): # C-1: статус добавлен в последнюю колонку после заполнения кэша open_keys = self.sprint_with(0, 1) self.ex.status_names() # кэш имён заполнен до появления нового статуса self.fake.statuses["10010"] = "Принято" self.fake.board_columns[4216] = [("В работе", ["10003"]), ("Готово", ["10006", "10010"])] self.fake.add("AURORA-3000", sprint="AWPCA_Sprint_3_Corrections", status="Принято") out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to="backlog", **B) self.assertEqual(out["result"], {"sprint_id": 3, "moved": open_keys}) self.assertEqual(self.fake.issues["AURORA-3000"]["sprint"], "AWPCA_Sprint_3_Corrections") def test_close_issue_without_status_id_is_open(self): # R-81: нет status.id — незавершённая self.sprint_with(1, 0) self.fake.add("AURORA-3001", sprint="AWPCA_Sprint_3_Corrections", status="Без id в statuses") out = self.run_cmd("ensure_sprint_state", sprint="AWPCA_Sprint_3_Corrections", state="closed", move_open_to="backlog", **B) self.assertEqual(out["result"], {"sprint_id": 3, "moved": ["AURORA-3001"]}) def test_meta_rename_to_closed_name_taken(self): # M-2 self.fake.board_sprints[4216] = [{"id": 1, "name": "T", "state": "closed"}] out = self.run_cmd("ensure_sprint_meta", sprint="S", rename_to="T", **B) self.assertEqual((out["status"], out["error"]), ("failed", "имя занято")) self.fake.board_sprints[4216].append({"id": 9, "name": "S", "state": "future"}) out = self.run_cmd("ensure_sprint_meta", sprint="S", rename_to="T", **B) self.assertEqual((out["status"], out["error"]), ("failed", "имя занято")) self.assertEqual(self.fake.writes(), []) def test_meta_repeat_after_rename_with_closed_namesake(self): # M-2 не ломает R-69 self.fake.board_sprints[4216] = [{"id": 1, "name": "T", "state": "closed"}, {"id": 9, "name": "T", "state": "future"}] out = self.run_cmd("ensure_sprint_meta", sprint="S", rename_to="T", **B) self.assertEqual(out["result"], {"sprint_id": 9, "created": False}) self.assertEqual(self.fake.writes(), []) # --- ensure_rank --- def test_rank_already(self): # сценарий «Ранг уже такой» self.fake.add("AURORA-1612", customfield_10105="0|i0001:") self.fake.add("AURORA-1601", customfield_10105="0|i0002:") out = self.run_cmd("ensure_rank", key="AURORA-1612", before="AURORA-1601") self.assertEqual((out["status"], out["result"]), ("done", {"changed": False})) self.assertEqual(self.fake.writes(), []) def test_rank_before_and_after_write(self): self.fake.add("AURORA-1612", customfield_10105="0|i0003:") self.fake.add("AURORA-1601", customfield_10105="0|i0002:") out = self.run_cmd("ensure_rank", key="AURORA-1612", before="AURORA-1601") self.assertEqual(out["result"], {"changed": True}) self.assertEqual(self.fake.rank_bodies[-1], {"issues": ["AURORA-1612"], "rankBeforeIssue": "AURORA-1601", "rankCustomFieldId": 10105}) self.assertLess(self.fake.issues["AURORA-1612"]["customfield_10105"], self.fake.issues["AURORA-1601"]["customfield_10105"]) self.assertEqual(self.run_cmd("ensure_rank", key="AURORA-1612", before="AURORA-1601")["result"], {"changed": False}) self.assertEqual(self.run_cmd("ensure_rank", key="AURORA-1612", after="AURORA-1601")["result"], {"changed": True}) self.assertGreater(self.fake.issues["AURORA-1612"]["customfield_10105"], self.fake.issues["AURORA-1601"]["customfield_10105"]) def test_rank_null_rank_writes(self): # у задачи нет ранга (None) — сравнивать нечего, пишем self.fake.add("AURORA-1612") self.fake.add("AURORA-1601", customfield_10105="0|i0002:") out = self.run_cmd("ensure_rank", key="AURORA-1612", before="AURORA-1601") self.assertEqual((out["status"], out["result"]), ("done", {"changed": True})) self.assertEqual(self.fake.rank_bodies[-1], {"issues": ["AURORA-1612"], "rankBeforeIssue": "AURORA-1601", "rankCustomFieldId": 10105}) self.assertLess(self.fake.issues["AURORA-1612"]["customfield_10105"], self.fake.issues["AURORA-1601"]["customfield_10105"]) def test_rank_bad_params(self): # до любых запросов к Jira — ни поиска, ни записи self.fake.add("AURORA-1612", customfield_10105="0|i0003:") self.fake.add("AURORA-1601", customfield_10105="0|i0002:") out = self.run_cmd("ensure_rank", key="AURORA-1612", before="AURORA-1601", after="AURORA-1601") self.assertEqual((out["status"], out["error"]), ("failed", "нужно ровно одно из before, after")) out = self.run_cmd("ensure_rank", key="AURORA-1612") self.assertEqual((out["status"], out["error"]), ("failed", "нужно ровно одно из before, after")) out = self.run_cmd("ensure_rank", key="AURORA-1612", before="AURORA-1612") self.assertEqual((out["status"], out["error"]), ("failed", "задача не может стоять выше или ниже самой себя")) self.assertEqual(self.fake.writes(), []) def test_rank_no_field(self): self.fake.add("AURORA-1"); self.fake.add("AURORA-2") self.fake.fields = [] self.assertEqual(self.run_cmd("ensure_rank", key="AURORA-1", after="AURORA-2")["error"], "поле ранга не найдено") def test_rank_207_error(self): # Review Focus 1 self.fake.add("AURORA-1", customfield_10105="0|i0009:"); self.fake.add("AURORA-2", customfield_10105="0|i0001:") self.fake.rank_errors["AURORA-1"] = "Ранжирование недоступно" out = self.run_cmd("ensure_rank", key="AURORA-1", before="AURORA-2") self.assertEqual((out["status"], out["error"]), ("failed", "Jira не переранжировала AURORA-1: Ранжирование недоступно"))