jira-agent/tests/test_executor_links.py

import unittest

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):
        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_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_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_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_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_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()