fbc9ca9048
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
195 lines
12 KiB
Python
195 lines
12 KiB
Python
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()
|