Commit b0e20d05 authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

add payload normalization helpers

parent c2b80bab

edge_cloud_management_api/controllers/app_controllers.py

+81 −6
Original line number Diff line number Diff line
@@ -6,6 +6,8 @@ from edge_cloud_management_api.services.federation_services import FederationMan 
from edge_cloud_management_api.services.storage_service import get_zone

from edge_cloud_management_api.services.storage_service import get_fed

import json

import re

import uuid



factory = FederationManagerClientFactory()

federation_client = factory.create_federation_client()
@@ -14,6 +16,79 @@ class NotFound404Exception(Exception): 
    pass





def _ensure_gsma_id(value, pattern, prefix, min_len, max_len, fallback_source):

    if value and re.match(pattern, value):

        return value

    base = re.sub(r"[^A-Za-z0-9_]", "", str(fallback_source or ""))

    candidate = f"{prefix}{base}"

    if not re.match(r"^[A-Za-z]", candidate):

        candidate = f"{prefix}{candidate}"

    candidate = candidate[:max_len]

    if len(candidate) < min_len:

        candidate = (candidate + uuid.uuid4().hex)[:max_len]

    if not re.match(r"^[A-Za-z]", candidate):

        candidate = f"{prefix}{candidate}"

        candidate = candidate[:max_len]

    return candidate





def _ensure_service_name(value, prefix, fallback_source):

    pattern = r"^[A-Za-z0-9][A-Za-z0-9_]{6,62}[A-Za-z0-9]$"

    if value and re.match(pattern, value):

        return value

    base = re.sub(r"[^A-Za-z0-9_]", "", str(fallback_source or ""))

    candidate = f"{prefix}{base}"

    candidate = re.sub(r"^_+", "", candidate)

    candidate = candidate[:64]

    if len(candidate) < 8:

        candidate = (candidate + uuid.uuid4().hex)[:64]

    if not re.match(r"^[A-Za-z0-9]", candidate):

        candidate = f"s{candidate}"

    if not re.match(r"[A-Za-z0-9]$", candidate):

        candidate = f"{candidate}0"

    return candidate[:64]





def _split_image_reference(image_path):

    if not image_path:

        return None, None

    clean_path = str(image_path).lstrip("/")

    parts = clean_path.split("/", 1)

    if len(parts) == 1:

        return "docker.io", clean_path

    registry_candidate = parts[0]

    if "." in registry_candidate or ":" in registry_candidate:

        return registry_candidate, parts[1]

    return "docker.io", clean_path





def _split_image_name_tag(image_ref):

    if not image_ref:

        return None, None

    if "@" in image_ref:

        name, digest = image_ref.split("@", 1)

        return name, digest

    if ":" in image_ref:

        name, tag = image_ref.rsplit(":", 1)

        return name, tag

    return image_ref, "latest"





def _ensure_res_pool(value, fallback_source):

    pattern = r"^[A-Za-z0-9][A-Za-z0-9_]{6,30}[A-Za-z0-9]$"

    if value and re.match(pattern, value):

        return value

    base = re.sub(r"[^A-Za-z0-9_]", "", str(fallback_source or ""))

    candidate = f"respool{base}"[:32]

    if len(candidate) < 8:

        candidate = (candidate + uuid.uuid4().hex)[:32]

    if not re.match(r"^[A-Za-z0-9]", candidate):

        candidate = f"r{candidate}"

    if not re.match(r"[A-Za-z0-9]$", candidate):

        candidate = f"{candidate}0"

    return candidate[:32]





def submit_app(body: dict):

    """

    Controller for submitting application metadata.