214 lines
11 KiB
Python
214 lines
11 KiB
Python
import json
|
|
import shutil
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, Workspace
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.profiles.vcore import AdapterError, VcoreImporter
|
|
from viano_bug_analysis.profiles.vcore import ArchiveLimits
|
|
|
|
|
|
def test_pair_roundtrip_and_provenance(source_pair, tmp_path):
|
|
diag, src, diag_entries, src_entries = source_pair
|
|
destination = tmp_path / "renamed.package"
|
|
importer = VcoreImporter()
|
|
result = importer.create(diag, destination, wowsrc=src, package_id="stable", case_id="opaque-case")
|
|
assert result.valid and result.manifest["package_generation"] == 0
|
|
assert result.manifest["history"]["reproductions"] == 1
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(destination)
|
|
built = CaseBuilder(store).build("opaque-case", "stable")
|
|
assert PackageValidator().validate(built.package).valid
|
|
with PackageReader(built.package) as reader:
|
|
case = json.loads(reader.read_bytes(built.case_ref))
|
|
observations = case["human_observations"]["observations"]
|
|
assert [row["text"] for row in observations] == ["Wolpertinger spawnen nicht", "No spawns here; the other location worked", "Still no spawns"]
|
|
assert case["expected_behavior"]["completeness"] == "missing"
|
|
assert case["human_observations"]["tester_hypotheses"] == []
|
|
note = observations[-1]["provenance"]["origin"]
|
|
assert note["json_pointer"] == "/0/data_json" and note["decoded_json_pointer"] == "/text"
|
|
assert note["source_timestamp"] == "2026-10-08T13:02:55"
|
|
assert observations[-1]["provenance"]["observed_at"] is None
|
|
raw_note = json.loads(reader.read_bytes(note["source_ref"]))[0]["data_json"]
|
|
assert json.loads(raw_note)["text"] == observations[-1]["text"]
|
|
assert reader.read_bytes(note["archive_ref"]) == diag.read_bytes()
|
|
assert any(row["type"] == "quest" and row["id"] == "11117" and "diagnostic_focus" in row["roles"] for row in case["entities"])
|
|
assert any(issue["code"] == "SOURCE_TIME_UNQUALIFIED" for issue in case["data_quality"]["issues"])
|
|
assert any(ref["path"].endswith("source.cpp") for ref in case["source_context"]["refs"])
|
|
assert any(ref["path"].endswith("unknown.extension") for ref in case["inputs"]["unclassified"])
|
|
assert "Source timestamp (unchanged):" in reader.read_bytes(built.summary_ref).decode()
|
|
# All materialized raw files and both compressed originals are preserved.
|
|
for entries in (diag_entries, src_entries):
|
|
for name, data in entries.items():
|
|
paths = [path for path in reader.files if "/members/" in path and path.endswith("/" + name)]
|
|
expected = data if isinstance(data, bytes) else json.dumps(data).encode()
|
|
assert any(reader.read_bytes(path) == expected for path in paths)
|
|
assert CaseBuilder(store).build("opaque-case", "stable").reused
|
|
|
|
|
|
def test_filename_has_no_meaning(source_pair, tmp_path):
|
|
diag, src, _, _ = source_pair
|
|
renamed_diag, renamed_src = tmp_path / "unrelated.one", tmp_path / "unrelated.two"
|
|
shutil.copyfile(diag, renamed_diag)
|
|
shutil.copyfile(src, renamed_src)
|
|
first = VcoreImporter().prepare(diag, src)
|
|
again = VcoreImporter().prepare(renamed_diag, renamed_src)
|
|
assert first == again
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("tool", "other"), ("format_version", 99), ("command", "quest"), ("bug_id", 21), ("core_revision", "different"), ("input_wowdiag_format", 3)])
|
|
def test_mismatched_pair_rejected(source_pair, make_tar, tmp_path, field, value):
|
|
diag, _, _, src_entries = source_pair
|
|
src_entries["manifest.json"] = dict(src_entries["manifest.json"], **{field: value})
|
|
src = make_tar(src_entries)
|
|
destination = tmp_path / "out"
|
|
with pytest.raises(AdapterError):
|
|
VcoreImporter().create(diag, destination, wowsrc=src, package_id="p", case_id="c")
|
|
assert not destination.exists()
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["../outside", "/external/file", "missing"])
|
|
def test_copied_source_references_not_followed(source_pair, make_tar, path):
|
|
diag, _, _, entries = source_pair
|
|
entries["source/development_files_index.json"][0]["package_path"] = path
|
|
with pytest.raises(AdapterError) as error:
|
|
VcoreImporter().prepare(diag, make_tar(entries))
|
|
assert error.value.code == "SOURCE_REFERENCE"
|
|
|
|
|
|
def test_missing_copied_source_reported(source_pair, make_tar):
|
|
diag, _, _, entries = source_pair
|
|
entries["source/development_files_index.json"][0] = {"file": "/external/large.cpp", "copied": False, "reason_not_copied": "size_limit"}
|
|
plan = VcoreImporter().prepare(diag, make_tar(entries))
|
|
receipt = json.loads(plan.additions[plan.receipt_ref])
|
|
assert any(issue["code"] == "SOURCE_NOT_COPIED" for issue in receipt["issues"])
|
|
|
|
|
|
def test_wrong_copied_size_rejected(source_pair, make_tar):
|
|
diag, _, _, entries = source_pair
|
|
entries["source/development_files_index.json"][0]["size"] = 123
|
|
with pytest.raises(AdapterError) as error:
|
|
VcoreImporter().prepare(diag, make_tar(entries))
|
|
assert error.value.code == "SOURCE_SIZE"
|
|
|
|
|
|
def test_augment_preserves_old_content_and_rejects_duplicate(source_pair, tmp_path):
|
|
diag, src, _, _ = source_pair
|
|
first, second = tmp_path / "first", tmp_path / "second"
|
|
importer = VcoreImporter()
|
|
importer.create(diag, first, package_id="stable", case_id="opaque")
|
|
result = importer.augment(first, second, wowdiag=diag, wowsrc=src)
|
|
assert result.manifest["package_generation"] == 1
|
|
with PackageReader(first) as before, PackageReader(second) as after:
|
|
for name in before.files:
|
|
if name != "manifest.json":
|
|
assert before.read_bytes(name) == after.read_bytes(name)
|
|
with pytest.raises(AdapterError) as error:
|
|
importer.augment(second, tmp_path / "third", wowdiag=diag, wowsrc=src)
|
|
assert error.value.code == "IMPORT_EXISTS"
|
|
|
|
|
|
def test_case_identity_is_caller_owned(source_pair, tmp_path):
|
|
diag, _, _, _ = source_pair
|
|
result = VcoreImporter().create(diag, tmp_path / "BUG-999.zip", package_id="caller-package", case_id="caller-case")
|
|
assert result.manifest["package_id"] == "caller-package"
|
|
assert result.manifest["case_id"] == "caller-case"
|
|
|
|
|
|
def test_combined_import_budget(source_pair):
|
|
diag, src, _, _ = source_pair
|
|
with pytest.raises(AdapterError) as error:
|
|
VcoreImporter(ArchiveLimits(max_import_bytes=100)).prepare(diag, src)
|
|
assert error.value.code == "IMPORT_LIMIT"
|
|
|
|
|
|
def test_unresolved_row_relations_rejected(source_pair, make_tar):
|
|
_, src, entries, _ = source_pair
|
|
entries["bug/reproduction_events.json"][0]["reproduction_id"] = 999
|
|
with pytest.raises(AdapterError) as error:
|
|
VcoreImporter().prepare(make_tar(entries))
|
|
assert error.value.code == "SOURCE_RELATION"
|
|
|
|
|
|
def test_duplicate_source_ids_rejected(source_pair, make_tar):
|
|
_, _, entries, _ = source_pair
|
|
entries["bug/reports.json"] *= 2
|
|
with pytest.raises(AdapterError) as error:
|
|
VcoreImporter().prepare(make_tar(entries))
|
|
assert error.value.code == "SOURCE_ID"
|
|
|
|
|
|
def test_explicit_expected_behavior_is_copied_not_completed(source_pair, make_tar):
|
|
_, _, entries, _ = source_pair
|
|
entries["bug/reports.json"][0]["expected_behavior"] = "Supplied expected behavior"
|
|
plan = VcoreImporter().prepare(make_tar(entries))
|
|
index = json.loads(plan.additions[plan.index_ref])
|
|
report = json.loads(plan.additions[index["reports"][0]])
|
|
assert report["expected_behavior"][0]["text"] == "Supplied expected behavior"
|
|
assert "expected_behavior_complete" not in report
|
|
|
|
|
|
def test_aware_source_time_preserved(source_pair, make_tar):
|
|
_, _, entries, _ = source_pair
|
|
entries["bug/reports.json"][0]["created_at"] = "2026-10-08T13:02:21+02:00"
|
|
plan = VcoreImporter().prepare(make_tar(entries))
|
|
report = json.loads(plan.additions[json.loads(plan.additions[plan.index_ref])["reports"][0]])
|
|
assert report["observations"][0]["observed_at"] == "2026-10-08T13:02:21+02:00"
|
|
|
|
|
|
def test_bad_existing_receipt_is_controlled(source_pair, manifest, tmp_path):
|
|
from viano_bug_analysis.package import PackageFactory
|
|
diag, _, _, _ = source_pair
|
|
manifest["project"] = {"type": "wow", "profile": "vcore"}
|
|
manifest["artifacts"]["vcore_import"] = "bad-receipt"
|
|
original = tmp_path / "old"
|
|
PackageFactory().create(original, manifest, {"bad-receipt": b"[]"})
|
|
with pytest.raises(AdapterError) as error:
|
|
VcoreImporter().augment(original, tmp_path / "out", wowdiag=diag)
|
|
assert error.value.code == "IMPORT_RECEIPT"
|
|
assert not (tmp_path / "out").exists()
|
|
|
|
|
|
def test_copied_attachments_are_classified(source_pair, make_tar):
|
|
_, _, entries, _ = source_pair
|
|
entries["attachments/32/screenshot.bin"] = b"copied attachment"
|
|
plan = VcoreImporter().prepare(make_tar(entries))
|
|
index = json.loads(plan.additions[plan.index_ref])
|
|
assert len(index["attachments"]) == 1
|
|
assert plan.additions[index["attachments"][0]] == b"copied attachment"
|
|
|
|
|
|
def test_report_link_allows_null_direct_bug_id(source_pair, make_tar):
|
|
_, _, entries, _ = source_pair
|
|
entries["bug/reproductions.json"][0]["bug_id"] = None
|
|
assert VcoreImporter().prepare(make_tar(entries)).reproductions_count == 1
|
|
|
|
|
|
def test_explicit_runtime_entities_are_normalized(source_pair, make_tar):
|
|
_, _, entries, _ = source_pair
|
|
entries["bug/runtime_entities.json"] = [{"type": "creature", "id": 123, "reasons": ["recorded_event"]}]
|
|
plan = VcoreImporter().prepare(make_tar(entries))
|
|
index = json.loads(plan.additions[plan.index_ref])
|
|
records = [json.loads(plan.additions[path]) for path in index["reports"]]
|
|
entities = [entity for record in records for entity in record.get("entities", [])]
|
|
assert any(entity["id"] == 123 and entity["roles"] == ["runtime_observed"] and entity["origin"]["json_pointer"] == "/0" for entity in entities)
|
|
|
|
|
|
def test_adding_wowsrc_does_not_duplicate_original_observations(source_pair, tmp_path):
|
|
diag, src, _, _ = source_pair
|
|
first, augmented = tmp_path / "first", tmp_path / "augmented"
|
|
importer = VcoreImporter()
|
|
importer.create(diag, first, package_id="p", case_id="c")
|
|
importer.augment(first, augmented, wowdiag=diag, wowsrc=src)
|
|
store = CheckpointStore(Workspace(tmp_path / "service"))
|
|
store.intake(augmented)
|
|
result = CaseBuilder(store).build("c", "p")
|
|
with PackageReader(result.package) as reader:
|
|
case = json.loads(reader.read_bytes(result.case_ref))
|
|
assert len(case["human_observations"]["observations"]) == 3
|
|
with pytest.raises(AdapterError) as error:
|
|
importer.augment(augmented, tmp_path / "repeated", wowdiag=diag, wowsrc=src)
|
|
assert error.value.code == "IMPORT_EXISTS"
|