Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 17 additions & 8 deletions tests/control_plane/test_remote_location_shape_owner.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,21 +38,24 @@
# (site label, entry point, the message that site raises for a raw location).
# Each row goes through the real entry point so the test proves the wiring, not just
# the pattern; the messages differ per site on purpose.
SITES: list[tuple[str, Callable[[str], str], str]] = [
SITES: list[tuple[str, Callable[[str], str], str, str]] = [
(
"decision_context",
lambda value: packets._compact_text(value, field="source_ref"),
"must use an opaque source reference, not a raw URL",
"must not contain a local path",
),
(
"material_lifecycle",
lambda value: _validation.compact_text(value, field="source_ref"),
"must use an opaque reference, not a raw URL",
"must not contain a local path",
),
(
"ml_experiment",
lambda value: ml_experiment._compact_public_text(value, field="dataset_ref"),
"must use a public alias, not a raw URL or remote path",
"must use a public alias, not a raw URL or remote path",
),
]

Expand Down Expand Up @@ -99,22 +102,28 @@ def test_owner_shape_still_leaves_unlisted_schemes_alone() -> None:
)


@pytest.mark.parametrize("label,call,message", SITES)
@pytest.mark.parametrize("label,call,remote_message,file_url_message", SITES)
def test_each_site_rejects_a_raw_location_through_its_own_entry_point(
label: str, call: Callable[[str], str], message: str
label: str,
call: Callable[[str], str],
remote_message: str,
file_url_message: str,
) -> None:
for value in (
"s3://loopx-artifacts/run-7/metrics.json",
"file:///Users/dev/model.bin",
for value, message in (
("s3://loopx-artifacts/run-7/metrics.json", remote_message),
("file:///Users/dev/model.bin", file_url_message),
):
with pytest.raises(ValueError) as caught:
call(value)
assert message in str(caught.value), (label, value)


@pytest.mark.parametrize("label,call,_message", SITES)
@pytest.mark.parametrize("label,call,_remote_message,_file_url_message", SITES)
def test_each_site_still_accepts_an_opaque_reference(
label: str, call: Callable[[str], str], _message: str
label: str,
call: Callable[[str], str],
_remote_message: str,
_file_url_message: str,
) -> None:
"""Positive control on the same entry point with no injected fault."""
assert call("run-7/metrics.json") == "run-7/metrics.json"
Expand Down
2 changes: 1 addition & 1 deletion tests/test_loopx_turn_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -1061,7 +1061,7 @@ def test_enabled_host_result_rejects_receipt_local_path() -> None:
(
"evidence_refs",
["file:/tmp/private-result.json"],
"opaque 1-192 character public-safe reference",
"contains a local path",
),
],
)
Expand Down
Loading