29fefb6114
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
86 lines
4.5 KiB
Python
86 lines
4.5 KiB
Python
import unittest
|
||
|
||
from _common import SETTINGS, agent, jira_cfg, no_sleep
|
||
from fake_hub import FakeHub
|
||
from fake_jira import FakeJira
|
||
|
||
|
||
class RefreshTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.fake = FakeJira()
|
||
self.hub = FakeHub(SETTINGS)
|
||
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
|
||
self.jira = agent.JiraClient(cfg)
|
||
self.jira.sleep = no_sleep
|
||
self.ex = agent.Executor(self.jira, dict(SETTINGS), agent.HubClient(cfg))
|
||
self.ex.sleep = no_sleep
|
||
|
||
def tearDown(self):
|
||
self.fake.stop()
|
||
self.hub.stop()
|
||
|
||
def test_to_state_and_links(self):
|
||
self.fake.add("AURORA-1", assignee="ivanov", customfield_10801="2026-10-20T00:00:00.000+0300",
|
||
customfield_15302="AURORA-1350", labels=["AWPCA", "kind-dev"], sprint="Спринт 2")
|
||
self.fake.add("AURORA-2")
|
||
self.fake.links.append({"id": 1, "type": "ASUPR_Блокер_Начала", "source": "AURORA-1", "target": "AURORA-2"})
|
||
raw = self.fake.issue_json("AURORA-1")
|
||
st = self.ex.to_state(raw)
|
||
self.assertEqual(st["key"], "AURORA-1")
|
||
self.assertEqual(st["assignee"], "ivanov")
|
||
self.assertEqual(st["planned_start"], "2026-10-20")
|
||
self.assertEqual(st["epic_key"], "AURORA-1350")
|
||
self.assertEqual(st["labels"], ["AWPCA", "kind-dev"])
|
||
self.assertEqual(st["issue_type"], "Задача")
|
||
self.assertEqual(st["jira_updated"], "2026-10-01T10:00:00.000+0300")
|
||
self.assertIsNone(st["sprint"]) # спринт заполняет do_refresh
|
||
self.assertEqual(self.ex.links_from(raw), [{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "ASUPR_Блокер_Начала"}])
|
||
self.assertEqual(self.ex.links_from(self.fake.issue_json("AURORA-2")),
|
||
[{"source_key": "AURORA-1", "target_key": "AURORA-2", "link_type": "ASUPR_Блокер_Начала"}])
|
||
self.assertIn("customfield_15302", self.ex.state_fields())
|
||
|
||
def test_full_refresh_with_neighbors_and_sprints(self):
|
||
for i in range(1, 4):
|
||
self.fake.add("AURORA-%d" % i, sprint="Спринт 2" if i < 3 else None)
|
||
self.fake.add("OTHER-9", in_base=False)
|
||
self.fake.add("AURORA-7", in_base=False)
|
||
self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "AURORA-1", "target": "OTHER-9"})
|
||
self.fake.links.append({"id": 2, "type": "ASUPR_Связанные", "source": "AURORA-7", "target": "AURORA-2"})
|
||
self.hub.known = ["AURORA-7"]
|
||
out = self.ex.run({"id": 5, "type": "refresh", "params": {"jql": "project = AURORA AND labels = AWPCA"}})
|
||
self.assertEqual(out["status"], "done")
|
||
self.assertEqual(out["result"], {"issues": 3, "neighbors": 1, "pages": 1})
|
||
state = self.hub.states[0]
|
||
self.assertEqual(state["command_id"], 5)
|
||
self.assertEqual(sorted(i["key"] for i in state["issues"]), ["AURORA-1", "AURORA-2", "AURORA-3", "OTHER-9"])
|
||
self.assertEqual(state["neighbor_keys"], ["OTHER-9"])
|
||
self.assertTrue(state["last_page"])
|
||
sprints = {i["key"]: i["sprint"] for i in state["issues"]}
|
||
self.assertEqual(sprints["AURORA-1"], "Спринт 2")
|
||
self.assertIsNone(sprints["AURORA-3"])
|
||
links = {(l["source_key"], l["target_key"]) for l in state["links"]}
|
||
self.assertEqual(links, {("AURORA-1", "OTHER-9"), ("AURORA-7", "AURORA-2")})
|
||
self.assertEqual([r[1] for r in self.fake.requests if r[1] == "/rest/api/2/search"].count("/rest/api/2/search"), 2)
|
||
|
||
def test_refresh_by_keys_pages(self):
|
||
for i in range(1, 121):
|
||
self.fake.add("AURORA-%d" % i)
|
||
keys = ["AURORA-%d" % i for i in range(1, 121)]
|
||
out = self.ex.run({"id": 6, "type": "refresh", "params": {"keys": keys}})
|
||
self.assertEqual(out["result"], {"issues": 120, "neighbors": 0, "pages": 2})
|
||
self.assertEqual(len(self.hub.states), 2)
|
||
self.assertEqual(self.hub.states[0]["page"], 1)
|
||
self.assertFalse(self.hub.states[0]["last_page"])
|
||
self.assertTrue(self.hub.states[1]["last_page"])
|
||
self.assertEqual(len(self.hub.states[0]["issues"]) + len(self.hub.states[1]["issues"]), 120)
|
||
|
||
def test_refresh_empty_keys(self):
|
||
out = self.ex.run({"id": 7, "type": "refresh", "params": {"keys": []}})
|
||
self.assertEqual(out["result"], {"issues": 0, "neighbors": 0, "pages": 0})
|
||
self.assertEqual(self.fake.requests, [])
|
||
self.assertEqual(self.hub.states, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|