Skip to content
Merged
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
27 changes: 25 additions & 2 deletions runtimeconditions_profiler/extension/identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,32 @@ def reference_object(reference: tuple[str, str]) -> dict[str, str]:
return {"id": reference[0], "version": reference[1]}


EXTENSION_FILENAME = "runtimeconditions.extension.yaml"


def short_identifier(identifier: str, version: str) -> str:
"""Drop the /<version>/runtimeconditions.extension.yaml tail of an https extension id.

Ids that are not https URLs, or that do not end that way, stay unchanged.
"""
tail = f"/{version}/{EXTENSION_FILENAME}"
if version and identifier.startswith("https://") and identifier.endswith(tail):
return identifier[: -len(tail)]
return identifier


def expand_identifier(identifier: str, version: str) -> str:
"""Rebuild the full extension id from a short https reference; anything else is returned as is."""
if version and identifier.startswith("https://") and not identifier.endswith("/" + EXTENSION_FILENAME):
return f"{identifier}/{version}/{EXTENSION_FILENAME}"
return identifier


def profile_reference(reference: tuple[str, str]) -> str:
"""Serialize an extension reference for a Runtime Conditions Profile."""
return f"{reference[0]}:{reference[1]}" if reference[1] else reference[0]
"""Serialize an extension reference for a Runtime Conditions Profile as <short-url>:<version>."""
identifier, version = reference
short = short_identifier(identifier, version)
return f"{short}:{version}" if version else short


def definition_identifier(metadata: dict[str, object]) -> str:
Expand Down
7 changes: 6 additions & 1 deletion runtimeconditions_profiler/profile/validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from typing import Any

from ..constants import API_VERSION
from ..extension.identity import parse_identifier
from ..extension.identity import expand_identifier, parse_identifier
from ..extension import ExtensionVocabulary
from ..models import Diagnostic, DiscoveryResult
from ..util import add_extension_closure, as_map
Expand Down Expand Up @@ -32,6 +32,11 @@ def validate(self, profile: dict[str, Any], discovery: DiscoveryResult) -> list[
if artifact.extension_id and artifact.extension_definition
}
definitions.update({(artifact.id, artifact.version): artifact.definition for artifact in discovery.sdk_extensions})
# A profile may name an extension by its short <url>:<version> reference.
extensions = [
(expand_identifier(*ref), ref[1]) if ref not in definitions and (expand_identifier(*ref), ref[1]) in definitions else ref
for ref in extensions
]
declared: set[tuple[str, str]] = set()
for extension in extensions:
if extension in declared:
Expand Down
41 changes: 41 additions & 0 deletions tests/test_extension_identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,3 +29,44 @@ def test_definition_identity_reads_id_and_version() -> None:
def test_invalid_extension_identity_is_rejected(identifier: object) -> None:
with pytest.raises(RuntimeConditionsError):
parse_identifier(identifier)


FULL = "https://runtimeconditions.io/extensions/rc/common-integrations/v1alpha1/runtimeconditions.extension.yaml"
SHORT = "https://runtimeconditions.io/extensions/rc/common-integrations"


def test_profile_reference_uses_the_short_url_and_version() -> None:
from runtimeconditions_profiler.extension.identity import profile_reference

assert profile_reference((FULL, "v1alpha1")) == f"{SHORT}:v1alpha1"


def test_profile_reference_keeps_ids_that_do_not_end_with_the_version_tail() -> None:
from runtimeconditions_profiler.extension.identity import profile_reference

assert profile_reference(("acme/jobs", "1.0.0")) == "acme/jobs:1.0.0"
assert profile_reference(("urn:acme:jobs", "")) == "urn:acme:jobs"
assert profile_reference((FULL, "v2")) == f"{FULL}:v2"


def test_expand_identifier_rebuilds_the_full_id() -> None:
from runtimeconditions_profiler.extension.identity import expand_identifier, parse_identifier

assert expand_identifier(*parse_identifier(f"{SHORT}:v1alpha1")) == FULL
assert expand_identifier(FULL, "v1alpha1") == FULL
assert expand_identifier("acme/jobs", "1.0.0") == "acme/jobs"


def test_validator_resolves_a_short_extension_reference() -> None:
from types import SimpleNamespace

from runtimeconditions_profiler.profile.validator import ProfileValidator

from runtimeconditions_profiler.extension.definition import parse_extension_definition

definition = parse_extension_definition({"metadata": {"id": FULL, "version": "v1alpha1"}, "spec": {}}, FULL, FULL)
artifact = SimpleNamespace(id=FULL, version="v1alpha1", definition=definition)
discovery = SimpleNamespace(validated_artifacts=[], sdk_extensions=[artifact])
profile = {"apiVersion": "runtimeconditions.io/v1alpha1", "kind": "RuntimeConditionsProfile", "extensions": [f"{SHORT}:v1alpha1"], "conditions": []}
messages = [item.message for item in ProfileValidator().validate(profile, discovery)]
assert not [m for m in messages if "missing extension definition" in m]