523 lines
31 KiB
Python
523 lines
31 KiB
Python
"""Фейковый Jira Server для тестов агента: состояние в памяти, http.server в потоке,
|
||
только маршруты, которые использует агент. Семантика связей — как у настоящего Jira Server."""
|
||
import json
|
||
import re
|
||
import socket
|
||
import threading
|
||
from datetime import datetime, timedelta, timezone
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from urllib.parse import parse_qs, quote, unquote, urlparse
|
||
|
||
MSK = timezone(timedelta(hours=3))
|
||
|
||
WORKFLOW = { # статус → [(имя перехода, целевой статус)]
|
||
"Ожидание разработки": [("Взять в разработку", "Разработка"), ("Ожидает анализа", "Ожидает анализа"), ("Снято", "Снято")],
|
||
"Разработка": [("Малое исправление", "Готово к тестированию"), ("Готово", "Готово"), ("Снято", "Снято")],
|
||
"Ожидает анализа": [("Анализ", "Анализ"), ("Снято", "Снято")],
|
||
"Анализ": [("Готово", "Готово")],
|
||
"Готово к тестированию": [("Тестирование", "Тестирование")],
|
||
"Готово": [],
|
||
"Снято": [],
|
||
}
|
||
LINK_TYPES = [{"name": "ASUPR_Блокер_Начала", "inward": "Начало заблокировано", "outward": "Блокирует начало"},
|
||
{"name": "ASUPR_Связанные", "inward": "Связанные", "outward": "Связанные"}]
|
||
SPRINT_FIELD = "customfield_10104"
|
||
RANK_FIELD = "customfield_10105"
|
||
POINTS_FIELD = "customfield_10106"
|
||
CUSTOM = ("customfield_10801", "customfield_10802", "customfield_15302", "customfield_15304", RANK_FIELD, POINTS_FIELD)
|
||
|
||
|
||
def parse_multipart(raw, content_type):
|
||
"""Строгий разбор multipart/form-data, как commons-fileupload Jira Server с кодировкой заголовков UTF-8:
|
||
[{"headers": {имя в нижнем регистре: значение}, "data": bytes}]. Нарушение формата — AssertionError (фейк ответит 400)."""
|
||
m = re.search(r"boundary=([^;\s]+)", content_type)
|
||
assert m, "нет boundary в Content-Type"
|
||
delim = b"--" + m.group(1).encode("ascii")
|
||
head, tail = delim + b"\r\n", b"\r\n" + delim + b"--\r\n"
|
||
assert raw.startswith(head), "тело не начинается с границы"
|
||
assert raw.endswith(tail), "нет закрывающей границы"
|
||
parts = []
|
||
for chunk in raw[len(head):len(raw) - len(tail)].split(b"\r\n" + delim + b"\r\n"):
|
||
hdr, sep, data = chunk.partition(b"\r\n\r\n")
|
||
assert sep, "у части нет пустой строки после заголовков"
|
||
headers = {}
|
||
for line in hdr.decode("utf-8").split("\r\n"):
|
||
name, _, value = line.partition(":")
|
||
headers[name.strip().lower()] = value.strip()
|
||
parts.append({"headers": headers, "data": data})
|
||
return parts
|
||
|
||
|
||
def disposition_params(value):
|
||
"""Параметры Content-Disposition части: name; filename — как в кавычках, без percent-декодирования (так читает
|
||
Jira Server); filename* — после UTF-8'' с percent-декодированием."""
|
||
out = {}
|
||
m = re.search(r'(?:^|;)\s*name="([^"]*)"', value)
|
||
if m:
|
||
out["name"] = m.group(1)
|
||
m = re.search(r'(?:^|;)\s*filename="([^"]*)"', value)
|
||
if m:
|
||
out["filename"] = m.group(1)
|
||
m = re.search(r"(?:^|;)\s*filename\*=UTF-8''([^;\s]+)", value)
|
||
if m:
|
||
out["filename*"] = unquote(m.group(1), encoding="utf-8", errors="strict")
|
||
return out
|
||
|
||
|
||
class FakeJira:
|
||
def __init__(self, project="AURORA"):
|
||
self.project = project
|
||
self.now = datetime.now(MSK) # часы фейка: окно created >= -Nd считается от них
|
||
self.issues = {}
|
||
self.comments = {}
|
||
self.links = []
|
||
self.sprints = [{"id": 1, "name": "Спринт 1", "state": "active"}, {"id": 2, "name": "Спринт 2", "state": "future"}]
|
||
self.boards = {4216: "scrum"} # доска → тип; GET /rest/agile/1.0/board/{id}
|
||
self.board_sprints = {} # доска → спринты; доски нет — общий self.sprints
|
||
self.sprint_field_format = "object" # элемент поля спринтов: словарь или строка greenhopper
|
||
self.sprint_members = {} # id спринта → ключи задач (иначе — по имени спринта задачи)
|
||
self.seq, self.link_seq, self.comment_seq = 1000, 100, 10
|
||
self.id_seq = 10000
|
||
self.fail_next = [] # [(код, {заголовки})] — следующие ответы с ошибкой
|
||
self.requests = [] # [(метод, путь)]
|
||
self.queries = [] # параметры запросов (q) — для проверки JQL
|
||
self.drop_next_post_issue = False # выполнить создание задачи и оборвать соединение без ответа
|
||
self.comment_denied = {} # ключ → код ответа на GET /issue/{key}/comment (403 — нет прав, 404 — нет задачи)
|
||
self.attachments = {} # ключ → [{id, filename, size, mimeType, data}]
|
||
self.attachment_seq = 20000
|
||
self.uploads = [] # разобранные POST /attachments: {key, token, content_type, headers, disposition}
|
||
self.drop_next_post_attachment = False # сохранить вложение и оборвать соединение без ответа
|
||
self.attachment_limit = None # байт: больше — 413, как предел вложений Jira
|
||
self.bad_attachment_response = False # ответить на загрузку пустым списком
|
||
self.hide_size = set() # id вложений без size в JSON (Jira не отдала размер)
|
||
self.hide_attachment_field = False # поле attachment скрыто настройками Jira
|
||
self.content_gets = 0 # GET /secure/attachment/<id>/<имя> — запросы содержимого
|
||
self.redirect_content = None # адрес для 302 на запрос содержимого (абсолютный или путь этого сервера)
|
||
self.short_content = set() # id вложений, содержимое которых отдаётся без последнего байта
|
||
self.lying_length = set() # id вложений: Content-Length больше тела, соединение рвётся после тела
|
||
self.redirects = {} # путь → адрес Location: 302 вместо ответа (любой метод)
|
||
self.last_headers = {} # заголовки текущего запроса (имена в нижнем регистре)
|
||
self.search_comment_cap = None # сколько комментариев задачи отдаёт search (None — все), как обрезка Jira
|
||
self.changelogs = {} # ключ → записи журнала Jira [{id, author, created, items}]
|
||
self.history_seq = 50000
|
||
self.changelog_cap = None # сколько записей журнала отдаёт search и GET /issue (None — все); total — полное
|
||
self.issue_changelog_cap = None # то же только для GET /issue/{key}?expand=changelog (None — как changelog_cap)
|
||
self.jql_results = {} # JQL → ключи (проверяется в search_keys первым)
|
||
self.fields = [{"id": "customfield_10201", "name": "Плановое окончание", "custom": True},
|
||
{"id": SPRINT_FIELD, "name": "Sprint", "custom": True,
|
||
"schema": {"type": "array", "items": "string",
|
||
"custom": "com.pyxis.greenhopper.jira:gh-sprint"}},
|
||
{"id": RANK_FIELD, "name": "Rank", "custom": True,
|
||
"schema": {"type": "any", "custom": "com.pyxis.greenhopper.jira:gh-lexo-rank"}},
|
||
{"id": POINTS_FIELD, "name": "Story Points", "custom": True,
|
||
"schema": {"type": "number", "custom": "com.atlassian.jira.plugin.system.customfieldtypes:float"}},
|
||
{"id": "duedate", "name": "Срок исполнения", "custom": False}] # GET /rest/api/2/field
|
||
self.lock = threading.Lock()
|
||
fake = self
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def log_message(self, *a):
|
||
pass
|
||
|
||
def do_GET(self):
|
||
fake.handle(self, "GET")
|
||
|
||
def do_POST(self):
|
||
fake.handle(self, "POST")
|
||
|
||
def do_PUT(self):
|
||
fake.handle(self, "PUT")
|
||
|
||
def do_DELETE(self):
|
||
fake.handle(self, "DELETE")
|
||
|
||
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||
|
||
@property
|
||
def base_url(self):
|
||
return "http://127.0.0.1:%d" % self.server.server_port
|
||
|
||
def stop(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
|
||
@staticmethod
|
||
def ts(dt):
|
||
return dt.astimezone(MSK).strftime("%Y-%m-%dT%H:%M:%S.000+0300")
|
||
|
||
def add(self, key, in_base=True, **fields):
|
||
base = {"summary": "Тема " + key, "description": None, "status": "Ожидание разработки", "priority": "Средний",
|
||
"assignee": None, "reporter": "someone", "duedate": None, "labels": ["AWPCA"] if in_base else [],
|
||
"issuetype": "Задача", "created": self.ts(self.now - timedelta(hours=1)), "updated": "2026-10-01T10:00:00.000+0300",
|
||
"sprint": None, "sprint_ids": None}
|
||
self.id_seq += 1
|
||
base["id"] = str(self.id_seq)
|
||
base.update({c: None for c in CUSTOM})
|
||
base.update(fields)
|
||
self.issues[key] = base
|
||
return base
|
||
|
||
def add_history(self, key, created, items, hid=None, name="petrov", display="Петров Пётр"):
|
||
"""Запись журнала задачи; возвращает её id."""
|
||
self.history_seq += 1
|
||
hid = hid or str(self.history_seq)
|
||
self.changelogs.setdefault(key, []).append({"id": hid, "author": {"name": name, "displayName": display},
|
||
"created": created, "items": items})
|
||
return hid
|
||
|
||
def changelog_json(self, key, cap):
|
||
lst = self.changelogs.get(key, [])
|
||
shown = lst if cap is None else lst[:cap]
|
||
return {"startAt": 0, "maxResults": len(shown) if cap is None else cap, "total": len(lst), "histories": shown}
|
||
|
||
def add_comment(self, key, body, name="ivanov", display="Иванов Иван", created="2026-10-14T12:00:00.000+0300",
|
||
updated=None):
|
||
author = {k: v for k, v in (("name", name), ("displayName", display)) if v is not None} if name or display else None
|
||
self.comment_seq += 1
|
||
c = {"id": str(self.comment_seq), "body": body, "author": author, "created": created, "updated": updated or created}
|
||
self.comments.setdefault(key, []).append(c)
|
||
return c
|
||
|
||
def add_attachment(self, key, filename, data, mime="application/octet-stream", name="potapova", display="Потапова Анна",
|
||
created="2026-10-20T09:15:00.000+0300"):
|
||
self.attachment_seq += 1
|
||
author = {"name": name, "displayName": display} if display else ({"name": name} if name else None)
|
||
a = {"id": str(self.attachment_seq), "filename": filename, "size": len(data), "mimeType": mime, "data": data,
|
||
"author": author, "created": created}
|
||
self.attachments.setdefault(key, []).append(a)
|
||
return a
|
||
|
||
def attachment_json(self, a):
|
||
out = {"self": "%s/rest/api/2/attachment/%s" % (self.base_url, a["id"]), "id": a["id"],
|
||
"filename": a["filename"], "size": a["size"], "mimeType": a["mimeType"], "author": a.get("author"),
|
||
"created": a.get("created"),
|
||
"content": "%s/secure/attachment/%s/%s" % (self.base_url, a["id"], quote(a["filename"]))}
|
||
if a["id"] in self.hide_size:
|
||
del out["size"]
|
||
return out
|
||
|
||
def writes(self):
|
||
return [r for r in self.requests if r[0] in ("POST", "PUT", "DELETE")]
|
||
|
||
# --- представление ---------------------------------------------------------
|
||
def _user(self, name):
|
||
return {"name": name, "displayName": name} if name else None
|
||
|
||
def issue_json(self, key, fields=None):
|
||
"""fields — список запрошенных полей (как параметр fields у search); None — все поля, как GET /issue/{key}."""
|
||
f = self.issues[key]
|
||
out = {"summary": f["summary"], "description": f["description"], "status": {"name": f["status"]},
|
||
"priority": {"name": f["priority"], "id": int(f["priority"]) if str(f["priority"]).isdigit() else f["priority"]}, "assignee": self._user(f["assignee"]), "reporter": self._user(f["reporter"]),
|
||
"duedate": f["duedate"], "labels": list(f["labels"]), "issuetype": {"name": f["issuetype"]},
|
||
"created": f["created"], "updated": f["updated"], "issuelinks": self.links_of(key)}
|
||
out.update({c: f.get(c) for c in CUSTOM})
|
||
out[SPRINT_FIELD] = self.sprint_value(f)
|
||
if fields is None or "comment" in fields:
|
||
lst = self.comments.get(key, [])
|
||
cap = self.search_comment_cap if fields is not None else None
|
||
shown = lst if cap is None else lst[:cap]
|
||
out["comment"] = {"startAt": 0, "maxResults": len(shown) if cap is None else cap, "total": len(lst),
|
||
"comments": shown}
|
||
if (fields is None or "attachment" in fields) and not self.hide_attachment_field:
|
||
out["attachment"] = [self.attachment_json(a) for a in self.attachments.get(key, [])]
|
||
return {"key": key, "id": f["id"], "fields": out}
|
||
|
||
def sprint_value(self, f):
|
||
"""Значение поля спринтов задачи: sprint_raw как есть; иначе спринты по sprint_ids, иначе по имени sprint."""
|
||
if "sprint_raw" in f:
|
||
return f["sprint_raw"]
|
||
allsp = self.all_sprints()
|
||
if f.get("sprint_ids") is not None:
|
||
sps = [s for i in f["sprint_ids"] for s in allsp if s["id"] == i]
|
||
else:
|
||
sps = [s for s in allsp if s["name"] == f.get("sprint")][:1] if f.get("sprint") else []
|
||
if not sps:
|
||
return None
|
||
if self.sprint_field_format == "string":
|
||
return ["com.atlassian.greenhopper.service.sprint.Sprint@1f2e3d[id=%d,rapidViewId=4216,state=%s,name=%s,"
|
||
"sequence=%d]" % (s["id"], s["state"].upper(), s["name"], s["id"]) for s in sps]
|
||
return [{"id": s["id"], "name": s["name"], "state": s["state"]} for s in sps]
|
||
|
||
def links_of(self, key):
|
||
out = []
|
||
for l in self.links:
|
||
if l["source"] == key:
|
||
out.append({"id": str(l["id"]), "type": {"name": l["type"]}, "outwardIssue": {"key": l["target"]}})
|
||
elif l["target"] == key:
|
||
out.append({"id": str(l["id"]), "type": {"name": l["type"]}, "inwardIssue": {"key": l["source"]}})
|
||
return out
|
||
|
||
def search_keys(self, jql):
|
||
if jql in self.jql_results:
|
||
return [k for k in self.jql_results[jql] if k in self.issues]
|
||
m = re.search(r"key in \(([^)]*)\)", jql or "")
|
||
if m:
|
||
keys = [k.strip() for k in m.group(1).split(",")]
|
||
return [k for k in keys if k in self.issues]
|
||
if "reporter = currentUser()" in (jql or ""):
|
||
keys = [k for k, f in self.issues.items() if f["reporter"] == "me"]
|
||
m = re.search(r"project = (\w+)", jql)
|
||
if m: # как Jira: только задачи проекта
|
||
keys = [k for k in keys if k.startswith(m.group(1) + "-")]
|
||
m = re.search(r"created >= -(\d+)d", jql)
|
||
if m: # как Jira: только созданные за последние N дней
|
||
since = self.now - timedelta(days=int(m.group(1)))
|
||
keys = [k for k in keys
|
||
if datetime.strptime(self.issues[k]["created"], "%Y-%m-%dT%H:%M:%S.%f%z") >= since]
|
||
return keys[::-1] if "ORDER BY created DESC" in jql else keys
|
||
return [k for k, f in self.issues.items() if "AWPCA" in f["labels"]]
|
||
|
||
# --- маршрутизация ----------------------------------------------------------
|
||
def handle(self, h, method):
|
||
url = urlparse(h.path)
|
||
q = {k: v[0] for k, v in parse_qs(url.query).items()}
|
||
length = int(h.headers.get("Content-Length") or 0)
|
||
raw = h.rfile.read(length) if length else b""
|
||
ctype = h.headers.get("Content-Type") or ""
|
||
if ctype.startswith("multipart/form-data"):
|
||
body = {"_raw": raw, "_content_type": ctype} # разбирает маршрут: ошибка формата — 400
|
||
else:
|
||
body = json.loads(raw.decode("utf-8")) if raw else {}
|
||
with self.lock:
|
||
self.last_headers = {k.lower(): v for k, v in h.headers.items()}
|
||
self.requests.append((method, url.path))
|
||
self.queries.append(q)
|
||
if self.fail_next:
|
||
code, headers = self.fail_next.pop(0)
|
||
return self._send(h, code, {"errorMessages": ["имитация ошибки"]}, headers)
|
||
try:
|
||
code, payload = self.route(method, url.path, q, body)
|
||
except KeyError as e:
|
||
code, payload = 404, {"errorMessages": ["Issue Does Not Exist: %s" % e]}
|
||
except Exception as e: # некорректный запрос к фейку
|
||
code, payload = 400, {"errorMessages": ["фейк: %s: %s" % (type(e).__name__, e)]}
|
||
if url.path in self.redirects:
|
||
h.send_response(302)
|
||
h.send_header("Location", self.redirects[url.path])
|
||
h.send_header("Content-Length", "0")
|
||
h.end_headers()
|
||
return
|
||
if method == "GET" and (url.path.startswith("/secure/attachment/") or url.path.startswith("/secure/moved/")):
|
||
return self._content(h, url.path)
|
||
if self.drop_next_post_issue and method == "POST" and url.path == "/rest/api/2/issue":
|
||
self.drop_next_post_issue = False
|
||
h.close_connection = True
|
||
h.connection.shutdown(socket.SHUT_RDWR) # задача создана, ответа клиент не получит
|
||
return
|
||
if self.drop_next_post_attachment and method == "POST" and url.path.endswith("/attachments"):
|
||
self.drop_next_post_attachment = False
|
||
h.close_connection = True
|
||
h.connection.shutdown(socket.SHUT_RDWR) # вложение сохранено, ответа клиент не получит
|
||
return
|
||
self._send(h, code, payload)
|
||
|
||
def _content(self, h, path):
|
||
"""Содержимое вложения, как /secure/attachment/<id>/<имя> Jira: байты, не JSON. /secure/moved/… — цель 302."""
|
||
with self.lock:
|
||
m = re.match(r"^/secure/(attachment|moved)/(\d+)/(.+)$", path)
|
||
found = None
|
||
if m:
|
||
for lst in self.attachments.values():
|
||
for a in lst:
|
||
if a["id"] == m.group(2) and a["filename"] == unquote(m.group(3)):
|
||
found = a
|
||
redirect = self.redirect_content if m and m.group(1) == "attachment" else None
|
||
if found is not None and redirect is None:
|
||
self.content_gets += 1
|
||
if redirect is not None:
|
||
h.send_response(302)
|
||
h.send_header("Location", redirect)
|
||
h.send_header("Content-Length", "0")
|
||
h.end_headers()
|
||
return
|
||
if found is None:
|
||
return self._send(h, 404, {"errorMessages": ["нет вложения"]})
|
||
data = found["data"][:-1] if found["id"] in self.short_content else found["data"]
|
||
h.send_response(200)
|
||
h.send_header("Content-Type", found["mimeType"])
|
||
h.send_header("Content-Length", str(len(data) + (60 if found["id"] in self.lying_length else 0)))
|
||
h.end_headers()
|
||
h.wfile.write(data)
|
||
if found["id"] in self.lying_length:
|
||
h.wfile.flush()
|
||
h.close_connection = True
|
||
h.connection.shutdown(socket.SHUT_RDWR) # обрыв посреди тела
|
||
|
||
def _send(self, h, code, payload, headers=None):
|
||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else b""
|
||
h.send_response(code)
|
||
for k, v in (headers or {}).items():
|
||
h.send_header(k, v)
|
||
h.send_header("Content-Type", "application/json")
|
||
h.send_header("Content-Length", str(len(data)))
|
||
h.end_headers()
|
||
h.wfile.write(data)
|
||
|
||
def route(self, method, path, q, body):
|
||
p = path
|
||
if p == "/rest/api/2/myself":
|
||
return 200, {"name": "me", "displayName": "Me"}
|
||
if p == "/rest/api/2/search":
|
||
m = re.search(r"key in \(([^)]*)\)", q.get("jql") or "")
|
||
if m and any(k.strip() not in self.issues for k in m.group(1).split(",")): # как Jira Server
|
||
return 400, {"errorMessages": ["Задача не существует или у вас нет прав на её просмотр."]}
|
||
keys = self.search_keys(q.get("jql"))
|
||
start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 50))
|
||
page = keys[start:start + maxr]
|
||
fields = (q.get("fields") or "").split(",") if q.get("fields") else []
|
||
issues = [self.issue_json(k, fields) for k in page]
|
||
if q.get("expand") == "changelog":
|
||
for it in issues:
|
||
it["changelog"] = self.changelog_json(it["key"], self.changelog_cap)
|
||
return 200, {"startAt": start, "maxResults": maxr, "total": len(keys), "issues": issues}
|
||
if p == "/rest/api/2/field":
|
||
return 200, self.fields
|
||
if p == "/rest/api/2/issue" and method == "POST":
|
||
self.seq += 1
|
||
key = "%s-%d" % (self.project, self.seq)
|
||
f = body["fields"]
|
||
self.add(key, summary=f.get("summary"), description=f.get("description"),
|
||
priority=(f.get("priority") or {}).get("name", "Средний"), duedate=f.get("duedate"),
|
||
labels=list(f.get("labels") or []), issuetype=self._type_name(f.get("issuetype")),
|
||
reporter="me", created=self.ts(self.now),
|
||
**{c: f.get(c) for c in CUSTOM})
|
||
return 201, {"id": "1", "key": key}
|
||
if p == "/rest/api/2/issueLinkType":
|
||
return 200, {"issueLinkTypes": LINK_TYPES}
|
||
if p == "/rest/api/2/issueLink" and method == "POST":
|
||
self.link_seq += 1
|
||
self.links.append({"id": self.link_seq, "type": body["type"]["name"],
|
||
"source": body["inwardIssue"]["key"], "target": body["outwardIssue"]["key"]})
|
||
return 201, {}
|
||
m = re.match(r"^/rest/api/2/issueLink/(\d+)$", p)
|
||
if m and method == "DELETE":
|
||
if not any(str(l["id"]) == m.group(1) for l in self.links):
|
||
return 404, {"errorMessages": ["Issue link does not exist: %s" % m.group(1)]}
|
||
self.links = [l for l in self.links if str(l["id"]) != m.group(1)]
|
||
return 204, None
|
||
m = re.match(r"^/rest/api/2/issue/([^/]+)$", p)
|
||
if m:
|
||
key = m.group(1)
|
||
if method == "GET":
|
||
out = self.issue_json(key)
|
||
if q.get("expand") == "changelog":
|
||
cap = self.issue_changelog_cap if self.issue_changelog_cap is not None else self.changelog_cap
|
||
out["changelog"] = self.changelog_json(key, cap)
|
||
return 200, out
|
||
if method == "PUT":
|
||
f = self.issues[key]
|
||
for k, v in body.get("fields", {}).items():
|
||
if k in ("priority", "assignee"):
|
||
f[k] = (v or {}).get("name")
|
||
elif k == "labels":
|
||
f[k] = list(v)
|
||
else:
|
||
f[k] = v
|
||
f["updated"] = "2026-10-11T13:00:00.000+0300"
|
||
return 204, None
|
||
m = re.match(r"^/rest/api/2/issue/([^/]+)/comment$", p)
|
||
if m:
|
||
key = m.group(1)
|
||
if method == "GET" and key in self.comment_denied:
|
||
return self.comment_denied[key], {"errorMessages": ["Нет прав на просмотр комментариев"]}
|
||
self.issues[key]
|
||
lst = self.comments.setdefault(key, [])
|
||
if method == "GET":
|
||
start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 1000))
|
||
return 200, {"startAt": start, "maxResults": maxr, "total": len(lst), "comments": lst[start:start + maxr]}
|
||
self.comment_seq += 1
|
||
ts = self.ts(self.now)
|
||
lst.append({"id": str(self.comment_seq), "body": body["body"], "author": self._user("me"),
|
||
"created": ts, "updated": ts})
|
||
return 201, lst[-1]
|
||
m = re.match(r"^/rest/api/2/issue/([^/]+)/comment/(\d+)$", p)
|
||
if m and method == "PUT":
|
||
for c in self.comments.get(m.group(1), []):
|
||
if c["id"] == m.group(2):
|
||
c["body"] = body["body"]
|
||
return 200, {}
|
||
m = re.match(r"^/rest/api/2/issue/([^/]+)/attachments$", p)
|
||
if m and method == "POST":
|
||
key = m.group(1)
|
||
self.issues[key]
|
||
if self.last_headers.get("x-atlassian-token") != "no-check":
|
||
return 403, {"errorMessages": ["XSRF check failed"]} # как Jira Server без заголовка
|
||
parts = parse_multipart(body["_raw"], body["_content_type"])
|
||
assert len(parts) == 1, "ожидалась одна часть, пришло %d" % len(parts)
|
||
part = parts[0]
|
||
disp = disposition_params(part["headers"].get("content-disposition", ""))
|
||
assert disp.get("name") == "file", "часть называется не file: %r" % disp
|
||
self.uploads.append({"key": key, "token": self.last_headers.get("x-atlassian-token"),
|
||
"content_type": body["_content_type"], "headers": part["headers"], "disposition": disp})
|
||
if self.attachment_limit is not None and len(part["data"]) > self.attachment_limit:
|
||
return 413, {"errorMessages": ["Размер вложения превышает предел %d байт" % self.attachment_limit]}
|
||
a = self.add_attachment(key, disp["filename"], part["data"],
|
||
part["headers"].get("content-type") or "application/octet-stream")
|
||
if self.bad_attachment_response:
|
||
return 200, []
|
||
return 200, [self.attachment_json(a)]
|
||
m = re.match(r"^/rest/api/2/issue/([^/]+)/transitions$", p)
|
||
if m:
|
||
key = m.group(1)
|
||
trs = WORKFLOW.get(self.issues[key]["status"], [])
|
||
if method == "GET":
|
||
return 200, {"transitions": [{"id": str(i + 1), "name": n, "to": {"name": to}} for i, (n, to) in enumerate(trs)]}
|
||
tid = int(body["transition"]["id"]) - 1
|
||
self.issues[key]["status"] = trs[tid][1]
|
||
return 204, None
|
||
m = re.match(r"^/rest/agile/1\.0/board/(\d+)$", p)
|
||
if m:
|
||
board = int(m.group(1))
|
||
if board not in self.boards:
|
||
return 404, {"errorMessages": ["Board does not exist"]}
|
||
return 200, {"id": board, "name": "Доска %d" % board, "type": self.boards[board]}
|
||
m = re.match(r"^/rest/agile/1\.0/board/(\d+)/sprint$", p)
|
||
if m:
|
||
board = int(m.group(1))
|
||
if board not in self.boards:
|
||
return 404, {"errorMessages": ["Board does not exist"]}
|
||
if self.boards.get(board) == "kanban":
|
||
return 400, {"errorMessages": ["The board does not support sprints"]}
|
||
states = (q.get("state") or "active,future").split(",")
|
||
vals = [s for s in self.board_sprints.get(board, self.sprints) if s["state"] in states]
|
||
start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 50))
|
||
return 200, {"values": vals[start:start + maxr], "isLast": start + maxr >= len(vals)}
|
||
m = re.match(r"^/rest/agile/1\.0/sprint/(\d+)/issue$", p)
|
||
if m:
|
||
sp = next(s for s in self.all_sprints() if s["id"] == int(m.group(1)))
|
||
if method == "GET":
|
||
members = self.sprint_members.get(sp["id"])
|
||
keys = [k for k in self.search_keys(q.get("jql") or "key in ()")
|
||
if (k in members if members is not None else self.issues[k]["sprint"] == sp["name"])]
|
||
start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 50))
|
||
return 200, {"startAt": start, "maxResults": maxr, "total": len(keys),
|
||
"issues": [{"key": k} for k in keys[start:start + maxr]]}
|
||
for k in body["issues"]:
|
||
self.issues[k]["sprint"] = sp["name"]
|
||
return 204, None
|
||
if p == "/rest/agile/1.0/backlog/issue" and method == "POST":
|
||
for k in body["issues"]:
|
||
self.issues[k]["sprint"] = None
|
||
return 204, None
|
||
m = re.match(r"^/rest/agile/1\.0/issue/([^/]+)$", p)
|
||
if m:
|
||
name = self.issues[m.group(1)]["sprint"]
|
||
sp = next((s for s in self.all_sprints() if s["name"] == name), None)
|
||
if sp is None: # как Jira банка (13.10.2026): без спринта ключа fields в ответе нет
|
||
return 200, {"key": m.group(1)}
|
||
return 200, {"key": m.group(1), "fields": {"sprint": sp}}
|
||
return 404, {"errorMessages": ["нет маршрута %s %s" % (method, p)]}
|
||
|
||
def all_sprints(self):
|
||
out = list(self.sprints)
|
||
for vals in self.board_sprints.values():
|
||
out += vals
|
||
return out
|
||
|
||
def _type_name(self, ref):
|
||
ref = ref or {}
|
||
if ref.get("name"):
|
||
return ref["name"]
|
||
return {"10100": "Задача", "12300": "Фича"}.get(str(ref.get("id")), "Задача")
|