218 lines
7.6 KiB
Python
218 lines
7.6 KiB
Python
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from artemis.data_engine.evidence_sources import recorded_sources
|
|
|
|
|
|
def record(kind, target, **extra):
|
|
return {
|
|
"plugin_id": "adapter",
|
|
"source": {"kind": kind, "target_id": target, "label": "安卓业务后台"},
|
|
**extra,
|
|
}
|
|
|
|
|
|
def test_business_backend_name_does_not_become_an_execution_platform_or_assertion():
|
|
steps = [
|
|
{
|
|
"step_id": "step-real-id",
|
|
"step_number": 7,
|
|
"last_execution_result": {
|
|
"execution": [record("api", "backend", call_id="call-real-id", read_only=True)]
|
|
},
|
|
}
|
|
]
|
|
before = deepcopy(steps)
|
|
value = recorded_sources(steps)
|
|
source = value["sources"][0]
|
|
assert source["kind"] == "api"
|
|
assert source["category"] == "auxiliary"
|
|
assert source["references"] == [
|
|
{"step_id": "step-real-id", "step_number": 7, "call_id": "call-real-id", "read_only": True}
|
|
]
|
|
assert not any(key in source for key in ("passed", "verified", "verdict"))
|
|
assert steps == before
|
|
|
|
|
|
def test_sources_remain_separate_and_references_use_host_ids():
|
|
steps = [
|
|
{
|
|
"step_id": "s1",
|
|
"step_number": 1,
|
|
"action_taken": record("computer", "window", call_id="c1"),
|
|
},
|
|
{
|
|
"step_id": "s2",
|
|
"step_number": 2,
|
|
"action_taken": record("computer", "window", call_id="c2"),
|
|
},
|
|
{
|
|
"step_id": "s3",
|
|
"step_number": 3,
|
|
"action_taken": record("android", "phone", call_id="c3"),
|
|
},
|
|
]
|
|
value = recorded_sources(steps)
|
|
assert value["total_sources"] == 2
|
|
assert value["sources"][0]["reference_count"] == 2
|
|
assert [r["call_id"] for r in value["sources"][0]["references"]] == ["c1", "c2"]
|
|
assert value["sources"][0]["id"] == recorded_sources(steps[:1])["sources"][0]["id"]
|
|
assert value["sources"][1]["category"] == "interface"
|
|
|
|
|
|
def test_window_label_change_does_not_attribute_all_steps_to_the_first_title():
|
|
first = record("computer", "window", call_id="c1")
|
|
second = record("computer", "window", call_id="c2")
|
|
first["source"]["label"] = "first page"
|
|
second["source"]["label"] = "second page"
|
|
result = recorded_sources(
|
|
[{"step_id": "s1", "action_taken": first}, {"step_id": "s2", "action_taken": second}]
|
|
)
|
|
source = result["sources"][0]
|
|
assert result["total_sources"] == 1
|
|
assert source["reference_count"] == 2
|
|
assert source["label"] == ""
|
|
assert source["label_varies"] is True
|
|
|
|
|
|
def test_model_arguments_and_unsourced_images_do_not_fabricate_source_metadata():
|
|
value = recorded_sources(
|
|
[
|
|
{
|
|
"step_id": "s",
|
|
"step_number": 3,
|
|
"pre_image_name": "hash",
|
|
"action_taken": {"args": {"source": {"kind": "computer", "target_id": "claim"}}},
|
|
}
|
|
]
|
|
)
|
|
assert value["sources"] == []
|
|
assert value["unattributed_references"] == [{"step_id": "s", "step_number": 3}]
|
|
|
|
|
|
def test_summary_is_bounded_and_reports_omissions_explicitly():
|
|
steps = [{"step_id": str(i), "action_taken": record("file", str(i))} for i in range(10)]
|
|
value = recorded_sources(steps, limit=2)
|
|
assert len(value["sources"]) == 2
|
|
assert value["total_sources"] == 10
|
|
assert value["sources_omitted"] == 8
|
|
repeated = recorded_sources(
|
|
[{"step_id": str(i), "action_taken": record("api", "one")} for i in range(20)]
|
|
)
|
|
assert len(repeated["sources"][0]["references"]) == 8
|
|
assert repeated["sources"][0]["reference_count"] == 20
|
|
|
|
|
|
def test_partial_source_metadata_does_not_hide_unknown_actions_in_the_same_step():
|
|
value = recorded_sources(
|
|
[
|
|
{
|
|
"step_id": "mixed",
|
|
"last_execution_result": {
|
|
"execution": [
|
|
record("api", "service"),
|
|
{"action": "tap"},
|
|
]
|
|
},
|
|
}
|
|
]
|
|
)
|
|
assert value["total_sources"] == 1
|
|
assert value["unattributed_step_count"] == 1
|
|
|
|
|
|
def android_receipt():
|
|
return {
|
|
"trace_id": "capture-trace",
|
|
"type": "tool",
|
|
"name": "android_file",
|
|
"status": "success",
|
|
"payload": {
|
|
"result": {
|
|
"ok": True,
|
|
"action": "screenshot",
|
|
"verified": True,
|
|
"device_serial": "test-phone",
|
|
"trace_path": "artifacts/android/home.png",
|
|
"sha256": "a" * 64,
|
|
"captured_at": "2026-10-28T00:00:00Z",
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
def test_android_receipts_join_computer_sources_and_deduplicate_nested_tool_traces():
|
|
tool = android_receipt()
|
|
duplicate = {**deepcopy(tool), "trace_id": "outer-wrapper"}
|
|
steps = [
|
|
{
|
|
"step_id": "mixed",
|
|
"action_taken": record("computer", "window"),
|
|
"generic_tools": [tool, duplicate],
|
|
},
|
|
{"step_id": "final", "generic_tools": [tool]},
|
|
{"step_id": "empty", "action_taken": None, "generic_tools": []},
|
|
]
|
|
before = deepcopy(steps)
|
|
result = recorded_sources(steps)
|
|
sources = {source["kind"]: source for source in result["sources"]}
|
|
assert set(sources) == {"android", "computer"}
|
|
assert sources["android"]["target_id"] == "test-phone"
|
|
assert sources["android"]["reference_count"] == 2
|
|
assert sources["android"]["references"][0]["trace_id"] == "capture-trace"
|
|
assert sources["android"]["references"][0]["artifact_path"] == "artifacts/android/home.png"
|
|
assert result["unattributed_step_count"] == 0
|
|
assert steps == before
|
|
|
|
|
|
@pytest.mark.parametrize("invalid", ["failed", "unverified", "different_tool", "arguments_only"])
|
|
def test_only_successful_android_capture_receipts_supply_provenance(invalid):
|
|
tool = android_receipt()
|
|
if invalid == "failed":
|
|
tool["status"] = "failed"
|
|
elif invalid == "unverified":
|
|
tool["payload"]["result"]["verified"] = False
|
|
elif invalid == "different_tool":
|
|
tool["name"] = "read_note"
|
|
else:
|
|
tool["payload"] = {"args": tool["payload"]["result"]}
|
|
value = recorded_sources(
|
|
[{"step_id": "s", "action_taken": {"action": "tap"}, "generic_tools": [tool]}]
|
|
)
|
|
assert value["sources"] == []
|
|
assert value["unattributed_step_count"] == 1
|
|
|
|
|
|
def test_capture_receipt_does_not_cover_an_unattributed_action_in_the_same_step():
|
|
value = recorded_sources(
|
|
[{"step_id": "s", "action_taken": {"action": "tap"}, "generic_tools": [android_receipt()]}]
|
|
)
|
|
assert value["sources"][0]["kind"] == "android"
|
|
assert value["unattributed_step_count"] == 1
|
|
|
|
|
|
def test_flash_capture_receipt_replaces_only_its_matching_action_record():
|
|
tool = android_receipt()
|
|
action = {"action": "android_file", "args": {"action": "screenshot", "filename": "home.png"}}
|
|
value = recorded_sources([{"step_id": "s", "action_taken": action, "generic_tools": [tool]}])
|
|
assert value["sources"][0]["reference_count"] == 1
|
|
assert value["unattributed_step_count"] == 0
|
|
action["args"]["filename"] = "missing.png"
|
|
value = recorded_sources([{"step_id": "s", "action_taken": action, "generic_tools": [tool]}])
|
|
assert value["unattributed_step_count"] == 1
|
|
|
|
|
|
def test_note_and_final_report_actions_do_not_claim_device_evidence_or_missing_sources():
|
|
value = recorded_sources(
|
|
[
|
|
{
|
|
"step_id": "note",
|
|
"action_taken": {"action": "save_note", "args": {"content": "device claim"}},
|
|
},
|
|
{"step_id": "final", "action_taken": {"action": "report_task_status"}},
|
|
]
|
|
)
|
|
assert value["sources"] == []
|
|
assert value["unattributed_step_count"] == 0
|