"""Маршруты вики и вложения по запросу (wiki-module, spec wiki «Маршруты вики», «Вложения вики по запросу»). Агента имитирует тест токеном agent wiki (фикстура api из test_wiki_sync).""" import base64 import os import urllib.error import urllib.request from urllib.parse import quote import pytest from jira_hub.core import files from jira_hub.core.client import HubError from test_wiki_convert import FILE_REF, FILE_URL, THUMB_REF, _html from test_wiki_sync import CODES, TREE, api, first_sync, page # noqa: F401 — api — фикстура ATTS = [{"ref": THUMB_REF, "url": FILE_URL, "name": "model.xlsx"}, {"ref": FILE_REF, "url": FILE_URL, "name": "model.xlsx"}] def synced(api, name_html=None, attachments=None): """Синхронизация дерева, где у DOC-3 есть вложение model.xlsx; возвращает id вложения.""" html = name_html if name_html is not None else _html("page_file.html") tree = [dict(p) for p in TREE] tree[2] = page("DOC-3", "Статусы", "DOC-1", 1, html, attachments=[{"url": FILE_URL, "name": "model.xlsx"}] if attachments is None else attachments) # вложения присылает агент уже разобранными: ref → url, как в test_wiki_sync tree[2]["attachments"] = ATTS if attachments is None else attachments assert api.run(tree, CODES)["status"] == "done" return api.rows("SELECT id FROM wiki_attachment ORDER BY id")[0]["id"] def upload(api, data: bytes) -> dict: return api.wiki.post("/agent/files", {"content": base64.b64encode(data).decode()}) def fetch_cmd(api): cmd = api.claim() assert cmd and cmd["type"] == "wiki_fetch", cmd return cmd def content(api, aid): req = urllib.request.Request(f"{api.base}/wiki/attachments/{aid}/content", headers={"Authorization": "Bearer " + api.client.token}) return urllib.request.urlopen(req) def err(fn, *a, **kw): with pytest.raises((HubError, urllib.error.HTTPError)) as e: fn(*a, **kw) if isinstance(e.value, urllib.error.HTTPError): e.value.close() # иначе ResourceWarning о незакрытом сокете return e.value def test_status_tree_page(api): aid = synced(api) st = api.client.get("/wiki/status", project="AWPCA") assert st["root"] == "DOC-1" and st["root_title"] == "АРМ КА" and st["pages"] == 3 and st["error"] is None assert st["synced_at"] and st["batch"]["status"] == "done" tree = api.client.get("/wiki/tree", project="AWPCA")["pages"] assert [(n["code"], n["depth"], n["parent_code"]) for n in tree] == [("DOC-1", 0, None), ("DOC-3", 1, "DOC-1"), ("DOC-2", 1, "DOC-1")] assert all(set(n) == {"code", "title", "depth", "parent_code", "size"} for n in tree) p = api.client.get("/wiki/pages/DOC-3", project="AWPCA") assert p["title"] == "Статусы" and p["url"].endswith("DOC-3") and p["synced_at"] assert f"attachment:{aid}" in p["markdown"] and not p["markdown"].startswith("---") assert p["attachments"] == [{"id": aid, "name": "model.xlsx", "state": "none", "size": None, "error": None}] assert err(api.client.get, "/wiki/pages/DOC-99", project="AWPCA").status == 404 def test_page_code_with_slash(api): first_sync(api) c = api.hub.conn() # код с «/» бывает только в адресе как %2F c.execute("UPDATE wiki_page SET code = 'A/B' WHERE code = 'DOC-2'") c.commit() c.close() assert api.client.get("/wiki/pages/" + quote("A/B", safe=""), project="AWPCA")["code"] == "A/B" def test_no_root(api): api.client.post("/projects", {"label": "sfm_kedr", "name": "СФМ"}) assert err(api.client.get, "/wiki/tree", project="sfm_kedr").error == "no_root" assert err(api.client.get, "/wiki/status", project="sfm_kedr").status == 409 assert err(api.client.get, "/wiki/tree", project="нет").status == 404 assert err(api.client.get, "/wiki/tree").status == 422 assert err(api.client.post, "/wiki/sync", {"project": "sfm_kedr"}).error == "no_root" def test_sync_in_progress(api): first_sync(api) r = api.client.post("/wiki/sync", {"project": "AWPCA", "author": "t"}) assert set(r) == {"batch_id"} e = err(api.client.post, "/wiki/sync", {"project": "AWPCA"}) assert e.status == 409 and e.error == "sync_in_progress" and e.detail["batch_id"] == r["batch_id"] assert api.client.get("/wiki/status", project="AWPCA")["batch"] == {"id": r["batch_id"], "status": "queued"} assert len(api.rows("SELECT id FROM batch WHERE project = 'AWPCA'")) == 2 def test_fetch_flow(api): aid = synced(api) data = b"\x00xlsx-bytes" * 100 r = api.client.post(f"/wiki/attachments/{aid}/fetch", {}) assert r == {"id": aid, "name": "model.xlsx", "page_code": "DOC-3", "url": FILE_URL, "state": "fetching", "size": None, "error": None} cmd = fetch_cmd(api) assert cmd["params"] == {"attachment_id": aid, "url": FILE_URL} assert err(content, api, aid).status == 409 f = upload(api, data) api.report(cmd, "done", {"file": f["file"], "size": f["size"]}) st = api.client.get(f"/wiki/attachments/{aid}") assert st["state"] == "ready" and st["size"] == len(data) and st["error"] is None with content(api, aid) as resp: assert resp.read() == data assert resp.headers["Content-Disposition"] == "attachment; filename*=UTF-8''model.xlsx" assert resp.headers["Content-Type"] == "application/octet-stream" and resp.headers["X-Content-Type-Options"] == "nosniff" again = api.client.post(f"/wiki/attachments/{aid}/fetch", {}) # ready — как есть, нового пакета нет assert again["state"] == "ready" and len(api.rows("SELECT id FROM command WHERE type = 'wiki_fetch'")) == 1 def test_fetch_repeat_while_fetching(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {}) r = api.client.post(f"/wiki/attachments/{aid}/fetch", {}) assert r["state"] == "fetching" and len(api.rows("SELECT id FROM command WHERE type = 'wiki_fetch'")) == 1 def test_fetch_requeues_after_cancel(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {}) batch = api.rows("SELECT batch_id FROM command WHERE type = 'wiki_fetch'")[0]["batch_id"] api.client.post(f"/batches/{batch}/cancel", {}) assert api.client.get(f"/wiki/attachments/{aid}")["state"] == "fetching" api.client.post(f"/wiki/attachments/{aid}/fetch", {}) # зависшее fetching — новый пакет assert len(api.rows("SELECT id FROM command WHERE type = 'wiki_fetch' AND status = 'queued'")) == 1 def test_fetch_too_big(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {}) api.report(fetch_cmd(api), "failed", error="файл больше 10 МБ — откройте в Eva") st = api.client.get(f"/wiki/attachments/{aid}") assert st["state"] == "failed" and st["error"] == "файл больше 10 МБ — откройте в Eva" e = err(content, api, aid) assert e.status == 409 assert api.client.post(f"/wiki/attachments/{aid}/fetch", {})["state"] == "fetching" # из failed — повтор def test_fetch_size_mismatch(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {}) cmd = fetch_cmd(api) f = upload(api, b"abc") api.report(cmd, "done", {"file": f["file"], "size": f["size"] + 1}) st = api.client.get(f"/wiki/attachments/{aid}") assert st["state"] == "failed" and "размер" in st["error"] assert err(content, api, aid).status == 409 def test_fetch_file_missing(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {}) api.report(fetch_cmd(api), "done", {"file": "0" * 64, "size": 5}) assert api.client.get(f"/wiki/attachments/{aid}")["state"] == "failed" def test_fetch_unknown_attachment(api): first_sync(api) assert err(api.client.post, "/wiki/attachments/999/fetch", {}).status == 404 assert err(api.client.get, "/wiki/attachments/999").status == 404 assert err(content, api, 999).status == 404 assert api.rows("SELECT id FROM batch WHERE title LIKE 'Вложение%'") == [] def test_fetch_url_not_from_client(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {"url": "https://evil.example/x", "author": "t"}) assert fetch_cmd(api)["params"]["url"] == FILE_URL def test_content_name_is_safe(api): aid = synced(api) c = api.hub.conn() f = upload(api, b"x") evil = 'a"\r\nSet-Cookie: x=1;/../é.txt' c.execute("UPDATE wiki_attachment SET state = 'ready', file = ?, size = 1, name = ? WHERE id = ?", (f["file"], evil, aid)) c.commit() c.close() with content(api, aid) as resp: disp = resp.headers["Content-Disposition"] assert disp == "attachment; filename*=UTF-8''" + quote(evil, safe="") assert "\r" not in disp and "\n" not in disp and resp.headers["Set-Cookie"] is None def test_file_refs_keep_ready(api): aid = synced(api) api.client.post(f"/wiki/attachments/{aid}/fetch", {}) cmd = fetch_cmd(api) f = upload(api, b"keep me") stale = upload(api, b"drop me") api.report(cmd, "done", {"file": f["file"], "size": f["size"]}) for sha in (f["file"], stale["file"]): os.utime(files.path(api.hub.files_dir, sha), (1, 1)) # старше KEEP_DAYS c = api.hub.conn() removed = files.sweep(c, api.hub.files_dir) c.close() assert stale["file"] in removed and f["file"] not in removed assert files.exists(api.hub.files_dir, f["file"]) def test_search_empty_query(api): first_sync(api) assert err(api.client.get, "/wiki/search", project="AWPCA", q="").error == "bad_query" assert err(api.client.get, "/wiki/search", project="AWPCA", q=" ").status == 422 assert err(api.client.get, "/wiki/search", project="AWPCA", q="x", limit=0).status == 422 assert err(api.client.get, "/wiki/search", project="AWPCA", q="x", limit=51).status == 422 def test_search_results(api): first_sync(api) r = api.client.get("/wiki/search", project="AWPCA", q="согласов")["results"] assert [x["code"] for x in r] == ["DOC-3"] and set(r[0]) == {"code", "title", "path", "snippet"} def test_page_attachment_failed_error_visible(api): """M-15: ответ страницы несёт error у вложения, текст ошибки виден без клика.""" aid = synced(api) c = api.hub.conn() try: c.execute("UPDATE wiki_attachment SET state = 'failed', error = 'Eva: 404' WHERE id = ?", (aid,)) finally: c.close() att = api.client.get("/wiki/pages/DOC-3", project="AWPCA")["attachments"][0] assert att["state"] == "failed" and att["error"] == "Eva: 404"