Artur Mukhamadiev
e65fbf4b67
FastAPI app, upstream Confluence/model adapters, authoritative history, rootless container lifecycle, artifact storage and downloads, fake peers under backend/dev, tests under tests/backend. Root pytest.ini deselects the live marker by default; requirements gain the backend dependencies.
113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
"""Unit tests for artifact validation, staging, storage, and downloads."""
|
|
|
|
import base64
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
import pytest
|
|
|
|
from backend.artifacts import (
|
|
ArtifactStore,
|
|
validate_artifact_name,
|
|
)
|
|
from backend.errors import InvalidInputError
|
|
|
|
|
|
def test_validate_artifact_name():
|
|
assert validate_artifact_name("checklist.md") == "checklist.md"
|
|
assert validate_artifact_name("docs/summary.txt") == "docs/summary.txt"
|
|
assert validate_artifact_name("تقرير.txt") == "تقرير.txt"
|
|
|
|
# Rejections
|
|
with pytest.raises(InvalidInputError):
|
|
validate_artifact_name("/absolute/path.txt")
|
|
|
|
with pytest.raises(InvalidInputError):
|
|
validate_artifact_name("../escape.txt")
|
|
|
|
with pytest.raises(InvalidInputError):
|
|
validate_artifact_name("dir/../escape.txt")
|
|
|
|
with pytest.raises(InvalidInputError):
|
|
validate_artifact_name("windows\\path.txt")
|
|
|
|
with pytest.raises(InvalidInputError):
|
|
validate_artifact_name("line\nbreak.txt")
|
|
|
|
with pytest.raises(InvalidInputError):
|
|
validate_artifact_name("a" * 1025)
|
|
|
|
|
|
def test_artifact_staging_and_commit(tmp_path: Path):
|
|
store = ArtifactStore(tmp_path / "artifacts")
|
|
staging = store.create_staging_session("query_123")
|
|
|
|
content = b"# Checklist\n\n- Deploy service X\n"
|
|
size = len(content)
|
|
assert size == 32
|
|
|
|
# 1. Begin transfer
|
|
dec, warn = staging.handle_begin("t_1", "checklist.md", size)
|
|
assert dec == "accept"
|
|
assert warn is None
|
|
|
|
# 2. Send chunk
|
|
b64_data = base64.b64encode(content).decode("utf-8")
|
|
staging.handle_chunk("t_1", index=0, data_base64=b64_data)
|
|
|
|
# 3. End transfer
|
|
end_dec, end_warn = staging.handle_end("t_1", size_bytes=size, chunks=1)
|
|
assert end_dec == "stored"
|
|
assert end_warn is None
|
|
|
|
# 4. Commit to session
|
|
session_cookie = "sess_abc123"
|
|
committed_list = staging.commit(session_id=session_cookie, ttl_seconds=60)
|
|
assert len(committed_list) == 1
|
|
art_info = committed_list[0]
|
|
assert art_info["name"] == "checklist.md"
|
|
assert art_info["size_bytes"] == 32
|
|
aid = art_info["id"]
|
|
|
|
# 5. Download with matching session
|
|
res = store.get_artifact_for_download(aid, session_id=session_cookie)
|
|
assert res is not None
|
|
fpath, fname, fsize = res
|
|
assert fname == "checklist.md"
|
|
assert fsize == 32
|
|
assert fpath.read_bytes() == content
|
|
|
|
# 6. Download with wrong session -> None (404)
|
|
assert store.get_artifact_for_download(aid, session_id="other_session") is None
|
|
|
|
# 7. Download with expired artifact -> None
|
|
# Force expiry
|
|
store._committed[aid].expires_at_ts = time.time() - 10
|
|
assert store.get_artifact_for_download(aid, session_id=session_cookie) is None
|
|
|
|
|
|
def test_artifact_size_skip(tmp_path: Path):
|
|
store = ArtifactStore(tmp_path / "artifacts")
|
|
staging = store.create_staging_session("query_skip")
|
|
|
|
# Try declaring 15 MiB file (limit is 10 MiB)
|
|
dec, warn = staging.handle_begin("t_large", "huge.bin", 15 * 1024 * 1024)
|
|
assert dec == "skip"
|
|
assert warn is not None
|
|
assert warn["code"] == "artifact_size_exceeded"
|
|
staging.discard()
|
|
|
|
|
|
def test_artifact_discard_on_failure(tmp_path: Path):
|
|
store = ArtifactStore(tmp_path / "artifacts")
|
|
staging = store.create_staging_session("query_fail")
|
|
|
|
dec, _ = staging.handle_begin("t_1", "file.txt", 10)
|
|
assert dec == "accept"
|
|
assert store.global_reserved_bytes == 10
|
|
|
|
# Failure occurred -> discard
|
|
staging.discard()
|
|
assert store.global_reserved_bytes == 0
|
|
assert not (store.staging_dir / "query_fail").exists()
|