249 lines
11 KiB
Python
249 lines
11 KiB
Python
import json
|
|
import shutil
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.lifecycle import FileLock, InboxProcessor, LifecycleError, State, Workspace
|
|
from viano_bug_analysis.lifecycle.storage import key
|
|
from viano_bug_analysis.package import PackageLimits, PackageReader
|
|
|
|
|
|
@pytest.fixture
|
|
def queue(tmp_path):
|
|
return InboxProcessor(Workspace(tmp_path / "service"))
|
|
|
|
|
|
def deliver(queue, source, name="anything.bin"):
|
|
path = queue.workspace.inbox / name
|
|
shutil.copyfile(source, path)
|
|
return path
|
|
|
|
|
|
def test_intake_and_renamed_duplicate(queue, make_package):
|
|
source = make_package(entries={"unknown": b"retained"})
|
|
before = source.read_bytes()
|
|
inbound = deliver(queue, source)
|
|
result = queue.process(inbound)
|
|
assert result.status == "ACCEPTED"
|
|
assert not inbound.exists()
|
|
assert next(queue.workspace.cache.glob("*.zip")).read_bytes() == before
|
|
duplicate = deliver(queue, source, "renamed.zip")
|
|
again = queue.process(duplicate)
|
|
assert again.status == "DUPLICATE"
|
|
assert again.package == result.package
|
|
assert not duplicate.exists()
|
|
assert len(list(result.package.parent.glob("g*.zip"))) == 1
|
|
|
|
|
|
def test_repacked_duplicate(queue, make_package):
|
|
import zipfile
|
|
first = make_package(entries={"unknown": b"retained"}, name="first")
|
|
original = queue.process(deliver(queue, first)).package
|
|
repacked = make_package(entries={"unknown": b"retained"}, name="second", compression=zipfile.ZIP_DEFLATED)
|
|
result = queue.process(deliver(queue, repacked))
|
|
assert result.status == "DUPLICATE"
|
|
assert result.package == original
|
|
|
|
|
|
def test_generation_conflict_quarantined(queue, make_package):
|
|
queue.process(deliver(queue, make_package(entries={"unknown": b"old"})))
|
|
conflict = deliver(queue, make_package(entries={"unknown": b"different"}))
|
|
result = queue.process(conflict)
|
|
assert result.status == "QUARANTINED"
|
|
assert result.code == "GENERATION_CONFLICT"
|
|
assert json.loads((result.package.parent / "validation_error.json").read_text())["issues"][0]["code"] == "GENERATION_CONFLICT"
|
|
|
|
|
|
@pytest.mark.parametrize("raw", [b"broken", b"{}", b"[]"])
|
|
def test_invalid_packages_preserved_in_quarantine(raw, queue, make_package):
|
|
source = make_package(raw=raw)
|
|
before = source.read_bytes()
|
|
result = queue.process(deliver(queue, source))
|
|
assert result.status == "QUARANTINED"
|
|
assert result.package.read_bytes() == before
|
|
assert (result.package.parent / "validation_error.json").exists()
|
|
|
|
|
|
def test_oversized_archive_quarantined_without_caching(tmp_path):
|
|
queue = InboxProcessor(Workspace(tmp_path / "service"), PackageLimits(max_archive_bytes=20))
|
|
source = queue.workspace.inbox / "huge"
|
|
source.write_bytes(b"x" * 100)
|
|
result = queue.process(source)
|
|
assert result.status == "QUARANTINED"
|
|
assert result.package.read_bytes() == b"x" * 100
|
|
assert not list(queue.workspace.cache.glob("*.zip"))
|
|
|
|
|
|
def test_case_lock_contention_leaves_input_pending(queue, make_package):
|
|
source = deliver(queue, make_package())
|
|
with FileLock(queue.workspace.runtime / "locks" / f"case-{key('case-A')}.lock"):
|
|
result = queue.process(source)
|
|
assert result.status == "BUSY"
|
|
assert source.exists()
|
|
assert queue.process(source).status == "ACCEPTED"
|
|
|
|
|
|
def test_worker_lock_contention(queue, make_package):
|
|
source = deliver(queue, make_package())
|
|
with FileLock(queue.workspace.runtime / "locks" / "worker.lock"):
|
|
assert queue.process(source).status == "BUSY"
|
|
assert queue.process(source).status == "ACCEPTED"
|
|
|
|
|
|
def test_future_blocked_case_does_not_hold_worker(queue, make_package, manifest):
|
|
first = queue.process(deliver(queue, make_package()))
|
|
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
|
blocked = queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="external-unavailable", reason="External dependency", retry_not_before=future)
|
|
# A returned blocked case keeps its retry time and does not occupy the worker.
|
|
other_service = InboxProcessor(Workspace(queue.workspace.root.parent / "other"))
|
|
deferred = other_service.process(deliver(other_service, blocked))
|
|
assert deferred.status == "DEFERRED"
|
|
manifest.update(case_id="case-B", package_id="other-package")
|
|
fresh = deliver(other_service, make_package(name="new-case"), "second.bin")
|
|
outcomes = other_service.poll()
|
|
assert any(outcome.status == "ACCEPTED" and outcome.source == fresh for outcome in outcomes)
|
|
|
|
|
|
def test_due_blocked_case_resumes_once(queue, make_package):
|
|
queue.process(deliver(queue, make_package()))
|
|
past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
|
|
queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="external-unavailable", reason="External dependency", retry_not_before=past)
|
|
outcomes = queue.poll()
|
|
assert [item.status for item in outcomes] == ["RESUMED"]
|
|
with PackageReader(outcomes[0].package) as reader:
|
|
assert reader.manifest["state"] == State.VALIDATED
|
|
assert reader.manifest["routing"]["retry_not_before"] is None
|
|
assert reader.manifest["routing"]["retry_count"] == 1
|
|
from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain
|
|
blocked_checkpoint = checkpoint_chain(reader)[1][1]
|
|
assert blocked_checkpoint["routing"]["reason"] == "External dependency"
|
|
assert blocked_checkpoint["routing"]["retry_not_before"] == past
|
|
assert queue.poll() == []
|
|
|
|
|
|
def test_crash_before_acknowledgement_does_not_generate_again(queue, make_package, monkeypatch):
|
|
source = deliver(queue, make_package())
|
|
def crash(*args):
|
|
raise LifecycleError("SIMULATED_CRASH", "before acknowledgement")
|
|
monkeypatch.setattr(queue, "_acknowledge", crash)
|
|
assert queue.process(source).status == "ERROR"
|
|
assert source.exists()
|
|
restarted = InboxProcessor(queue.workspace)
|
|
outcome = restarted.process(source)
|
|
assert outcome.status == "DUPLICATE"
|
|
assert len(list(outcome.package.parent.glob("g*.zip"))) == 1
|
|
|
|
|
|
def test_changed_inbox_file_is_not_deleted(queue, make_package, monkeypatch):
|
|
source = deliver(queue, make_package())
|
|
original = queue.store.intake
|
|
def change(snapshot):
|
|
result = original(snapshot)
|
|
source.write_bytes(b"new producer content")
|
|
return result
|
|
monkeypatch.setattr(queue.store, "intake", change)
|
|
outcome = queue.process(source)
|
|
assert outcome.status == "ERROR"
|
|
assert outcome.code == "SOURCE_CHANGED"
|
|
assert source.read_bytes() == b"new producer content"
|
|
|
|
|
|
def test_poll_limit_and_continues_after_bad_archive(queue, make_package):
|
|
(queue.workspace.inbox / "a-invalid").write_bytes(b"not zip")
|
|
deliver(queue, make_package(), "b-valid")
|
|
assert len(queue.poll(max_jobs=1)) == 1
|
|
assert queue.poll()[0].status == "ACCEPTED"
|
|
with pytest.raises(ValueError):
|
|
queue.poll(max_jobs=0)
|
|
|
|
|
|
def test_unsafe_source_rejected(queue, make_package):
|
|
source = make_package()
|
|
assert queue.process(source).code == "UNSAFE_INPUT"
|
|
assert source.exists()
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["cycle", "identity", "version", "generation", "state", "parent_path"])
|
|
def test_invalid_checkpoint_chain_quarantined(queue, make_package, change, tmp_path):
|
|
import hashlib
|
|
import zipfile
|
|
from viano_bug_analysis.lifecycle.checkpoints import CHECKPOINT_POINTER
|
|
admitted = queue.process(deliver(queue, make_package())).package
|
|
with zipfile.ZipFile(admitted) as archive:
|
|
entries = {name: archive.read(name) for name in archive.namelist()}
|
|
manifest = json.loads(entries["manifest.json"])
|
|
ref = manifest["artifacts"][CHECKPOINT_POINTER]
|
|
checkpoint = json.loads(entries[ref])
|
|
if change == "cycle":
|
|
checkpoint["parent"] = ref
|
|
elif change == "identity":
|
|
checkpoint["case_id"] = "other"
|
|
elif change == "version":
|
|
checkpoint["format_version"] = 2
|
|
elif change == "generation":
|
|
checkpoint["source_generation"] = 42
|
|
elif change == "state":
|
|
checkpoint["state"] = "BUILDING_CASE"
|
|
else:
|
|
checkpoint["parent"] = "../escape"
|
|
entries[ref] = json.dumps(checkpoint).encode()
|
|
manifest["integrity"][ref].update(size=len(entries[ref]), sha256=hashlib.sha256(entries[ref]).hexdigest())
|
|
entries["manifest.json"] = json.dumps(manifest).encode()
|
|
bad = tmp_path / "bad-checkpoint"
|
|
with zipfile.ZipFile(bad, "w") as archive:
|
|
for name, data in entries.items():
|
|
archive.writestr(name, data)
|
|
other = InboxProcessor(Workspace(tmp_path / "other-service"))
|
|
result = other.process(deliver(other, bad))
|
|
assert result.status == "QUARANTINED"
|
|
assert result.code == "CHECKPOINT_INVALID"
|
|
|
|
|
|
def test_return_without_old_files_is_quarantined(queue, make_package, manifest):
|
|
queue.process(deliver(queue, make_package(entries={"unknown": b"required"})))
|
|
manifest["package_generation"] = 2
|
|
manifest["state"] = "VALIDATED"
|
|
result = queue.process(deliver(queue, make_package()))
|
|
assert result.status == "QUARANTINED"
|
|
assert result.code == "CHECKPOINT_HISTORY_LOST"
|
|
|
|
|
|
def test_managed_corruption_does_not_quarantine_valid_inbound(queue, make_package):
|
|
source = make_package()
|
|
admitted = queue.process(deliver(queue, source)).package
|
|
admitted.write_bytes(b"corrupt managed package")
|
|
inbound = deliver(queue, source)
|
|
result = queue.process(inbound)
|
|
assert result.status == "ERROR"
|
|
assert result.code == "STORAGE_PACKAGE_INVALID"
|
|
assert inbound.exists()
|
|
|
|
|
|
def test_runtime_stale_on_crash_does_not_block_other_cases(queue, make_package, manifest, monkeypatch):
|
|
original_index = queue.store._index
|
|
def crash(*args):
|
|
raise LifecycleError("SIMULATED_CRASH", "after publication")
|
|
monkeypatch.setattr(queue.store, "_index", crash)
|
|
source = deliver(queue, make_package())
|
|
assert queue.process(source).status == "ERROR"
|
|
monkeypatch.setattr(queue.store, "_index", original_index)
|
|
manifest.update(case_id="second", package_id="second-package")
|
|
second = deliver(queue, make_package(name="second"), "second-case")
|
|
assert queue.process(second).status == "ACCEPTED"
|
|
assert queue.process(source).status == "DUPLICATE"
|
|
|
|
|
|
def test_deferred_jobs_do_not_consume_admission_limit(queue, make_package, manifest, tmp_path):
|
|
queue.process(deliver(queue, make_package()))
|
|
future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
|
|
blocked = queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="block", reason="Dependency", retry_not_before=future)
|
|
other = InboxProcessor(Workspace(tmp_path / "another-service"))
|
|
for index in range(3):
|
|
deliver(other, blocked, f"a-blocked-{index}")
|
|
manifest.update(case_id="new-case", package_id="new-package")
|
|
fresh = deliver(other, make_package(name="new"), "z-fresh")
|
|
outcomes = other.poll(max_jobs=1)
|
|
assert [item.status for item in outcomes] == ["DEFERRED", "DEFERRED", "DEFERRED", "ACCEPTED"]
|
|
assert not fresh.exists()
|