281 lines
13 KiB
Python
281 lines
13 KiB
Python
"""Маршруты вики и вложения по запросу (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 sdlc_hub.core import files
|
||
from sdlc_hub.core.client import HubError
|
||
from sdlc_hub.modules.wiki import attachments
|
||
from test_wiki_convert import FILE_REF, FILE_URL, IMG_REF, 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, "image": False}]
|
||
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"
|
||
|
||
|
||
@pytest.mark.parametrize("project", [5, ["AWPCA"], {"a": 1}, True])
|
||
def test_sync_project_not_string_422(api, project):
|
||
first_sync(api)
|
||
before = api.rows("SELECT id FROM batch")
|
||
e = err(api.client.post, "/wiki/sync", {"project": project})
|
||
assert e.status == 422 and e.error == "bad_request" and e.detail == "project — строка"
|
||
assert api.rows("SELECT id FROM batch") == before # пакета не создано
|
||
|
||
|
||
# пакет D (#135): T8.2 — вложение без файла, признак картинки
|
||
def _ready(api, aid, data=b"hello"):
|
||
api.client.post(f"/wiki/attachments/{aid}/fetch", {})
|
||
f = upload(api, data)
|
||
api.report(fetch_cmd(api), "done", {"file": f["file"], "size": f["size"]})
|
||
return f["file"]
|
||
|
||
|
||
def test_fetch_ready_file_missing_requeues(api):
|
||
aid = synced(api)
|
||
sha = _ready(api, aid)
|
||
(api.wiki_dir.parent / "files" / sha).unlink()
|
||
r = api.client.post(f"/wiki/attachments/{aid}/fetch", {})
|
||
assert r["state"] == "fetching"
|
||
assert fetch_cmd(api)["params"]["attachment_id"] == aid
|
||
|
||
|
||
def test_fetch_ready_file_present_no_batch(api):
|
||
aid = synced(api)
|
||
_ready(api, aid)
|
||
n = len(api.rows("SELECT id FROM batch"))
|
||
assert api.client.post(f"/wiki/attachments/{aid}/fetch", {})["state"] == "ready"
|
||
assert len(api.rows("SELECT id FROM batch")) == n
|
||
|
||
|
||
def test_page_attachment_image_flag(api):
|
||
html = f'<p><a href="{FILE_REF}">файл</a> <img src="{IMG_REF}"></p>'
|
||
synced(api, html, [{"ref": FILE_REF, "url": FILE_URL, "name": "attachment"},
|
||
{"ref": IMG_REF, "url": "https://eva.example.test" + IMG_REF, "name": "Схема без расширения"}])
|
||
atts = api.client.get("/wiki/pages/DOC-3", project="AWPCA")["attachments"]
|
||
assert [(a["name"], a["image"]) for a in atts] == [("attachment", False), ("Схема без расширения", True)]
|
||
|
||
|
||
@pytest.mark.parametrize("url,expected", [("https://e/files/a/x.PNG?version=1", True), ("https://e/files/a/x.svg", False),
|
||
("https://e/files/a/%D1%81.jpeg", True), (None, False),
|
||
("https://e/files/a/x.png.meta/", False), ("https://e/files/a/x.xlsx", False)])
|
||
def test_is_image(url, expected):
|
||
assert attachments.is_image(url) is expected
|