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
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,7 @@ dependencies = [
"requests>=2.32.3",
"requests-oauthlib>=2.0.0",
"s5cmd>=0.2.0",
"semver>=3.0.4",
"shapely>=2.1.1",
"show-in-file-manager>=1.1.5",
"tqdm>=4.67.1",
Expand Down
17 changes: 10 additions & 7 deletions src/aignostics/application/_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

import google_crc32c
import requests
import semver
from pydantic import BaseModel, computed_field

from aignostics.bucket import Service as BucketService
Expand Down Expand Up @@ -264,21 +265,23 @@ def application_version(
# Validate format: application_id:vX.Y.Z (where X.Y.Z is a semver)
# This checks for proper format like "he-tme:v0.50.0" where "he-tme" is the application id
# and "v0.50.0" is the version with proper semver format
if not re.match(r"^[^:]+:v\d+\.\d+\.\d+$", application_version_id):
match = re.match(r"^([^:]+):v(.+)$", application_version_id)
if not match or not semver.Version.is_valid(match.group(2)):
if use_latest_if_no_version_given:
latest_version = self.application_version_latest(self.application(application_version_id))
application_id = match.group(1) if match else application_version_id
latest_version = self.application_version_latest(self.application(application_id))
if latest_version:
return latest_version
message = f"No valid application version found for '{application_version_id}' "
message += "and no latest version available."
message = (
f"No valid application version found for '{application_version_id}'no latest version available."
)
logger.warning(message)
raise ValueError(message)
message = f"Invalid application version id format: {application_version_id}. "
"Expected format: application_id:vX.Y.Z"
message += "Expected format: application_id:vX.Y.Z"
raise ValueError(message)

application_id = application_version_id.split(":", maxsplit=1)[0]

application_id = match.group(1)
application = self.application(application_id)
for version in self.application_versions(application):
if version.application_version_id == application_version_id:
Expand Down
26 changes: 12 additions & 14 deletions src/aignostics/platform/resources/applications.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import typing as t
from operator import itemgetter

import semver
from aignx.codegen.api.public_api import PublicApi
from aignx.codegen.models import ApplicationReadResponse as Application
from aignx.codegen.models import ApplicationVersionReadResponse as ApplicationVersion
Expand All @@ -22,7 +23,7 @@ class Versions:
Provides operations to list and retrieve application versions.
"""

APPLICATION_VERSION_REGEX = re.compile(r"(?P<application_id>[^:]+):v?(?P<version>[^:]+)")
APPLICATION_VERSION_REGEX = re.compile(r"^(?P<application_id>[^:]+):v(?P<version>[^:].+)$")

def __init__(self, api: PublicApi) -> None:
"""Initializes the Versions resource with the API platform.
Expand Down Expand Up @@ -51,9 +52,6 @@ def list(self, application: Application | str) -> t.Iterator[ApplicationVersion]
application_id=application_id,
)

# TODO(Andreas,Helmut): Discuss. This is getting an application version and returning it.
# Can we make this a find_by_id, just getting an application_version_id as a string,
# and returning the application version object?
def details(self, application_version: ApplicationVersion | str) -> ApplicationVersion:
"""Retrieves details for a specific application version.

Expand All @@ -71,13 +69,14 @@ def details(self, application_version: ApplicationVersion | str) -> ApplicationV
application_id = application_version.application_id
version = application_version.version
else:
# split by colon
m = Versions.APPLICATION_VERSION_REGEX.match(application_version)
if not m:
# Parse and validate the application version ID
match = self.APPLICATION_VERSION_REGEX.match(application_version)
if not match:
msg = f"Invalid application_version_id: {application_version}"
raise RuntimeError(msg)
application_id = m.group("application_id")
version = m.group("version")

application_id = match.group("application_id")
version = match.group("version")

application_versions = self._api.list_versions_by_application_id_v1_applications_application_id_versions_get(
application_id=application_id,
Expand Down Expand Up @@ -106,18 +105,17 @@ def list_sorted(self, application: Application | str) -> builtins.list[Applicati
if not versions:
return []

# Extract semantic versions from the version property
# Extract semantic versions using proper semver parsing
versions_with_semver = []
for v in versions:
try:
# Split into major, minor, patch components for proper comparison
version_parts = [int(x) for x in v.version.split(".")]
versions_with_semver.append((v, version_parts))
parsed_version = semver.Version.parse(v.version)
versions_with_semver.append((v, parsed_version))
except (ValueError, AttributeError):
# If we can't parse the version or version attribute doesn't exist, skip it
continue

# Sort by semantic version (major, minor, patch)
# Sort by semantic version (semver objects have built-in comparison)
if versions_with_semver:
versions_with_semver.sort(key=itemgetter(1), reverse=True)
# Return just the version objects, not the tuples
Expand Down
83 changes: 83 additions & 0 deletions tests/aignostics/application/cli_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from aignostics.application import Service as ApplicationService
from aignostics.cli import cli
from aignostics.platform import NotFoundException
from aignostics.utils import sanitize_path
from tests.conftest import normalize_output, print_directory_structure

Expand Down Expand Up @@ -62,6 +63,88 @@ def test_cli_application_dump_schemata(runner: CliRunner, tmp_path: Path) -> Non
assert zip_file.exists(), f"Expected zip file {zip_file} not found"


def test_application_version_valid_semver_formats(runner: CliRunner) -> None:
"""Test that valid semver formats are accepted."""
from aignostics.application import Service as ApplicationService

service = ApplicationService()

# These should work if the application exists
valid_formats = [
"test-app:v1.0.0",
"test-app:v1.2.3",
"test-app:v10.20.30",
"test-app:v1.1.2-prerelease+meta",
"test-app:v1.1.2+meta",
"test-app:v1.1.2+meta-valid",
"test-app:v1.0.0-alpha",
"test-app:v1.0.0-beta",
"test-app:v1.0.0-alpha.beta",
"test-app:v1.0.0-alpha.1",
"test-app:v1.0.0-alpha0.beta",
"test-app:v1.0.0-alpha.alpha",
"test-app:v1.0.0-alpha+metadata",
"test-app:v1.0.0-rc.1+meta",
]

for version_id in valid_formats:
try:
service.application_version(version_id)
except ValueError as e:
pytest.fail(f"Valid semver format '{version_id}' was rejected: {e}")
except NotFoundException:
pytest.skip(f"Application '{version_id.split(':')[0]}' not found, skipping test for this version format.")


def test_application_version_invalid_semver_formats(runner: CliRunner) -> None:
"""Test that invalid semver formats are rejected with ValueError."""
from aignostics.application import Service as ApplicationService

service = ApplicationService()

invalid_formats = [
"test-app:1.0.0", # Missing 'v' prefix
"test-app:v1", # Incomplete version
"test-app:v1.0", # Incomplete version
"test-app:v1.0.0-", # Trailing dash
"test-app:v1.0.0+", # Trailing plus
"test-app:v1.0.0-+", # Invalid prerelease
"test-app:v1.0.0-+123", # Invalid prerelease
"test-app:v+invalid", # Invalid format
"test-app:v-invalid", # Invalid format
"test-app:v1.0.0.DEV.SNAPSHOT", # Too many version parts
"test-app:v1.0-SNAPSHOT-123", # Invalid format
"test-app:v", # Just 'v'
"test-app:vx.y.z", # Non-numeric
"test-app:v1.0.0-αα", # Non-ASCII in prerelease # noqa: RUF001
":v1.0.0", # Missing application ID
"test-app:", # Missing version
"no-colon-v1.0.0", # Missing colon separator
]

for version_id in invalid_formats:
with pytest.raises(ValueError, match=r"Invalid application version id format"):
service.application_version(version_id)


def test_application_version_use_latest_fallback(runner: CliRunner) -> None:
"""Test that use_latest_if_no_version_given works correctly."""
service = ApplicationService()

try:
result = service.application_version(HETA_APPLICATION_ID, use_latest_if_no_version_given=True)
assert result is not None
assert result.application_version_id.startswith(f"{HETA_APPLICATION_ID}:v")
except ValueError as e:
if "no latest version available" in str(e):
pass # This is expected behavior
else:
pytest.fail(f"Unexpected error: {e}")

with pytest.raises(ValueError, match=r"Invalid application version id format"):
service.application_version("invalid-format", use_latest_if_no_version_given=False)


def test_cli_application_run_prepare_upload_submit_fail_on_mpp(runner: CliRunner, tmp_path: Path) -> None:
"""Check application run prepare command and upload works and submit fails on mpp not supported."""
# Step 1: Prepare the file, by scanning for wsi and generating metadata
Expand Down
Loading