Skip to content
Merged
34 changes: 32 additions & 2 deletions tests/app/sep/snippets/test_script_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,9 @@
HTTPUnprocessableEntityException,
)
from app.core.pagination import Pagination
from app.core.security import crypto_timestamp_serializer
from app.sep.apps.framework.script_source import ScriptExecuteWrite
from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT
from app.sep.snippets.config import snippets_settings, SnippetSudoOption
from app.sep.snippets.crud import SnippetManager
from app.sep.snippets.deps import build_snippet_execution_meta
Expand Down Expand Up @@ -89,6 +91,34 @@ def _snippet_query(
)


def _normalize_snippet_source(url: str) -> tuple[str, dict[str, object]]:
"""Return the URL prefix plus the decoded token payload.

Keeps scheme, host, and path so parity tests still catch base-URL drift,
while replacing the timed token with its payload so two mintings in
different seconds still compare equal.

:param url: The signed artifact-download URL to normalize.
:return: ``(prefix, payload)`` where ``prefix`` ends at the final slash
and ``payload`` is the decoded token body.
"""
prefix, token = url.rsplit("/", 1)
payload = crypto_timestamp_serializer.loads(token, salt=ARTIFACT_DOWNLOAD_SALT)
return f"{prefix}/", payload


def _meta_dump_decoded(meta: SnippetExecutionMeta) -> dict[str, object]:
"""Return ``meta.model_dump()`` with ``snippet_source`` timestamp-normalized.

:param meta: The execution meta whose signed URL should be normalized.
:return: A dump identical to ``meta.model_dump()`` except ``snippet_source``
is ``(url_prefix, decoded_payload)``.
"""
d = meta.model_dump()
d["snippet_source"] = _normalize_snippet_source(d["snippet_source"])
return d


def _framework_processed_body(
script: SnippetScript, body: ScriptExecuteWrite
) -> ScriptExecuteWrite:
Expand Down Expand Up @@ -480,7 +510,7 @@ async def test_meta_matches_legacy_path(
)
legacy_meta = _legacy_execution_meta(script.snippet, body)

assert hook_meta.model_dump() == legacy_meta.model_dump()
assert _meta_dump_decoded(hook_meta) == _meta_dump_decoded(legacy_meta)

async def test_extra_args_reach_command_via_new_schema_alias(
self,
Expand Down Expand Up @@ -548,7 +578,7 @@ async def test_extra_args_still_bind_via_legacy_alias(
)
legacy_meta = _legacy_execution_meta(script.snippet, legacy_body)

assert new_alias_meta.model_dump() == legacy_meta.model_dump()
assert _meta_dump_decoded(new_alias_meta) == _meta_dump_decoded(legacy_meta)

@pytest.mark.parametrize("requested_sudo", [True, False])
async def test_optional_sudo_toggle_is_honored(
Expand Down
Loading