chemenu/tools/chemenu/tests/test_upload.py

"""Tests for the MCP upload quarantine (Gitea #32) - `chemenu/upload.py`.

`upload.py` takes `root` explicitly and never touches `config.ROOT`, so these
tests point straight at `tmp_path` without repointing global state the way
`kb_dir`/`raw_dir` do.
"""
from __future__ import annotations

import base64
import json
import sys
from pathlib import Path

import pytest

from chemenu import upload
from chemenu.errors import ValidationError

CONTENT = b"hello world, this is a test submission\n"
CONTENT_B64 = base64.b64encode(CONTENT).decode("ascii")


def _cfg(**overrides) -> upload.UploadConfig:
    defaults = dict(
        identity_header="X-Forwarded-User",
        max_bytes=1024,
        allowed_extensions=(".md", ".pdf"),
        submissions_per_day=5,
        bytes_per_day=10_000,
    )
    defaults.update(overrides)
    return upload.UploadConfig(**defaults)


def _tree(root: Path) -> dict[str, bytes]:
    return {
        str(p.relative_to(root)): p.read_bytes()
        for p in sorted(root.rglob("*"))
        if p.is_file()
    }


# --- sanitize_filename ------------------------------------------------------

@pytest.mark.parametrize("name", ["a/b.md", "a\\b.md", "..", ".", "", "   ", ".hidden",
                                    "a\x00b.md"])
def test_sanitize_filename_rejects_unsafe_names(name):
    with pytest.raises(ValidationError):
        upload.sanitize_filename(name)


def test_sanitize_filename_accepts_a_bare_name():
    assert upload.sanitize_filename("report.md") == "report.md"


# --- read_config -------------------------------------------------------------

def test_read_config_absent_is_none(tmp_path):
    assert upload.read_config(tmp_path) is None


def test_read_config_malformed_json_is_an_error_not_unrestricted(tmp_path):
    (tmp_path / ".wikitool-upload.json").write_text("{not json", encoding="utf-8")
    with pytest.raises(ValidationError):
        upload.read_config(tmp_path)


def test_read_config_missing_field_is_an_error(tmp_path):
    (tmp_path / ".wikitool-upload.json").write_text(
        json.dumps({"schema": 1, "max_bytes": 100}), encoding="utf-8"
    )
    with pytest.raises(ValidationError):
        upload.read_config(tmp_path)


def test_read_config_extension_without_dot_is_rejected(tmp_path):
    (tmp_path / ".wikitool-upload.json").write_text(
        json.dumps({
            "schema": 1, "max_bytes": 100, "allowed_extensions": ["md"],
            "quota": {"submissions_per_day": 1, "bytes_per_day": 1},
        }),
        encoding="utf-8",
    )
    with pytest.raises(ValidationError):
        upload.read_config(tmp_path)


def test_read_config_rejects_a_non_object_json(tmp_path):
    (tmp_path / ".wikitool-upload.json").write_text("[1, 2, 3]", encoding="utf-8")
    with pytest.raises(ValidationError):
        upload.read_config(tmp_path)


@pytest.mark.parametrize("value", [-1, 0])
def test_read_config_rejects_a_non_positive_max_bytes(tmp_path, value):
    (tmp_path / ".wikitool-upload.json").write_text(
        json.dumps({
            "schema": 1, "max_bytes": value, "allowed_extensions": [".md"],
            "quota": {"submissions_per_day": 1, "bytes_per_day": 1},
        }),
        encoding="utf-8",
    )
    with pytest.raises(ValidationError):
        upload.read_config(tmp_path)


@pytest.mark.parametrize("key,value", [("submissions_per_day", 0), ("bytes_per_day", -5)])
def test_read_config_rejects_non_positive_quota_values(tmp_path, key, value):
    quota = {"submissions_per_day": 1, "bytes_per_day": 1}
    quota[key] = value
    (tmp_path / ".wikitool-upload.json").write_text(
        json.dumps({"schema": 1, "max_bytes": 100, "allowed_extensions": [".md"], "quota": quota}),
        encoding="utf-8",
    )
    with pytest.raises(ValidationError):
        upload.read_config(tmp_path)


def test_read_config_valid_file_parses(tmp_path):
    (tmp_path / ".wikitool-upload.json").write_text(
        json.dumps({
            "schema": 1, "identity_header": "X-User", "max_bytes": 2048,
            "allowed_extensions": [".MD", ".pdf"],
            "quota": {"submissions_per_day": 3, "bytes_per_day": 999},
        }),
        encoding="utf-8",
    )
    cfg = upload.read_config(tmp_path)
    assert cfg.identity_header == "X-User"
    assert cfg.max_bytes == 2048
    assert cfg.allowed_extensions == (".md", ".pdf")
    assert cfg.submissions_per_day == 3
    assert cfg.bytes_per_day == 999


def test_read_config_default_identity_header(tmp_path):
    (tmp_path / ".wikitool-upload.json").write_text(
        json.dumps({
            "schema": 1, "max_bytes": 10,
            "allowed_extensions": [".md"],
            "quota": {"submissions_per_day": 1, "bytes_per_day": 10},
        }),
        encoding="utf-8",
    )
    assert upload.read_config(tmp_path).identity_header == upload.DEFAULT_IDENTITY_HEADER


# --- the write choke point ----------------------------------------------------

def test_write_primitive_refuses_a_relative_escape(tmp_path):
    with pytest.raises(ValidationError):
        upload._write_atomic_within(tmp_path, Path("../outside.txt"), b"x")
    assert not (tmp_path / "outside.txt").exists()
    assert not (tmp_path / "mcp-upload").exists()


def test_write_primitive_refuses_an_absolute_path(tmp_path):
    with pytest.raises(ValidationError):
        upload._write_atomic_within(tmp_path, Path("/etc/passwd"), b"x")


@pytest.mark.skipif(sys.platform == "win32", reason="symlinks need admin on Windows")
def test_write_primitive_refuses_a_symlink_escape(tmp_path):
    outside = tmp_path / "outside"
    outside.mkdir()
    upload_dir = tmp_path / "mcp-upload"
    upload_dir.mkdir()
    (upload_dir / "escape").symlink_to(outside, target_is_directory=True)

    with pytest.raises(ValidationError):
        upload._write_atomic_within(tmp_path, Path("escape") / "pwned.txt", b"x")
    assert not (outside / "pwned.txt").exists()


def test_write_primitive_writes_inside_and_is_atomic(tmp_path):
    dest = upload._write_atomic_within(tmp_path, Path("sub") / "file.txt", b"payload")
    assert dest.read_bytes() == b"payload"
    # no leftover temp file
    assert list(dest.parent.glob(".*.tmp")) == []


# --- submit --------------------------------------------------------------------

def test_submit_happy_path_writes_manifest_and_content(tmp_path):
    manifest = upload.submit(
        tmp_path, _cfg(), filename="report.md", content_b64=CONTENT_B64, submitter="torben"
    )
    assert manifest["filename"] == "report.md"
    assert manifest["submitter"] == "torben"
    assert manifest["submitter_source"] == "X-Forwarded-User"
    assert manifest["size"] == len(CONTENT)

    submission_dir = tmp_path / "mcp-upload" / manifest["id"]
    assert (submission_dir / "report.md").read_bytes() == CONTENT
    stored_manifest = json.loads((submission_dir / "manifest.json").read_text())
    assert stored_manifest == manifest

    ledger_lines = (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()
    assert len(ledger_lines) == 1
    event = json.loads(ledger_lines[0])
    assert event["event"] == "submitted"
    assert event["submitter"] == "torben"


def test_submit_without_identity_writes_nothing(tmp_path):
    before = _tree(tmp_path)
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, _cfg(), filename="report.md", content_b64=CONTENT_B64, submitter=None)
    assert _tree(tmp_path) == before
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, _cfg(), filename="report.md", content_b64=CONTENT_B64, submitter="  ")
    assert _tree(tmp_path) == before


def test_submit_rejects_disallowed_extension(tmp_path):
    before = _tree(tmp_path)
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, _cfg(), filename="evil.exe", content_b64=CONTENT_B64, submitter="t")
    assert _tree(tmp_path) == before


def test_submit_rejects_oversized_before_decoding(tmp_path, monkeypatch):
    """The base64-length precheck fires before base64.b64decode is even
    called - simulated by making decode raise if it is ever reached."""
    import chemenu.upload as upload_module

    def _boom(*args, **kwargs):
        raise AssertionError("decode should not run past the length precheck")

    monkeypatch.setattr(upload_module.base64, "b64decode", _boom)
    before = _tree(tmp_path)
    huge_b64 = base64.b64encode(b"x" * 10_000).decode("ascii")
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, _cfg(max_bytes=10), filename="a.md", content_b64=huge_b64, submitter="t")
    assert _tree(tmp_path) == before


def test_submit_rejects_oversized_by_a_wide_margin(tmp_path):
    """`max_bytes` far below the real payload size - whichever of the two
    checks (base64-length precheck, post-decode check) fires, nothing is
    written."""
    cfg = _cfg(max_bytes=len(CONTENT) - 10)
    before = _tree(tmp_path)
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="t")
    assert _tree(tmp_path) == before


def test_submit_does_not_reject_a_payload_sitting_exactly_at_the_limit(tmp_path):
    """The base64-length precheck overestimates the decoded size by up to the
    2 padding characters a valid encoding carries - it must not turn that
    slack into a false rejection of a payload that is genuinely within
    `max_bytes`."""
    manifest = upload.submit(
        tmp_path, _cfg(max_bytes=len(CONTENT)), filename="a.md",
        content_b64=CONTENT_B64, submitter="t",
    )
    assert manifest["size"] == len(CONTENT)


def test_submit_rejects_one_byte_over_the_limit(tmp_path):
    """One byte over is still refused - whichever of the two checks catches
    it, the boundary itself is exact."""
    before = _tree(tmp_path)
    with pytest.raises(ValidationError):
        upload.submit(
            tmp_path, _cfg(max_bytes=len(CONTENT) - 1), filename="a.md",
            content_b64=CONTENT_B64, submitter="t",
        )
    assert _tree(tmp_path) == before


def test_submit_rejects_invalid_base64(tmp_path):
    before = _tree(tmp_path)
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, _cfg(), filename="a.md", content_b64="not base64!!", submitter="t")
    assert _tree(tmp_path) == before


def test_submit_rejects_empty_payload(tmp_path):
    with pytest.raises(ValidationError):
        upload.submit(tmp_path, _cfg(), filename="a.md", content_b64="", submitter="t")


def test_submit_rejects_a_duplicate_pending_hash_naming_the_waiting_id(tmp_path):
    first = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    with pytest.raises(ValidationError, match=first["id"]):
        upload.submit(tmp_path, _cfg(), filename="b.md", content_b64=CONTENT_B64, submitter="t")


def test_submit_allows_resubmission_after_rejection(tmp_path):
    first = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    upload.reject(tmp_path, first["id"], "not needed")
    second = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    assert second["id"] != first["id"]


def test_submit_enforces_submissions_per_day_quota(tmp_path):
    cfg = _cfg(submissions_per_day=1)
    upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="t")
    other_content = base64.b64encode(b"different content here\n").decode("ascii")
    with pytest.raises(ValidationError, match="Quota"):
        upload.submit(tmp_path, cfg, filename="b.md", content_b64=other_content, submitter="t")


def test_submit_enforces_bytes_per_day_quota(tmp_path):
    cfg = _cfg(bytes_per_day=len(CONTENT))
    upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="t")
    other_content = base64.b64encode(b"more bytes than allowed now\n").decode("ascii")
    with pytest.raises(ValidationError, match="Quota"):
        upload.submit(tmp_path, cfg, filename="b.md", content_b64=other_content, submitter="t")


def test_submit_quota_is_per_submitter(tmp_path):
    cfg = _cfg(submissions_per_day=1)
    upload.submit(tmp_path, cfg, filename="a.md", content_b64=CONTENT_B64, submitter="alice")
    other_content = base64.b64encode(b"different content again\n").decode("ascii")
    # bob has his own quota - not blocked by alice's submission
    upload.submit(tmp_path, cfg, filename="b.md", content_b64=other_content, submitter="bob")


# --- list / read ----------------------------------------------------------------

def test_list_submissions_oldest_first(tmp_path):
    a = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    other = base64.b64encode(b"second file content\n").decode("ascii")
    b = upload.submit(tmp_path, _cfg(), filename="b.md", content_b64=other, submitter="t")
    ids = [m["id"] for m in upload.list_submissions(tmp_path)]
    assert ids == sorted([a["id"], b["id"]])


def test_read_manifest_unknown_id_is_an_error(tmp_path):
    with pytest.raises(ValidationError):
        upload.read_manifest(tmp_path, "no-such-id")


def test_read_manifest_path_traversal_id_is_rejected(tmp_path):
    with pytest.raises(ValidationError):
        upload.read_manifest(tmp_path, "../../etc/passwd")


# --- confirm_token ----------------------------------------------------------------

def test_confirm_token_is_deterministic_and_moves_with_content(tmp_path):
    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    token1 = upload.confirm_token(manifest)
    token2 = upload.confirm_token(dict(manifest))
    assert token1 == token2
    mutated = dict(manifest, size=manifest["size"] + 1)
    assert upload.confirm_token(mutated) != token1


# --- promote ------------------------------------------------------------------

def test_promote_moves_file_deletes_dir_and_ledgers_accepted(tmp_path):
    (tmp_path / "incoming").mkdir()
    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    dest = upload.promote(tmp_path, manifest["id"])

    assert dest == tmp_path / "incoming" / "a.md"
    assert dest.read_bytes() == CONTENT
    assert not (tmp_path / "mcp-upload" / manifest["id"]).exists()

    events = [json.loads(line) for line in
              (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()]
    assert [e["event"] for e in events] == ["submitted", "accepted"]

    # nothing else in the tree changed
    for stray in ("raw", "kb", "work", "reports"):
        assert not (tmp_path / stray).exists()


def test_promote_unknown_id_is_rejected(tmp_path):
    with pytest.raises(ValidationError):
        upload.promote(tmp_path, "no-such-id")


def test_promote_refuses_when_the_file_is_missing_from_disk(tmp_path):
    """The manifest exists but the content file was removed underneath it -
    defense in depth, since nothing in this module deletes a content file
    without its manifest."""
    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    (tmp_path / "mcp-upload" / manifest["id"] / "a.md").unlink()

    with pytest.raises(ValidationError):
        upload.promote(tmp_path, manifest["id"])


def test_promote_refuses_when_incoming_already_has_the_name(tmp_path):
    (tmp_path / "incoming").mkdir()
    (tmp_path / "incoming" / "a.md").write_text("already here", encoding="utf-8")
    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")

    with pytest.raises(ValidationError):
        upload.promote(tmp_path, manifest["id"])

    # nothing moved: quarantine still holds it, incoming/ untouched
    assert (tmp_path / "mcp-upload" / manifest["id"] / "a.md").read_bytes() == CONTENT
    assert (tmp_path / "incoming" / "a.md").read_text() == "already here"


# --- reject -------------------------------------------------------------------

def test_reject_deletes_material_and_ledgers_with_reason(tmp_path):
    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    upload.reject(tmp_path, manifest["id"], "license unclear")

    assert not (tmp_path / "mcp-upload" / manifest["id"]).exists()
    events = [json.loads(line) for line in
              (tmp_path / "mcp-upload" / "ledger.jsonl").read_text().splitlines()]
    assert events[-1]["event"] == "rejected"
    assert events[-1]["reason"] == "license unclear"
    assert events[-1]["sha256"] == manifest["sha256"]


def test_reject_requires_a_reason(tmp_path):
    manifest = upload.submit(tmp_path, _cfg(), filename="a.md", content_b64=CONTENT_B64, submitter="t")
    with pytest.raises(ValidationError):
        upload.reject(tmp_path, manifest["id"], "")
    # nothing deleted
    assert (tmp_path / "mcp-upload" / manifest["id"]).exists()


def test_reject_unknown_id_is_rejected(tmp_path):
    with pytest.raises(ValidationError):
        upload.reject(tmp_path, "no-such-id", "reason")