jira-agent/tests/test_executor_fields.py
Andrey Epifancev 20747c2b65 Агент v16: зазор между записями и сверка поля оценки доски (#53, I-1, I-2)
I-1: если ensure_fields записал поля через PUT /issue, перед записью оценки — зазор
ensure_fields. I-2: fieldId оценки доски сверяется с field_ids.story_points; расхождение —
failed без записи.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-25 22:28:19 +04:00

378 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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(), [])
# --- story_points через доску (#53) ------------------------------------------
def points(self, key, value):
self.fake.add(key, **({} if value is None else {"customfield_10106": value}))
def test_points_by_board_set(self): # сценарий «Оценка не на экране редактирования»
self.points("AURORA-1704", 1.0)
out = self.run_cmd("ensure_fields", key="AURORA-1704", story_points=2, board_id=4216)
self.assertEqual((out["status"], out["result"], out["jira_calls"]), ("done", {"changed": ["story_points"]}, 2))
self.assertEqual(self.fake.estimation_bodies, [{"value": 2}])
self.assertEqual(self.fake.issues["AURORA-1704"]["customfield_10106"], 2)
self.assertEqual(self.fake.queries[-1], {"boardId": "4216"})
for call in (("GET", "/rest/api/2/issue/AURORA-1704"), ("PUT", "/rest/api/2/issue/AURORA-1704"),
("GET", "/rest/api/2/field")):
self.assertNotIn(call, self.fake.requests)
def test_points_by_board_same(self): # сценарий «Оценка уже стоит»
self.points("AURORA-1704", 2.0)
out = self.run_cmd("ensure_fields", key="AURORA-1704", story_points=2, board_id=4216)
self.assertEqual((out["result"], out["jira_calls"]), ({"changed": []}, 1))
self.assertEqual(self.fake.estimation_bodies, [])
def test_points_by_board_clear(self):
self.points("AURORA-1", 1.0)
out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=None, board_id=4216)
self.assertEqual(out["result"], {"changed": ["story_points"]})
self.assertEqual(self.fake.estimation_bodies, [{"value": None}])
again = self.run_cmd("ensure_fields", key="AURORA-1", story_points=None, board_id=4216)
self.assertEqual((again["result"], again["jira_calls"]), ({"changed": []}, 1))
def test_points_by_board_with_fields(self):
self.points("AURORA-1", None)
out = self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-30", story_points=3, board_id=4216)
self.assertEqual(out["result"], {"changed": ["duedate", "story_points"]})
writes = [r for r in self.fake.requests if r[0] == "PUT"]
self.assertEqual(writes, [("PUT", "/rest/api/2/issue/AURORA-1"), ("PUT", "/rest/agile/1.0/issue/AURORA-1/estimation")])
self.assertEqual(self.fake.issues["AURORA-1"]["duedate"], "2026-10-30")
self.assertEqual(self.fake.issues["AURORA-1"]["customfield_10106"], 3)
def test_points_by_board_jira_error(self):
self.points("AURORA-1", None)
out = self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-30", story_points=3, board_id=9999)
self.assertEqual(out["status"], "failed")
self.assertIn("9999", out["error"])
self.assertEqual(self.fake.issues["AURORA-1"]["duedate"], "2026-10-30")
def test_points_by_board_no_field_lookup(self):
self.points("AURORA-1", None)
self.fake.fields = []
self.ex.s["field_ids"] = {k: v for k, v in SETTINGS["field_ids"].items() if k != "story_points"}
out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=3, board_id=4216)
self.assertEqual((out["status"], out["result"]), ("done", {"changed": ["story_points"]}))
def test_points_by_board_bad_value(self):
self.points("AURORA-1", None)
out = self.run_cmd("ensure_fields", key="AURORA-1", story_points="3", board_id=4216)
self.assertEqual((out["status"], out["error"]), ("failed", "story_points — число ≥ 0 или null"))
self.assertEqual(self.fake.requests, [])
def test_points_by_board_gap_after_fields(self): # I-1: зазор ensure_fields между двумя записями
gaps = []
self.ex.sleep = gaps.append
self.ex.s["pacing"] = {"ensure_fields": 120}
self.points("AURORA-1", None)
self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-30", story_points=3, board_id=4216)
self.assertEqual(gaps, [120])
self.run_cmd("ensure_fields", key="AURORA-1", duedate="2026-10-30", story_points=5, board_id=4216)
self.assertEqual(gaps, [120]) # PUT /issue не было — паузы нет
def test_points_by_board_other_field(self): # I-2: оценка доски — не поле story points
self.points("AURORA-1", 1.0)
self.ex.s["field_ids"] = dict(SETTINGS["field_ids"], story_points="customfield_20000")
out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=2, board_id=4216)
self.assertEqual((out["status"], out["error"]),
("failed", "у доски 4216 оценка — поле customfield_10106, а не story points (customfield_20000)"))
self.assertEqual(self.fake.estimation_bodies, [])
self.ex.s["field_ids"] = dict(SETTINGS["field_ids"], story_points="customfield_10106")
out = self.run_cmd("ensure_fields", key="AURORA-1", story_points=2, board_id=4216)
self.assertEqual(out["result"], {"changed": ["story_points"]})
if __name__ == "__main__":
unittest.main()