import unittest from datetime import timedelta from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_jira import FakeJira class ExecutorFieldsTest(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.ex.sleep = no_sleep def tearDown(self): self.fake.stop() def run_cmd(self, ctype, **params): return self.ex.run({"id": 1, "type": ctype, "params": params}) def test_ensure_issue_creates_then_finds(self): out = self.run_cmd("ensure_issue", temp_key="new:1", issue_type="task", summary="Новая задача", description="h3. Контекст", priority="Высокий", duedate="2026-10-23", planned_start="2026-10-20", labels=["AWPCA", "kind-dev"], epic_key="AURORA-1350", assignee="ivanov") self.assertEqual(out["status"], "done") key = out["result"]["key"] self.assertTrue(out["result"]["created"]) f = self.fake.issues[key] self.assertEqual(f["priority"], "Высокий") self.assertEqual(f["customfield_10801"], "2026-10-20") self.assertEqual(f["customfield_15302"], "AURORA-1350") self.assertEqual(f["labels"], ["AWPCA", "kind-dev"]) self.assertEqual(out["jira_calls"], 2) again = self.run_cmd("ensure_issue", temp_key="new:1", issue_type="task", summary="Новая задача") self.assertEqual(again["status"], "done") self.assertEqual(again["result"], {"key": key, "existed": True}) self.assertEqual(len(self.fake.issues), 1) self.assertEqual(again["jira_calls"], 1) def test_ensure_issue_epic_sets_epic_name(self): out = self.run_cmd("ensure_issue", temp_key="new:2", issue_type="epic", summary="Приёмка") f = self.fake.issues[out["result"]["key"]] self.assertEqual(f["issuetype"], "Фича") self.assertEqual(f["customfield_15304"], "Приёмка") def test_ensure_issue_priority_mapped(self): out = self.run_cmd("ensure_issue", temp_key="new:3", issue_type="task", summary="x", priority="Наивысший") self.assertEqual(self.fake.issues[out["result"]["key"]]["priority"], "Критичный") def test_ensure_issue_bug_by_name(self): self.assertEqual(self.ex.issue_type_ref("bug"), {"name": "Ошибка"}) # id нет — по имени self.assertEqual(self.ex.issue_type_ref("task"), {"id": "10100"}) # id есть — по id out = self.run_cmd("ensure_issue", temp_key="new:1", issue_type="bug", summary="ОСВ 60 не распознана", labels=["AWPCA"]) self.assertEqual(out["status"], "done") self.assertTrue(out["result"]["created"]) self.assertEqual(self.fake.issues[out["result"]["key"]]["issuetype"], "Ошибка") def test_ensure_issue_type_not_configured_no_jira_calls(self): self.ex.s = dict(SETTINGS, issue_types={"task": {"name": "Задача", "id": "10100"}}) out = self.run_cmd("ensure_issue", temp_key="new:1", issue_type="bug", summary="x") self.assertEqual(out["status"], "failed") self.assertEqual(out["error"], "нет настройки issue_types.bug") self.assertNotIn("retryable", out) self.assertEqual(self.fake.requests, []) # ни поиска дубля, ни создания def test_issue_type_entry_without_name_and_id(self): for entry in ({}, {"name": None, "id": None}, {"name": "", "id": ""}, None, "Ошибка"): self.ex.s = dict(SETTINGS, issue_types={"bug": entry}) with self.assertRaises(agent.ConfigGap) as cm: self.ex.issue_type_ref("bug") self.assertEqual(str(cm.exception), "нет настройки issue_types.bug") self.ex.s = dict(SETTINGS) del self.ex.s["issue_types"] with self.assertRaises(agent.ConfigGap): self.ex.issue_type_ref("task") def test_ensure_fields_writes_only_diff(self): self.fake.add("AURORA-1", duedate="2026-10-20", labels=["AWPCA", "fact-open"], priority="Средний") out = self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-23", priority="Средний", labels_add=["kind-dev", "AWPCA"], labels_remove=["fact-open"], assignee="ivanov", planned_end="2026-10-23", epic_key="AURORA-1350") self.assertEqual(out["status"], "done") self.assertEqual(out["result"]["changed"], ["duedate", "planned_end", "epic_key", "assignee", "labels"]) f = self.fake.issues["AURORA-1"] self.assertEqual(f["labels"], ["AWPCA", "kind-dev"]) self.assertEqual(f["assignee"], "ivanov") self.assertEqual(f["customfield_10802"], "2026-10-23") self.assertEqual(len(self.fake.writes()), 1) again = self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-23", labels_add=["kind-dev"], assignee="ivanov") self.assertEqual(again["result"], {"changed": []}) self.assertEqual(len(self.fake.writes()), 1) def test_fields_description_exact_compare(self): text = "h3. Контекст\n* пункт *жирный*\n\nh3. Что сделать" self.fake.add("AURORA-1", description=text) out = self.run_cmd("ensure_fields", key="AURORA-1", description=text) self.assertEqual(out["result"]["changed"], []) out = self.run_cmd("ensure_fields", key="AURORA-1", description=text + "\n") self.assertEqual(out["result"]["changed"], ["description"]) def test_fields_date_with_time_from_jira(self): self.fake.add("AURORA-1", customfield_10802="2026-10-23T00:00:00.000+0300") out = self.run_cmd("ensure_fields", key="AURORA-1", planned_end="2026-10-23") self.assertEqual(out["result"]["changed"], []) def test_fields_clear_assignee_and_epic(self): self.fake.add("AURORA-1", assignee="ivanov", customfield_15302="AURORA-1350") out = self.run_cmd("ensure_fields", key="AURORA-1", assignee=None, epic_key=None) self.assertEqual(out["result"]["changed"], ["epic_key", "assignee"]) self.assertIsNone(self.fake.issues["AURORA-1"]["assignee"]) self.assertIsNone(self.fake.issues["AURORA-1"]["customfield_15302"]) def test_fields_missing_issue_is_failed(self): out = self.run_cmd("ensure_fields", key="AURORA-404", duedate="2026-10-23") self.assertEqual(out["status"], "failed") self.assertIn("HTTP 404", out["error"]) def test_ensure_comment_upsert(self): self.fake.add("AURORA-1") out = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="Сделано: список") self.assertEqual(out["status"], "done") self.assertTrue(out["result"]["created"]) self.assertEqual(self.fake.comments["AURORA-1"][0]["body"], "*Актуализация 10.10.2026.* Сделано: список") same = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="Сделано: список") self.assertEqual(same["result"]["changed"], False) upd = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="Сделано: всё") self.assertEqual(upd["result"]["changed"], True) self.assertEqual(len(self.fake.comments["AURORA-1"]), 1) self.assertEqual(self.fake.comments["AURORA-1"][0]["body"], "*Актуализация 10.10.2026.* Сделано: всё") def test_fields_description_crlf_equals_lf(self): self.fake.add("AURORA-1", description="а\r\nб") # так Jira хранит текст, правленный в UI out = self.run_cmd("ensure_fields", key="AURORA-1", description="а\nб") self.assertEqual(out["result"]["changed"], []) out = self.run_cmd("ensure_fields", key="AURORA-1", description="а\nв") self.assertEqual(out["result"]["changed"], ["description"]) self.assertEqual(self.fake.issues["AURORA-1"]["description"], "а\nв") # отправлен текст команды как есть def test_fields_description_none_equals_empty(self): self.fake.add("AURORA-1", description=None) out = self.run_cmd("ensure_fields", key="AURORA-1", description="") self.assertEqual(out["result"]["changed"], []) def test_comment_crlf_equals_lf(self): self.fake.add("AURORA-1") self.fake.comments["AURORA-1"] = [{"id": "10", "body": "*Актуализация 10.10.2026.* строка1\r\nстрока2"}] out = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="строка1\nстрока2") self.assertEqual(out["result"], {"comment_id": "10", "changed": False}) def test_unknown_type(self): self.assertEqual(self.run_cmd("ensure_magic", key="x")["status"], "failed") # --- ensure_issue: поиск дубля ----------------------------------------------- def test_ensure_issue_finds_on_later_page(self): self.fake.add("AURORA-500", reporter="me", summary="Нужная задача") for i in range(250): self.fake.add("AURORA-%d" % (600 + i), reporter="me", summary="Другая %d" % i) out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="Нужная задача") self.assertEqual(out["result"], {"key": "AURORA-500", "existed": True}) searches = [q for q in self.fake.queries if "jql" in q] self.assertEqual(len(searches), 2) # вторая страница self.assertIn("reporter = currentUser()", searches[0]["jql"]) self.assertIn("created >= -7d", searches[0]["jql"]) self.assertIn("ORDER BY created DESC", searches[0]["jql"]) self.assertEqual(searches[0]["fields"], "summary") self.assertEqual(self.fake.writes(), []) def test_ensure_issue_substring_is_not_duplicate_and_strip(self): self.fake.add("AURORA-500", reporter="me", summary="Нужная задача (старая)") self.fake.add("AURORA-501", reporter="me", summary=" Точная тема \n") out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="Нужная задача") self.assertTrue(out["result"]["created"]) out = self.run_cmd("ensure_issue", temp_key="n:2", issue_type="task", summary="Точная тема") self.assertEqual(out["result"], {"key": "AURORA-501", "existed": True}) def test_ensure_issue_old_or_foreign_is_not_duplicate(self): self.fake.add("AURORA-500", reporter="me", summary="Нужная задача", created=self.fake.ts(self.fake.now - timedelta(days=8))) # за окном created >= -7d self.fake.add("OTHER-1", reporter="me", summary="Нужная задача") # чужой проект out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="Нужная задача") self.assertTrue(out["result"]["created"]) self.assertNotIn(out["result"]["key"], ("AURORA-500", "OTHER-1")) def test_ensure_issue_within_window_is_duplicate(self): self.fake.add("AURORA-500", reporter="me", summary="Нужная задача", created=self.fake.ts(self.fake.now - timedelta(days=6))) out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="Нужная задача") self.assertEqual(out["result"], {"key": "AURORA-500", "existed": True}) def test_http_error_becomes_failed(self): self.fake.fail_next.append((400, {})) out = self.run_cmd("ensure_issue", temp_key="n:1", issue_type="task", summary="x") self.assertEqual(out["status"], "failed") self.assertIn("HTTP 400", out["error"]) self.fake.add("AURORA-1") self.fake.fail_next.append((400, {})) out = self.run_cmd("ensure_comment", key="AURORA-1", marker="М", body="текст") self.assertEqual(out["status"], "failed") # --- настройки (R-11) -------------------------------------------------------- def with_settings(self, **over): s = dict(SETTINGS) s.update(over) self.ex = agent.Executor(self.jira, s) self.ex.sleep = no_sleep def test_epic_key_without_field_id_fails(self): self.fake.add("AURORA-1") self.with_settings(epic_link={"mode": "field"}) for ctype, params in (("ensure_issue", dict(temp_key="n", issue_type="task", summary="Э", epic_key="AURORA-9")), ("ensure_fields", dict(key="AURORA-1", epic_key="AURORA-9"))): out = self.run_cmd(ctype, **params) self.assertEqual(out["status"], "failed") self.assertIn("epic_link.field_id", out["error"]) self.assertEqual(self.fake.writes(), []) def test_epic_key_with_link_mode_fails_same_in_both(self): self.fake.add("AURORA-1") self.with_settings(epic_link={"mode": "link", "field_id": "customfield_15302"}) a = self.run_cmd("ensure_issue", temp_key="n", issue_type="task", summary="Э2", epic_key="AURORA-9") b = self.run_cmd("ensure_fields", key="AURORA-1", epic_key="AURORA-9") self.assertEqual((a["status"], b["status"]), ("failed", "failed")) self.assertIn("epic_link.mode", a["error"]) self.assertIn("epic_link.mode", b["error"]) def test_planned_without_field_id_fails(self): self.fake.add("AURORA-1") ids = dict(SETTINGS["field_ids"]) del ids["planned_start"] self.with_settings(field_ids=ids) a = self.run_cmd("ensure_issue", temp_key="n", issue_type="task", summary="П", planned_start="2026-10-20") b = self.run_cmd("ensure_fields", key="AURORA-1", planned_start="2026-10-20") for out in (a, b): self.assertEqual(out["status"], "failed") self.assertIn("field_ids.planned_start", out["error"]) self.assertEqual(self.fake.writes(), []) # --- комментарии ------------------------------------------------------------- def test_comment_marker_only_at_start(self): self.fake.add("AURORA-1") self.fake.comments["AURORA-1"] = [{"id": "1", "body": "Обсуждали, см. Актуализация 10.10.2026 в письме"}] out = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="новое") self.assertTrue(out["result"]["created"]) # маркер в середине чужого комментария — не наш again = self.run_cmd("ensure_comment", key="AURORA-1", marker="Актуализация 10.10.2026", body="новое") self.assertEqual(again["result"]["changed"], False) def test_comment_found_on_later_page(self): self.fake.add("AURORA-1") self.fake.comments["AURORA-1"] = [{"id": str(i), "body": "чужой %d" % i} for i in range(1, 121)] self.fake.comments["AURORA-1"].append({"id": "999", "body": "*Маркер.* старое"}) out = self.run_cmd("ensure_comment", key="AURORA-1", marker="Маркер", body="новое") self.assertEqual(out["result"], {"comment_id": "999", "changed": True}) def test_priority_compare_by_id_as_string(self): self.fake.add("AURORA-1", priority="3") self.with_settings(priority_value_key="id", priority_map={}) out = self.run_cmd("ensure_fields", key="AURORA-1", priority="3") self.assertEqual(out["result"], {"changed": []}) # --- story_points (Task 4, #49) ---------------------------------------------- def test_story_points_set_clear_idempotent(self): self.fake.add("AURORA-1") out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=3) self.assertEqual(out["result"], {"changed": ["story_points"]}) self.assertEqual(self.fake.issues["AURORA-1"]["customfield_10106"], 3) for v in (3, 3.0): again = self.run_cmd("ensure_fields", key="AURORA-1", story_points=v) self.assertEqual(again["result"], {"changed": []}) self.assertEqual(len(self.fake.writes()), 1) self.run_cmd("ensure_fields", key="AURORA-1", story_points=None) self.assertIsNone(self.fake.issues["AURORA-1"]["customfield_10106"]) def test_story_points_no_field(self): # сценарий «Нет поля оценки» self.fake.add("AURORA-1") self.fake.fields = [] out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=3) self.assertEqual((out["status"], out["error"]), ("failed", "поле оценки не найдено в Jira: задайте field_ids.story_points")) self.assertEqual(self.fake.writes(), []) def test_story_points_override(self): self.fake.add("AURORA-1") self.ex.s["field_ids"] = dict(SETTINGS["field_ids"], story_points="customfield_20000") self.run_cmd("ensure_fields", key="AURORA-1", story_points=5) self.assertEqual(self.fake.issues["AURORA-1"]["customfield_20000"], 5) self.assertNotIn(("GET", "/rest/api/2/field"), self.fake.requests) def test_story_points_bad_value(self): self.fake.add("AURORA-1") for v in ("3", -1, True): out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=v) self.assertEqual((out["status"], out["error"]), ("failed", "story_points — число ≥ 0 или null"), v) self.assertEqual(self.fake.writes(), []) if __name__ == "__main__": unittest.main()