261 lines
7.6 KiB
Python
261 lines
7.6 KiB
Python
"""Tests targeting previously uncovered branches."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def _status(base: str, path: str, method: str = "GET", body: bytes | None = None):
|
|
req = urllib.request.Request(
|
|
f"{base}{path}",
|
|
data=body,
|
|
headers={"Content-Type": "application/json"} if body else {},
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
return resp.status, resp.read()
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code, exc.read()
|
|
|
|
|
|
# --- server.py ---
|
|
|
|
|
|
def test_file_preview_bad_docx_zip():
|
|
import server
|
|
|
|
out = server.file_preview_payload_bytes(b"PK\x03\x04garbage", "bad.docx")
|
|
assert out["ok"] is False
|
|
assert "DOCX" in out["error"] or "Could not" in out["error"]
|
|
|
|
|
|
def test_file_preview_quip_html():
|
|
import server
|
|
|
|
html = '<div class="quip-canvas-content"><p>Canvas</p></div>'
|
|
out = server.file_preview_payload_bytes(html.encode(), "doc.quip")
|
|
assert out["ok"] and "Canvas" in out["html"]
|
|
|
|
|
|
def test_static_file_not_found(api):
|
|
status, _ = _status(api["base"], "/static/no-such-asset.css")
|
|
assert status == 404
|
|
|
|
|
|
def test_uploads_missing_file(api):
|
|
status, _ = _status(api["base"], "/uploads/__uploads/missing/deadbeef.bin")
|
|
assert status in (400, 404)
|
|
|
|
|
|
def test_file_preview_by_file_id_not_found(api):
|
|
status, body = _status(api["base"], "/api/file-preview?file_id=UNKNOWN_FILE_XYZ")
|
|
assert status == 404
|
|
data = json.loads(body.decode())
|
|
assert data["ok"] is False
|
|
|
|
|
|
def test_file_preview_by_rel_on_disk(api, review_server):
|
|
rel_dir = review_server["httpd"].RequestHandlerClass.export.paths.root
|
|
target = rel_dir / "__uploads" / "T1"
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
(target / "note.txt").write_text("fixture preview text", encoding="utf-8")
|
|
status, body = _status(api["base"], "/api/file-preview?rel=__uploads/T1/note.txt")
|
|
assert status == 200
|
|
data = json.loads(body.decode())
|
|
assert data["ok"]
|
|
assert "fixture preview" in data["html"]
|
|
|
|
|
|
def test_manifest_post_empty_body(api):
|
|
status, _ = _status(
|
|
api["base"],
|
|
"/api/manifest",
|
|
method="POST",
|
|
body=b"",
|
|
)
|
|
assert status == 400
|
|
|
|
|
|
def test_manifest_post_wrong_path(api):
|
|
status, _ = _status(api["base"], "/api/manifest/extra", method="POST", body=b"{}")
|
|
assert status == 404
|
|
|
|
|
|
def test_manifest_post_invalid_import_mode_defaults(api):
|
|
api["post"](
|
|
"/api/manifest",
|
|
{
|
|
"channels": {
|
|
"medical": {
|
|
"status": "partial",
|
|
"import_mode": "bogus",
|
|
"messages": {},
|
|
}
|
|
}
|
|
},
|
|
)
|
|
med = api["get"]("/api/manifest")["channels"]["medical"]
|
|
assert med["import_mode"] == "all"
|
|
|
|
|
|
def test_manifest_post_skips_non_dict_user(api):
|
|
users = api["get"]("/api/users")["users"]
|
|
uid = users[0]["id"]
|
|
api["post"](
|
|
"/api/manifest",
|
|
{"users": {"bad": "nope", uid: {"import": True, "notes": "", "mattermost_username": ""}}},
|
|
)
|
|
again = api["get"]("/api/users")["users"]
|
|
assert any(u["id"] == uid for u in again)
|
|
|
|
|
|
def test_safe_export_path_rejects_dotdot():
|
|
from unittest.mock import Mock
|
|
|
|
import server
|
|
from slack_export import SlackExport
|
|
|
|
root = Path(__file__).resolve().parents[1] / "tests/fixtures/mini-export"
|
|
handler = Mock(spec=server.ReviewHandler)
|
|
handler.export = SlackExport(root)
|
|
assert server.ReviewHandler._safe_export_path(handler, "../outside") is None
|
|
assert server.ReviewHandler._safe_export_path(handler, "/etc/passwd") is None
|
|
|
|
|
|
# --- slack_export.py ---
|
|
|
|
|
|
@pytest.fixture
|
|
def rich_export(tmp_path):
|
|
from slack_export import SlackExport
|
|
|
|
root = tmp_path / "rich"
|
|
root.mkdir()
|
|
(root / "channels.json").write_text("[]", encoding="utf-8")
|
|
(root / "users.json").write_text(
|
|
json.dumps([{"id": "U1", "name": "u", "profile": {"real_name": "U"}}]),
|
|
encoding="utf-8",
|
|
)
|
|
(root / "canvases.json").write_text(
|
|
json.dumps([{"id": "CV1", "name": "canvas.pdf", "mimetype": "application/pdf"}]),
|
|
encoding="utf-8",
|
|
)
|
|
ch = root / "ch1"
|
|
ch.mkdir()
|
|
(ch / "meta.json").write_text('{"not": "a list"}', encoding="utf-8")
|
|
(ch / "2016-01-01.json").write_text(
|
|
json.dumps(
|
|
[
|
|
{"subtype": "channel_leave", "ts": "1.0"},
|
|
{"subtype": "channel_purpose", "purpose": "Purpose here", "ts": "2.0"},
|
|
{"files": [{"id": "NO_NAME"}]},
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return SlackExport(root)
|
|
|
|
|
|
def test_file_index_from_canvases_and_skips_bad_rows(rich_export):
|
|
idx = rich_export.ensure_file_index()
|
|
assert "CV1" in idx
|
|
assert rich_export.file_record("NO_NAME") is None
|
|
|
|
|
|
def test_fetch_no_download_url(rich_export):
|
|
rich_export._file_index = {
|
|
"X1": {
|
|
"id": "X1",
|
|
"name": "a.bin",
|
|
"local_relpath": "__uploads/X1/a.bin",
|
|
"local_exists": False,
|
|
"url_private_download": None,
|
|
}
|
|
}
|
|
with pytest.raises(FileNotFoundError, match="No download URL"):
|
|
rich_export.fetch_file_bytes("X1")
|
|
|
|
|
|
def test_fetch_http_error(rich_export, monkeypatch):
|
|
rich_export._file_index = {
|
|
"X2": {
|
|
"id": "X2",
|
|
"name": "b.bin",
|
|
"local_relpath": "__uploads/X2/b.bin",
|
|
"local_exists": False,
|
|
"url_private_download": "https://example.com/x?token=xoxe-REDACTED",
|
|
}
|
|
}
|
|
|
|
def _fail(*args, **kwargs):
|
|
raise urllib.error.HTTPError("url", 401, "Unauthorized", None, None)
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", _fail)
|
|
with pytest.raises(FileNotFoundError, match="401"):
|
|
rich_export.fetch_file_bytes("X2")
|
|
|
|
|
|
def test_normalize_leave_and_purpose(rich_export):
|
|
msgs = rich_export.iter_channel_messages("ch1")
|
|
texts = [m["text"] for m in msgs]
|
|
assert any("left" in t for t in texts)
|
|
assert any("Purpose" in t for t in texts)
|
|
|
|
|
|
def test_read_non_list_json_returns_empty(rich_export):
|
|
assert rich_export.read_messages_file("ch1", "meta") == []
|
|
|
|
|
|
def test_list_dates_missing_channel(rich_export):
|
|
assert rich_export.list_channel_date_files("missing-channel") == []
|
|
|
|
|
|
def test_search_files_query_filters_out(rich_export):
|
|
ch = rich_export.paths.root / "hits"
|
|
ch.mkdir()
|
|
(ch / "2017-01-01.json").write_text(
|
|
json.dumps(
|
|
[
|
|
{
|
|
"type": "message",
|
|
"user": "U1",
|
|
"ts": "1.0",
|
|
"text": "alpha",
|
|
"files": [{"id": "F9", "name": "pic.png", "mimetype": "image/png"}],
|
|
}
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
ex = rich_export
|
|
out = ex.search_files_global("zzzznomatch", file_type="image", limit=10)
|
|
assert out["hits"] == []
|
|
|
|
|
|
def test_search_global_hits_limit(export):
|
|
out = export.search_global("a", limit=1)
|
|
assert len(out["messages"]) <= 1
|
|
|
|
|
|
def test_search_messages_stops_at_limit(export):
|
|
hits = export.search_messages("zoey", "a", limit=2)
|
|
assert len(hits) <= 2
|
|
|
|
|
|
def test_classify_word_doc(export):
|
|
assert export.classify_file({"name": "x.doc", "mimetype": "application/msword"}) == "doc"
|
|
|
|
|
|
def test_register_file_raw_skips_incomplete(export):
|
|
export._file_index = {}
|
|
export._register_file_raw({"name": "onlyname"})
|
|
export._register_file_raw({"id": "idonly"})
|
|
assert export.file_record("onlyname") is None
|
|
assert export.file_record("idonly") is None
|