Add baseline CI checks (adab6f6a) · Commits · OOP / code / open-exposure-gateway

.gitlab-ci.yml

+31 −23

Original line number	Diff line number	Diff line
		default:
		image: python:3.12-slim
		cache:
		paths:
		- .cache/pip
		before_script:
		- pip install --upgrade pip
		- pip install -e ".[dev,postgres]"

		stages:
		- quality
		- test
		- build
		- push

		before_script:
		- docker info
		variables:
		PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"

		build:
		stage: build
		tags:
		- shell
		quality:
		stage: quality
		script:
		- docker build
		-t $CI_REGISTRY_IMAGE:$CI_COMMIT_TAG
		-t $CI_REGISTRY_IMAGE:latest .
		rules:
		- if: $CI_COMMIT_TAG =~ /^\d+\.\d+\.\d+$/
		- pre-commit run --all-files

		push:
		stage: push
		tags:
		- shell
		needs:
		- build
		test:
		stage: test
		script:
		- pytest \|\| [ $? -eq 5 ]

		build:
		stage: build
		image: docker:29.3.0
		services:
		- docker:29.3.0-dind
		variables:
		DOCKER_TLS_CERTDIR: ""
		script:
		- echo "$CI_REGISTRY_PASSWORD" \| docker login -u "$CI_REGISTRY_USER" $CI_REGISTRY --password-stdin
		- docker push $CI_REGISTRY_IMAGE:$CI_COMMIT_TAG
		- docker push $CI_REGISTRY_IMAGE:latest
		- docker logout $CI_REGISTRY
		- docker build -t oeg-ci-test .
		rules:
		- if: $CI_COMMIT_TAG =~ /^\d+\.\d+\.\d+$/
		No newline at end of file
		- if: '$CI_PIPELINE_SOURCE == "merge_request_event"'
		- if: '$CI_COMMIT_BRANCH == "main"'

.pre-commit-config.yaml

0 → 100644

+26 −0

Original line number	Diff line number	Diff line
		repos:
		- repo: https://github.com/pre-commit/pre-commit-hooks
		rev: v6.0.0
		hooks:
		- id: check-yaml
		- id: check-toml
		- id: end-of-file-fixer
		- id: trailing-whitespace
		args: [--markdown-linebreak-ext=md]

		- repo: https://github.com/astral-sh/ruff-pre-commit
		rev: v0.15.6
		hooks:
		- id: ruff-check
		- id: ruff-format

		- repo: https://github.com/pre-commit/mirrors-mypy
		rev: v1.19.1
		hooks:
		- id: mypy
		files: ^open_exposure_gateway/app/\|^open_exposure_gateway/test/
		additional_dependencies:
		- "fastapi[standard]>=0.135.1"
		- "pydantic>=2.0"
		- "pydantic-settings>=2.0"
		- "httpx>=0.27"

open_exposure_gateway/test/unit/test_domain_models.py

deleted100644 → 0

+0 −483

Original line number	Diff line number	Diff line
		import uuid
		from typing import Any, Literal

		import pytest
		from pydantic import ValidationError

		from open_exposure_gateway.app.api.camara.edge_application_management.ceee3a6.schemas import (
		AppInstanceStatus,
		AppManifest,
		AppRepo,
		ContainerResources,
		DockerComposeResources,
		EdgeCloudZone,
		EdgeCloudZoneStatus,
		KubernetesResources,
		NetworkInterface,
		OperatingSystem,
		VmResources,
		)

		_ZONE_ID = uuid.uuid4()

		_VALID_APP_REPO = AppRepo(
		type="PRIVATEREPO",
		imagePath="https://charts.example.com/nginx:1.0",
		)


		# ---------------------------------------------------------------------------
		# EdgeCloudZone
		# ---------------------------------------------------------------------------


		class TestEdgeCloudZone:
		def test_valid_zone(self) -> None:
		zone = EdgeCloudZone(
		edgeCloudZoneId=_ZONE_ID,
		edgeCloudZoneName="zone-1",
		edgeCloudProvider="acme",
		)
		assert zone.edgeCloudZoneId == _ZONE_ID

		def test_status_defaults_to_unknown(self) -> None:
		zone = EdgeCloudZone(
		edgeCloudZoneId=_ZONE_ID,
		edgeCloudZoneName="z",
		edgeCloudProvider="p",
		)
		assert zone.edgeCloudZoneStatus == EdgeCloudZoneStatus.UNKNOWN

		def test_explicit_status(self) -> None:
		zone = EdgeCloudZone(
		edgeCloudZoneId=_ZONE_ID,
		edgeCloudZoneName="z",
		edgeCloudProvider="p",
		edgeCloudZoneStatus=EdgeCloudZoneStatus.ACTIVE,
		)
		assert zone.edgeCloudZoneStatus == EdgeCloudZoneStatus.ACTIVE

		def test_region_is_optional(self) -> None:
		zone = EdgeCloudZone(
		edgeCloudZoneId=_ZONE_ID,
		edgeCloudZoneName="z",
		edgeCloudProvider="p",
		)
		assert zone.edgeCloudRegion is None

		def test_name_exceeds_max_length_raises(self) -> None:
		with pytest.raises(ValidationError):
		EdgeCloudZone(
		edgeCloudZoneId=_ZONE_ID,
		edgeCloudZoneName="x" * 65,
		edgeCloudProvider="p",
		)

		def test_provider_exceeds_max_length_raises(self) -> None:
		with pytest.raises(ValidationError):
		EdgeCloudZone(
		edgeCloudZoneId=_ZONE_ID,
		edgeCloudZoneName="z",
		edgeCloudProvider="p" * 65,
		)

		def test_invalid_uuid_raises(self) -> None:
		invalid: Any = "not-a-uuid"
		with pytest.raises(ValidationError):
		EdgeCloudZone(
		edgeCloudZoneId=invalid,
		edgeCloudZoneName="z",
		edgeCloudProvider="p",
		)


		# ---------------------------------------------------------------------------
		# NetworkInterface
		# ---------------------------------------------------------------------------


		class TestNetworkInterface:
		def test_valid_interface(self) -> None:
		iface = NetworkInterface(
		interfaceId="eth0",
		protocol="TCP",
		port=80,
		visibilityType="VISIBILITY_EXTERNAL",
		)
		assert iface.interfaceId == "eth0"
		assert iface.port == 80

		@pytest.mark.parametrize("port", [1, 80, 443, 8080, 65535])
		def test_valid_port_range(self, port: int) -> None:
		iface = NetworkInterface(
		interfaceId="eth0",
		protocol="TCP",
		port=port,
		visibilityType="VISIBILITY_INTERNAL",
		)
		assert iface.port == port

		@pytest.mark.parametrize("port", [0, -1, 65536, 99999])
		def test_port_out_of_range_raises(self, port: int) -> None:
		with pytest.raises(ValidationError):
		NetworkInterface(
		interfaceId="eth0",
		protocol="TCP",
		port=port,
		visibilityType="VISIBILITY_EXTERNAL",
		)

		@pytest.mark.parametrize("interface_id", ["eth0", "abcde", "A1b2C3d4"])
		def test_valid_interface_id(self, interface_id: str) -> None:
		iface = NetworkInterface(
		interfaceId=interface_id,
		protocol="UDP",
		port=1234,
		visibilityType="VISIBILITY_EXTERNAL",
		)
		assert iface.interfaceId == interface_id

		@pytest.mark.parametrize(
		"interface_id",
		[
		"ab", # too short (min 4 chars)
		"a" * 33, # too long (max 32 chars)
		"-eth0", # starts with dash
		"eth-", # ends with dash
		],
		)
		def test_invalid_interface_id_raises(self, interface_id: str) -> None:
		with pytest.raises(ValidationError):
		NetworkInterface(
		interfaceId=interface_id,
		protocol="TCP",
		port=80,
		visibilityType="VISIBILITY_EXTERNAL",
		)

		@pytest.mark.parametrize("protocol", ["TCP", "UDP", "ANY"])
		def test_valid_protocols(self, protocol: Literal["TCP", "UDP", "ANY"]) -> None:
		iface = NetworkInterface(
		interfaceId="eth0",
		protocol=protocol,
		port=80,
		visibilityType="VISIBILITY_EXTERNAL",
		)
		assert iface.protocol == protocol

		def test_invalid_protocol_raises(self) -> None:
		invalid: Any = "ICMP"
		with pytest.raises(ValidationError):
		NetworkInterface(
		interfaceId="eth0",
		protocol=invalid,
		port=80,
		visibilityType="VISIBILITY_EXTERNAL",
		)

		@pytest.mark.parametrize("visibility", ["VISIBILITY_EXTERNAL", "VISIBILITY_INTERNAL"])
		def test_valid_visibility_types(
		self, visibility: Literal["VISIBILITY_EXTERNAL", "VISIBILITY_INTERNAL"]
		) -> None:
		iface = NetworkInterface(
		interfaceId="eth0",
		protocol="TCP",
		port=80,
		visibilityType=visibility,
		)
		assert iface.visibilityType == visibility

		def test_invalid_visibility_type_raises(self) -> None:
		invalid: Any = "VISIBILITY_PUBLIC"
		with pytest.raises(ValidationError):
		NetworkInterface(
		interfaceId="eth0",
		protocol="TCP",
		port=80,
		visibilityType=invalid,
		)


		# ---------------------------------------------------------------------------
		# AppRepo
		# ---------------------------------------------------------------------------


		class TestAppRepo:
		def test_private_repo(self) -> None:
		repo = AppRepo(
		type="PRIVATEREPO",
		imagePath="https://example.com/image:1.0",
		userName="user",
		credentials="secret",
		)
		assert repo.type == "PRIVATEREPO"

		def test_public_repo(self) -> None:
		repo = AppRepo(type="PUBLICREPO", imagePath="https://hub.docker.com/nginx")
		assert repo.type == "PUBLICREPO"

		def test_invalid_type_raises(self) -> None:
		invalid: Any = "S3REPO"
		with pytest.raises(ValidationError):
		AppRepo(type=invalid, imagePath="https://example.com")

		def test_image_path_exceeds_max_length_raises(self) -> None:
		with pytest.raises(ValidationError):
		AppRepo(type="PUBLICREPO", imagePath="x" * 2049)

		@pytest.mark.parametrize("auth_type", ["DOCKER", "HTTP_BASIC", "HTTP_BEARER", "NONE"])
		def test_valid_auth_types(
		self, auth_type: Literal["DOCKER", "HTTP_BASIC", "HTTP_BEARER", "NONE"]
		) -> None:
		repo = AppRepo(
		type="PRIVATEREPO",
		imagePath="https://example.com",
		authType=auth_type,
		)
		assert repo.authType == auth_type

		def test_invalid_auth_type_raises(self) -> None:
		invalid: Any = "SSH_KEY"
		with pytest.raises(ValidationError):
		AppRepo(
		type="PRIVATEREPO",
		imagePath="https://example.com",
		authType=invalid,
		)

		def test_optional_fields_default_to_none(self) -> None:
		repo = AppRepo(type="PUBLICREPO", imagePath="https://example.com")
		assert repo.userName is None
		assert repo.credentials is None
		assert repo.authType is None
		assert repo.checksum is None


		# ---------------------------------------------------------------------------
		# AppManifest
		# ---------------------------------------------------------------------------


		class TestAppManifestName:
		@pytest.mark.parametrize("name", ["ab", "nginx_app", "MyApp123"])
		def test_valid_name(self, name: str) -> None:
		manifest = AppManifest(
		name=name,
		version="1.0",
		packageType="HELM",
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)
		assert manifest.name == name

		@pytest.mark.parametrize(
		"name",
		[
		"a", # only 1 char (pattern requires {1,63} after first letter = min 2 total)
		"1nginx", # starts with digit
		"_nginx", # starts with underscore
		"a" * 65, # exceeds max_length=64
		"my-app", # hyphen not allowed by pattern
		],
		)
		def test_invalid_name_raises(self, name: str) -> None:
		with pytest.raises(ValidationError):
		AppManifest(
		name=name,
		version="1.0",
		packageType="HELM",
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)


		class TestAppManifestProvider:
		@pytest.mark.parametrize("provider", ["nginx_inc_x", "Acme_Corp_Ltd"])
		def test_valid_provider(self, provider: str) -> None:
		manifest = AppManifest(
		name="myapp",
		appProvider=provider,
		version="1.0",
		packageType="HELM",
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)
		assert manifest.appProvider == provider

		@pytest.mark.parametrize(
		"provider",
		[
		"short", # too short: pattern requires {7,63} chars after first letter = min 8
		"1provider", # starts with digit
		"a" * 65, # exceeds max_length=64
		],
		)
		def test_invalid_provider_raises(self, provider: str) -> None:
		with pytest.raises(ValidationError):
		AppManifest(
		name="myapp",
		appProvider=provider,
		version="1.0",
		packageType="HELM",
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)

		def test_provider_is_optional(self) -> None:
		manifest = AppManifest(
		name="myapp",
		version="1.0",
		packageType="HELM",
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)
		assert manifest.appProvider is None


		class TestAppManifestPackageType:
		@pytest.mark.parametrize("package_type", ["QCOW2", "OVA", "CONTAINER", "HELM", "CSAR"])
		def test_valid_package_types(
		self, package_type: Literal["QCOW2", "OVA", "CONTAINER", "HELM", "CSAR"]
		) -> None:
		manifest = AppManifest(
		name="myapp",
		version="1.0",
		packageType=package_type,
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)
		assert manifest.packageType == package_type

		def test_invalid_package_type_raises(self) -> None:
		invalid: Any = "DOCKER"
		with pytest.raises(ValidationError):
		AppManifest(
		name="myapp",
		version="1.0",
		packageType=invalid,
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)

		def test_app_id_is_optional(self) -> None:
		manifest = AppManifest(
		name="myapp",
		version="1.0",
		packageType="HELM",
		appRepo=_VALID_APP_REPO,
		componentSpec=[],
		)
		assert manifest.appId is None


		# ---------------------------------------------------------------------------
		# OperatingSystem
		# ---------------------------------------------------------------------------


		class TestOperatingSystem:
		def test_valid_os(self) -> None:
		os = OperatingSystem(
		architecture="x86_64",
		family="UBUNTU",
		version="OS_VERSION_UBUNTU_2204_LTS",
		license="OS_LICENSE_TYPE_FREE",
		)
		assert os.architecture == "x86_64"

		def test_invalid_architecture_raises(self) -> None:
		invalid: Any = "arm64"
		with pytest.raises(ValidationError):
		OperatingSystem(
		architecture=invalid,
		family="UBUNTU",
		version="OS_VERSION_UBUNTU_2204_LTS",
		license="OS_LICENSE_TYPE_FREE",
		)

		def test_invalid_family_raises(self) -> None:
		invalid: Any = "DEBIAN"
		with pytest.raises(ValidationError):
		OperatingSystem(
		architecture="x86_64",
		family=invalid,
		version="OS_VERSION_UBUNTU_2204_LTS",
		license="OS_LICENSE_TYPE_FREE",
		)


		# ---------------------------------------------------------------------------
		# RequiredResources variants
		# ---------------------------------------------------------------------------


		class TestVmResources:
		def test_valid(self) -> None:
		r = VmResources(infraKind="virtualMachine", numCPU=4, memory=8192)
		assert r.infraKind == "virtualMachine"

		def test_cpu_below_min_raises(self) -> None:
		with pytest.raises(ValidationError):
		VmResources(infraKind="virtualMachine", numCPU=0, memory=1024)

		def test_cpu_above_max_raises(self) -> None:
		with pytest.raises(ValidationError):
		VmResources(infraKind="virtualMachine", numCPU=257, memory=1024)

		def test_memory_below_min_raises(self) -> None:
		with pytest.raises(ValidationError):
		VmResources(infraKind="virtualMachine", numCPU=1, memory=0)

		def test_memory_above_max_raises(self) -> None:
		with pytest.raises(ValidationError):
		VmResources(infraKind="virtualMachine", numCPU=1, memory=32769)


		class TestContainerResources:
		@pytest.mark.parametrize("num_cpu", ["1", "2.5", "500m", "0.125"])
		def test_valid_cpu_formats(self, num_cpu: str) -> None:
		r = ContainerResources(infraKind="container", numCPU=num_cpu, memory=512)
		assert r.numCPU == num_cpu

		@pytest.mark.parametrize("num_cpu", ["invalid", "1.5.5", "cpu2", "500k"])
		def test_invalid_cpu_format_raises(self, num_cpu: str) -> None:
		with pytest.raises(ValidationError):
		ContainerResources(infraKind="container", numCPU=num_cpu, memory=512)


		class TestDockerComposeResources:
		def test_valid(self) -> None:
		r = DockerComposeResources(infraKind="dockerCompose", numCPU=2, memory=2048)
		assert r.infraKind == "dockerCompose"


		class TestKubernetesResources:
		def test_valid(self) -> None:
		r = KubernetesResources(
		infraKind="kubernetes",
		applicationResources={"cpuPool": {"numCPU": 2}},
		isStandalone=True,
		)
		assert r.isStandalone is True


		# ---------------------------------------------------------------------------
		# Enums
		# ---------------------------------------------------------------------------


		class TestEdgeCloudZoneStatus:
		def test_values(self) -> None:
		assert EdgeCloudZoneStatus.ACTIVE.value == "active"
		assert EdgeCloudZoneStatus.INACTIVE.value == "inactive"
		assert EdgeCloudZoneStatus.UNKNOWN.value == "unknown"


		class TestAppInstanceStatus:
		def test_values(self) -> None:
		assert AppInstanceStatus.READY.value == "ready"
		assert AppInstanceStatus.INSTANTIATING.value == "instantiating"
		assert AppInstanceStatus.FAILED.value == "failed"
		assert AppInstanceStatus.TERMINATING.value == "terminating"
		assert AppInstanceStatus.UNKNOWN.value == "unknown"