181 lines
11 KiB
Python
181 lines
11 KiB
Python
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(), [])
|