From 40b6d1c0300945705c86229d932108b777c1b9dd Mon Sep 17 00:00:00 2001 From: Patricia Reinoso Date: Mon, 2 Dec 2024 11:00:39 +0100 Subject: [PATCH] feat: extend fiveg_f1 interface library (#53) --- lib/charms/oai_ran_cu_k8s/v0/fiveg_f1.py | 310 ++++++++---------- src/charm.py | 13 +- .../test_provider_charm/src/charm.py | 38 ++- .../test_requirer_charm/src/charm.py | 32 +- .../v0/test_fiveg_f1_provider.py | 231 +++++++++---- .../v0/test_fiveg_f1_requirer.py | 154 ++++++--- tests/unit/test_charm_configure.py | 8 + 7 files changed, 478 insertions(+), 308 deletions(-) diff --git a/lib/charms/oai_ran_cu_k8s/v0/fiveg_f1.py b/lib/charms/oai_ran_cu_k8s/v0/fiveg_f1.py index 36e2611..c239660 100644 --- a/lib/charms/oai_ran_cu_k8s/v0/fiveg_f1.py +++ b/lib/charms/oai_ran_cu_k8s/v0/fiveg_f1.py @@ -28,16 +28,18 @@ Example: ```python -from ops.charm import CharmBase, RelationJoinedEvent -from ops.main import main +from ops import main +from ops.charm import CharmBase, RelationChangedEvent, RelationJoinedEvent -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Provides +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Provides, PLMNConfig class DummyFivegF1ProviderCharm(CharmBase): IP_ADDRESS = "192.168.70.132" PORT = 2153 + TAC = 1 + PLMNS = [PLMNConfig(mcc="123", mnc="12", sst=1, sd=1)] def __init__(self, *args): super().__init__(*args) @@ -45,14 +47,24 @@ def __init__(self, *args): self.framework.observe( self.on.fiveg_f1_relation_joined, self._on_fiveg_f1_relation_joined ) + self.framework.observe( + self.on.fiveg_f1_relation_changed, self._on_fiveg_f1_relation_changed + ) def _on_fiveg_f1_relation_joined(self, event: RelationJoinedEvent): if self.unit.is_leader(): self.f1_provider.set_f1_information( ip_address=self.IP_ADDRESS, port=self.PORT, + tac=self.TAC, + plmns=self.PLMNS, ) + def _on_fiveg_f1_relation_changed(self, event: RelationChangedEvent): + requirer_f1_port = self.f1_provider.requirer_f1_port + if requirer_f1_port: + + if __name__ == "__main__": main(DummyFivegF1ProviderCharm) @@ -65,12 +77,10 @@ def _on_fiveg_f1_relation_joined(self, event: RelationJoinedEvent): Example: ```python -from ops.charm import CharmBase -from ops.main import main - -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import FivegF1ProviderAvailableEvent, F1Requires +from ops import main +from ops.charm import CharmBase, RelationChangedEvent, RelationJoinedEvent -logger = logging.getLogger(__name__) +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Requires class DummyFivegF1Requires(CharmBase): @@ -84,17 +94,19 @@ def __init__(self, *args): self.on.fiveg_f1_relation_joined, self._on_fiveg_f1_relation_joined ) self.framework.observe( - self.f1_requirer.on.fiveg_f1_provider_available, self._on_f1_information_available + self.on.fiveg_f1_relation_changed, self._on_fiveg_f1_relation_changed ) def _on_fiveg_f1_relation_joined(self, event: RelationJoinedEvent): if self.unit.is_leader(): self.f1_requirer.set_f1_information(port=self.PORT) - def _on_f1_information_available(self, event: FivegF1ProviderAvailableEvent): - provider_f1_ip_address = event.f1_ip_address - provider_f1_port = event.f1_port - + def _on_fiveg_f1_relation_changed(self, event: RelationChangedEvent): + provider_f1_ip_address = self.f1_requirer.f1_ip_address + provider_f1_port = self.f1_requirer.f1_port + provider_f1_tac = self.f1_requirer.tac + provider_f1_plmn = self.f1_requirer.plmn + if __name__ == "__main__": @@ -103,14 +115,17 @@ def _on_f1_information_available(self, event: FivegF1ProviderAvailableEvent): """ +import json import logging -from typing import Dict, Optional, cast +from dataclasses import dataclass +from json.decoder import JSONDecodeError +from typing import Any, Dict, Optional from interface_tester.schema_base import DataBagSchema -from ops.charm import CharmBase, CharmEvents, RelationChangedEvent, RelationJoinedEvent -from ops.framework import EventBase, EventSource, Handle, Object +from ops.charm import CharmBase +from ops.framework import Object from ops.model import Relation -from pydantic import BaseModel, Field, IPvAnyAddress, ValidationError +from pydantic import BaseModel, Field, IPvAnyAddress, ValidationError, conlist # The unique Charmhub library identifier, never change it LIBID = "544f1e90a3bd49c68d523c506e383579" @@ -120,7 +135,7 @@ def _on_f1_information_available(self, event: FivegF1ProviderAvailableEvent): # Increment this PATCH version before using `charmcraft publish-lib` or reset # to 0 if you are raising the major API version -LIBPATCH = 2 +LIBPATCH = 3 logger = logging.getLogger(__name__) @@ -133,7 +148,16 @@ def _on_f1_information_available(self, event: FivegF1ProviderAvailableEvent): unit: app: { "f1_ip_address": "192.168.70.132" - "f1_port": 2153 + "f1_port": 2153, + "tac": 1, + "plmns": [ + { + "mcc": "001", + "mnc": "01", + "sst": 1, + "sd": 1, + } + ], } RequirerSchema: unit: @@ -143,6 +167,42 @@ def _on_f1_information_available(self, event: FivegF1ProviderAvailableEvent): """ +@dataclass +class PLMNConfig(BaseModel): + """Dataclass representing the configuration for a PLMN.""" + + def __init__(self, mcc: str, mnc: str, sst: int, sd: Optional[int] = None) -> None: + super().__init__(mcc=mcc, mnc=mnc, sst=sst, sd=sd) + + mcc: str = Field( + description="Mobile Country Code", + examples=["001", "208", "302"], + pattern=r"^[0-9][0-9][0-9]$", + ) + mnc: str = Field( + description="Mobile Network Code", + examples=["01", "001", "999"], + pattern=r"^[0-9][0-9][0-9]?$", + ) + sst: int = Field( + description="Slice/Service Type", + examples=[1, 2, 3, 4], + ge=0, + le=255, + ) + sd: Optional[int] = Field( + description="Slice Differentiator", + default=None, + examples=[1], + ge=0, + le=16777215, + ) + + def asdict(self): + """Convert the dataclass into a dictionary.""" + return {"mcc": self.mcc, "mnc": self.mnc, "sst": self.sst, "sd": self.sd} + + class ProviderAppData(BaseModel): """Provider app data for fiveg_f1.""" @@ -154,6 +214,14 @@ class ProviderAppData(BaseModel): description="Number of the port used for F1 traffic.", examples=[2153], ) + tac: int = Field( + description="Tracking Area Code", + strict=True, + examples=[1], + ge=1, + le=16777215, + ) + plmns: conlist(PLMNConfig, min_length=1) # type: ignore[reportInvalidTypeForm] class ProviderSchema(DataBagSchema): @@ -211,96 +279,6 @@ def requirer_data_is_valid(data: dict) -> bool: return False -class FivegF1ProviderAvailableEvent(EventBase): - """Charm event emitted when the F1 provider info is available. - - The event carries the F1 provider's IP address and port. - """ - - def __init__(self, handle: Handle, f1_ip_address: str, f1_port: int): - """Init.""" - super().__init__(handle) - self.f1_ip_address = f1_ip_address - self.f1_port = f1_port - - def snapshot(self) -> dict: - """Return snapshot.""" - return { - "f1_ip_address": self.f1_ip_address, - "f1_port": self.f1_port, - } - - def restore(self, snapshot: dict) -> None: - """Restores snapshot.""" - self.f1_ip_address = snapshot["f1_ip_address"] - self.f1_port = snapshot["f1_port"] - - -class FivegF1RequestEvent(EventBase): - """Charm event emitted when the F1 requirer joins.""" - - def __init__(self, handle: Handle, relation_id: int): - """Set relation id. - - Args: - handle (Handle): Juju framework handle. - relation_id : ID of the relation. - """ - super().__init__(handle) - self.relation_id = relation_id - - def snapshot(self) -> dict: - """Return event data. - - Returns: - (dict): contains the relation ID. - """ - return { - "relation_id": self.relation_id, - } - - def restore(self, snapshot: dict) -> None: - """Restore event data. - - Args: - snapshot (dict): contains the relation ID. - """ - self.relation_id = snapshot["relation_id"] - - -class FivegF1RequirerAvailableEvent(EventBase): - """Charm event emitted when the F1 requirer info is available. - - The event carries the F1 requirer's port. - """ - - def __init__(self, handle: Handle, f1_port: int): - """Init.""" - super().__init__(handle) - self.f1_port = f1_port - - def snapshot(self) -> dict: - """Return snapshot.""" - return {"f1_port": self.f1_port} - - def restore(self, snapshot: dict) -> None: - """Restores snapshot.""" - self.f1_port = snapshot["f1_port"] - - -class FivegF1ProviderCharmEvents(CharmEvents): - """List of events that the F1 provider charm can leverage.""" - - fiveg_f1_request = EventSource(FivegF1RequestEvent) - fiveg_f1_requirer_available = EventSource(FivegF1RequirerAvailableEvent) - - -class FivegF1RequirerCharmEvents(CharmEvents): - """List of events that the F1 requirer charm can leverage.""" - - fiveg_f1_provider_available = EventSource(FivegF1ProviderAvailableEvent) - - class FivegF1Error(Exception): """Custom error class for the `fiveg_f1` library.""" @@ -312,52 +290,44 @@ def __init__(self, message: str): class F1Provides(Object): """Class to be instantiated by the charm providing relation using the `fiveg_f1` interface.""" - on = FivegF1ProviderCharmEvents() # type: ignore - def __init__(self, charm: CharmBase, relation_name: str): """Init.""" super().__init__(charm, relation_name) self.relation_name = relation_name self.charm = charm - self.framework.observe(charm.on[relation_name].relation_joined, self._on_relation_joined) - self.framework.observe(charm.on[relation_name].relation_changed, self._on_relation_changed) - - def _on_relation_joined(self, event: RelationJoinedEvent) -> None: - """Handle relation joined event. - - Args: - event (RelationJoinedEvent): Juju event. - """ - self.on.fiveg_f1_request.emit(relation_id=event.relation.id) - - def _on_relation_changed(self, event: RelationChangedEvent) -> None: - """Handle relation changed event. - Args: - event (RelationChangedEvent): Juju event. - """ - if remote_app_relation_data := self._get_remote_app_relation_data(event.relation): - self.on.fiveg_f1_requirer_available.emit(f1_port=remote_app_relation_data["f1_port"]) - - def set_f1_information(self, ip_address: str, port: int) -> None: + def set_f1_information( + self, ip_address: str, port: int, tac: int, plmns: list[PLMNConfig] + ) -> None: """Push the information about the F1 interface in the application relation data. Args: ip_address (str): IPv4 address of the network interface used for F1 traffic. port (int): Number of the port used for F1 traffic. + tac (int): Tracking Area Code. + plmns (list[PLMNConfig]): Configured PLMNs. """ if not self.charm.unit.is_leader(): raise FivegF1Error("Unit must be leader to set application relation data.") relations = self.model.relations[self.relation_name] if not relations: raise FivegF1Error(f"Relation {self.relation_name} not created yet.") - if not provider_data_is_valid({"f1_ip_address": ip_address, "f1_port": port}): + if not provider_data_is_valid( + { + "f1_ip_address": ip_address, + "f1_port": port, + "tac": tac, + "plmns": plmns, + } + ): raise FivegF1Error("Invalid relation data") for relation in relations: relation.data[self.charm.app].update( { "f1_ip_address": ip_address, "f1_port": str(port), + "tac": str(tac), + "plmns": json.dumps([plmn.asdict() for plmn in plmns]), } ) @@ -366,22 +336,22 @@ def requirer_f1_port(self) -> Optional[int]: """Return the number of the port used for F1 traffic. Returns: - int: Port number. + Optional[int]: Port number. """ if remote_app_relation_data := self._get_remote_app_relation_data(): - return cast(Optional[int], remote_app_relation_data.get("f1_port")) + return remote_app_relation_data.f1_port return None def _get_remote_app_relation_data( self, relation: Optional[Relation] = None - ) -> Optional[Dict[str, str]]: + ) -> Optional[RequirerAppData]: """Get relation data for the remote application. Args: relation: Juju relation object (optional). Returns: - Dict: Relation data for the remote application or None if the relation data is invalid. + RequirerAppData: Relation data for the remote application if valid, None otherwise. """ relation = relation or self.model.get_relation(self.relation_name) if not relation: @@ -390,36 +360,23 @@ def _get_remote_app_relation_data( if not relation.app: logger.warning("No remote application in relation: %s", self.relation_name) return None - remote_app_relation_data = dict(relation.data[relation.app]) - if not requirer_data_is_valid(remote_app_relation_data): + remote_app_relation_data: Dict[str, Any] = dict(relation.data[relation.app]) + try: + requirer_app_data = RequirerAppData(**remote_app_relation_data) + except ValidationError: logger.error("Invalid relation data: %s", remote_app_relation_data) return None - return remote_app_relation_data + return requirer_app_data class F1Requires(Object): """Class to be instantiated by the charm requiring relation using the `fiveg_f1` interface.""" - on = FivegF1RequirerCharmEvents() # type: ignore - def __init__(self, charm: CharmBase, relation_name: str): """Init.""" super().__init__(charm, relation_name) self.charm = charm self.relation_name = relation_name - self.framework.observe(charm.on[relation_name].relation_changed, self._on_relation_changed) - - def _on_relation_changed(self, event: RelationChangedEvent) -> None: - """Handle relation changed event. - - Args: - event (RelationChangedEvent): Juju event. - """ - if remote_app_relation_data := self._get_remote_app_relation_data(event.relation): - self.on.fiveg_f1_provider_available.emit( - f1_ip_address=remote_app_relation_data["f1_ip_address"], - f1_port=remote_app_relation_data["f1_port"], - ) def set_f1_information(self, port: int) -> None: """Push the information about the F1 interface in the application relation data. @@ -437,38 +394,16 @@ def set_f1_information(self, port: int) -> None: for relation in relations: relation.data[self.charm.app].update({"f1_port": str(port)}) - @property - def f1_ip_address(self) -> Optional[str]: - """Return IPv4 address of the network interface used for F1 traffic. - - Returns: - str: IPv4 address. - """ - if remote_app_relation_data := self._get_remote_app_relation_data(): - return remote_app_relation_data.get("f1_ip_address") - return None - - @property - def f1_port(self) -> Optional[int]: - """Return the number of the port used for F1 traffic. - - Returns: - int: Port number. - """ - if remote_app_relation_data := self._get_remote_app_relation_data(): - return cast(Optional[int], remote_app_relation_data.get("f1_port")) - return None - - def _get_remote_app_relation_data( + def get_provider_f1_information( self, relation: Optional[Relation] = None - ) -> Optional[Dict[str, str]]: + ) -> Optional[ProviderAppData]: """Get relation data for the remote application. Args: relation: Juju relation object (optional). Returns: - Dict: Relation data for the remote application or None if the relation data is invalid. + ProviderAppData: Relation data for the remote application if valid, None otherwise. """ relation = relation or self.model.get_relation(self.relation_name) if not relation: @@ -477,8 +412,19 @@ def _get_remote_app_relation_data( if not relation.app: logger.warning("No remote application in relation: %s", self.relation_name) return None - remote_app_relation_data = dict(relation.data[relation.app]) - if not provider_data_is_valid(remote_app_relation_data): + remote_app_relation_data: Dict[str, Any] = dict(relation.data[relation.app]) + remote_plmns = remote_app_relation_data.get("plmns", "") + try: + remote_app_relation_data["tac"] = int(remote_app_relation_data.get("tac", "")) + remote_app_relation_data["plmns"] = [ + PLMNConfig(**data) for data in json.loads(remote_plmns) + ] + except (JSONDecodeError, ValidationError, ValueError): + logger.error("Invalid relation data: %s", remote_app_relation_data) + return None + try: + provider_app_data = ProviderAppData(**remote_app_relation_data) + except ValidationError: logger.error("Invalid relation data: %s", remote_app_relation_data) return None - return remote_app_relation_data + return provider_app_data diff --git a/src/charm.py b/src/charm.py index fab9dae..bc38364 100755 --- a/src/charm.py +++ b/src/charm.py @@ -16,7 +16,7 @@ NetworkAttachmentDefinition, ) from charms.loki_k8s.v1.loki_push_api import LogForwarder -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Provides +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Provides, PLMNConfig from charms.sdcore_amf_k8s.v0.fiveg_n2 import N2Requires from charms.sdcore_gnbsim_k8s.v0.fiveg_gnb_identity import ( GnbIdentityProvides, @@ -40,6 +40,8 @@ DU_F1_DEFAULT_PORT = 2152 WORKLOAD_VERSION_FILE_NAME = "/etc/workload-version" LOGGING_RELATION_NAME = "logging" +HARDCODED_PLMNS = [PLMNConfig(mcc="001", mnc="01", sst=1, sd=12)] +HARDCODED_TAC = 1 class OAIRANCUOperator(CharmBase): @@ -79,8 +81,8 @@ def __init__(self, *args): self.framework.observe(self.on.cu_pebble_ready, self._configure) self.framework.observe(self.on.fiveg_n2_relation_joined, self._configure) self.framework.observe(self._n2_requirer.on.n2_information_available, self._configure) - self.framework.observe(self._f1_provider.on.fiveg_f1_request, self._configure) - self.framework.observe(self._f1_provider.on.fiveg_f1_requirer_available, self._configure) + self.framework.observe(self.on[F1_RELATION_NAME].relation_joined, self._configure) + self.framework.observe(self.on[F1_RELATION_NAME].relation_changed, self._configure) self.framework.observe( self._gnb_identity_provider.on.fiveg_gnb_identity_request, self._configure, @@ -407,7 +409,10 @@ def _update_fiveg_f1_relation_data(self) -> None: logger.error("F1 IP address is not available") return self._f1_provider.set_f1_information( - ip_address=f1_ip.split("/")[0], port=self._charm_config.f1_port + ip_address=f1_ip.split("/")[0], + port=self._charm_config.f1_port, + tac=HARDCODED_TAC, + plmns=HARDCODED_PLMNS, ) def _exec_command_in_workload_container( diff --git a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_provider_charm/src/charm.py b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_provider_charm/src/charm.py index 3710398..3e075f9 100644 --- a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_provider_charm/src/charm.py +++ b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_provider_charm/src/charm.py @@ -1,11 +1,12 @@ # Copyright 2024 Canonical Ltd. # See LICENSE file for licensing details. +import json import logging -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Provides +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Provides, PLMNConfig +from ops import main from ops.charm import ActionEvent, CharmBase -from ops.main import main logger = logging.getLogger(__name__) @@ -18,11 +19,42 @@ def __init__(self, *args): self.framework.observe( self.on.set_f1_information_action, self._on_set_f1_information_action ) + self.framework.observe( + self.on.set_f1_information_as_string_action, + self._on_set_f1_information_as_string_action, + ) + self.framework.observe( + self.on.get_f1_information_action, self._on_get_f1_information_action + ) def _on_set_f1_information_action(self, event: ActionEvent): ip_address = event.params.get("ip-address", "") port = event.params.get("port", "") - self.fiveg_f1_provider.set_f1_information(ip_address=ip_address, port=port) + tac = event.params.get("tac", "") + plmns = event.params.get("plmns", "") + self.fiveg_f1_provider.set_f1_information( + ip_address=ip_address, + port=port, + tac=int(tac), + plmns=[PLMNConfig(**data) for data in json.loads(plmns)], + ) + + def _on_set_f1_information_as_string_action(self, event: ActionEvent): + ip_address = event.params.get("ip-address", "") + port = event.params.get("port", "") + tac = event.params.get("tac", "") + plmns = event.params.get("plmns", "") + self.fiveg_f1_provider.set_f1_information( + ip_address=ip_address, + port=port, + tac=tac, + plmns=[PLMNConfig(**data) for data in json.loads(plmns)], + ) + + def _on_get_f1_information_action(self, event: ActionEvent): + port_value = event.params.get("expected_port") + expected_port = int(port_value) if port_value else None + assert expected_port == self.fiveg_f1_provider.requirer_f1_port if __name__ == "__main__": diff --git a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_requirer_charm/src/charm.py b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_requirer_charm/src/charm.py index 83da566..671c758 100644 --- a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_requirer_charm/src/charm.py +++ b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_charms/test_requirer_charm/src/charm.py @@ -1,10 +1,11 @@ # Copyright 2024 Canonical Ltd. # See LICENSE file for licensing details. +import json -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Requires +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import F1Requires, PLMNConfig, ProviderAppData +from ops import main from ops.charm import ActionEvent, CharmBase -from ops.main import main class WhateverCharm(CharmBase): @@ -13,14 +14,37 @@ def __init__(self, *args): super().__init__(*args) self.fiveg_f1_requirer = F1Requires(self, "fiveg_f1") self.framework.observe( - self.on.set_f1_information_action, - self._on_set_f1_information_action, + self.on.set_f1_information_action, self._on_set_f1_information_action + ) + self.framework.observe( + self.on.get_f1_information_action, self._on_get_f1_information_action + ) + self.framework.observe( + self.on.get_f1_information_invalid_action, self._on_get_f1_information_action_invalid ) def _on_set_f1_information_action(self, event: ActionEvent): port = event.params.get("port", "") self.fiveg_f1_requirer.set_f1_information(port=port) + def _on_get_f1_information_action(self, event: ActionEvent): + ip_address = event.params.get("expected_ip_address", "") + port = event.params.get("expected_port", "") + tac = event.params.get("expected_tac", "") + plmns = event.params.get("expected_plmns", "") + validated_data = { + "f1_ip_address": ip_address, + "f1_port": port, + "tac": int(tac), + "plmns": [PLMNConfig(**data) for data in json.loads(plmns)], + } + provider_app_data = ProviderAppData(**validated_data) + + assert provider_app_data == self.fiveg_f1_requirer.get_provider_f1_information() + + def _on_get_f1_information_action_invalid(self, event: ActionEvent): + assert self.fiveg_f1_requirer.get_provider_f1_information() is None + if __name__ == "__main__": main(WhateverCharm) diff --git a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_provider.py b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_provider.py index 1f3ba50..aa0ee7d 100644 --- a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_provider.py +++ b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_provider.py @@ -1,26 +1,22 @@ # Copyright 2024 Canonical Ltd. # See LICENSE file for licensing details. -from unittest.mock import patch +import json import pytest -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import FivegF1RequirerAvailableEvent +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import PLMNConfig from ops import testing +from pydantic import ValidationError from tests.unit.lib.charms.oai_ran_cu_k8s.v0.test_charms.test_provider_charm.src.charm import ( WhateverCharm, ) +VALID_IP = "1.2.3.4" +VALID_PLMN = PLMNConfig(mcc="123", mnc="12", sst=1, sd=12) -class TestFivegF1Provides: - @pytest.fixture(autouse=True) - def setUp(self, request): - yield - request.addfinalizer(self.tearDown) - - def tearDown(self) -> None: - patch.stopall() +class TestFivegF1Provides: @pytest.fixture(autouse=True) def context(self): self.ctx = testing.Context( @@ -31,86 +27,193 @@ def context(self): }, actions={ "set-f1-information": { - "params": {"ip-address": {"type": "string"}, "port": {"type": "string"}} - } + "params": { + "ip-address": {"type": "string"}, + "port": {"type": "string"}, + "tac": {"type": "string"}, + "plmns": {"type": "string"}, + } + }, + "set-f1-information-as-string": { + "params": { + "ip-address": {"type": "string"}, + "port": {"type": "string"}, + "tac": {"type": "string"}, + "plmns": {"type": "string"}, + } + }, + "get-f1-information": { + "params": { + "expected_port": {"type": "string"}, + } + }, }, ) - def test_given_valid_f1_interface_data_when_set_f1_information_then_f1_ip_address_and_port_are_pushed_to_the_relation_databag( # noqa: E501 - self, + @pytest.mark.parametrize( + "plmns", + [ + pytest.param([VALID_PLMN], id="sd_is_present"), + pytest.param([PLMNConfig(mcc="123", mnc="12", sst=1)], id="sd_is_none"), + pytest.param([VALID_PLMN, VALID_PLMN], id="many_lists"), + ], + ) + def test_given_valid_f1_interface_data_when_set_f1_information_then_f1_ip_address_port_tac_and_plmns_are_pushed_to_the_relation_databag( # noqa: E501 + self, plmns ): - fiveg_f1_relation = testing.Relation( - endpoint="fiveg_f1", - interface="fiveg_f1", - ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, - ) - params = { - "ip-address": "1.2.3.4", - "port": "1234", - } + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + plmns_as_string = json.dumps([plmn.asdict() for plmn in plmns]) + params = {"ip-address": VALID_IP, "port": "1234", "tac": "12", "plmns": plmns_as_string} state_out = self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) relation = state_out.get_relation(fiveg_f1_relation.id) - assert relation.local_app_data["f1_ip_address"] == "1.2.3.4" + assert relation.local_app_data["f1_ip_address"] == VALID_IP assert relation.local_app_data["f1_port"] == "1234" + assert relation.local_app_data["tac"] == "12" + assert relation.local_app_data["plmns"] == plmns_as_string - def test_given_invalid_f1_ip_address_when_set_f1_information_then_error_is_raised(self): - fiveg_f1_relation = testing.Relation( - endpoint="fiveg_f1", - interface="fiveg_f1", - ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, - ) - params = { - "ip-address": "1111.1111.1111.1111", - "port": "1234", - } + def test_given_charm_is_not_leader_when_set_f1_information_then_error_is_raised(self): + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=False) + plmns_as_string = json.dumps([plmn.asdict() for plmn in [VALID_PLMN]]) + params = {"ip-address": VALID_IP, "port": "3", "tac": "123", "plmns": plmns_as_string} + + with pytest.raises(Exception) as e: + self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) + + assert "Unit must be leader to set application relation data." in str(e.value) + + def test_given_f1_relation_does_not_exist_when_set_f1_information_then_error_is_raised(self): + state_in = testing.State(relations=[], leader=True) + plmns_as_string = json.dumps([plmn.asdict() for plmn in [VALID_PLMN]]) + params = {"ip-address": VALID_IP, "port": "3", "tac": "123", "plmns": plmns_as_string} + + with pytest.raises(Exception) as e: + self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) + + assert "Relation fiveg_f1 not created yet." in str(e.value) + + @pytest.mark.parametrize( + "tac", + [ + pytest.param("0", id="too_small_tac"), + pytest.param("16777216", id="too_big_tac"), + ], + ) + def test_given_invalid_range_tac_when_set_f1_information_then_error_is_raised(self, tac): + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + plmns_as_string = json.dumps([plmn.asdict() for plmn in [VALID_PLMN]]) + params = {"ip-address": VALID_IP, "port": "3", "tac": tac, "plmns": plmns_as_string} with pytest.raises(Exception) as e: self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) assert "Invalid relation data" in str(e.value) - def test_given_invalid_f1_port_when_set_f1_information_then_error_is_raised(self): - fiveg_f1_relation = testing.Relation( - endpoint="fiveg_f1", - interface="fiveg_f1", - ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, - ) - params = { - "ip-address": "1.2.3.4", - "port": "that's wrong", - } + def test_given_empty_plmns_list_when_set_f1_information_then_error_is_raised(self): + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + params = {"ip-address": VALID_IP, "port": "3", "tac": "23", "plmns": "[]"} with pytest.raises(Exception) as e: self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) assert "Invalid relation data" in str(e.value) - def test_given_fiveg_f1_relation_created_when_relation_changed_then_event_with_requirer_f1_port_is_emitted( # noqa: E501 - self, + @pytest.mark.parametrize( + "ip_address,port,tac", + [ + pytest.param("1111.1111.1111.1111", "1234", "12", id="invalid_ip_address"), + pytest.param("", "1234", "12", id="empty_ip_address"), + pytest.param(VALID_IP, "port", "12", id="invalid_port"), + pytest.param(VALID_IP, "", "12", id="empty_port"), + pytest.param(VALID_IP, "12", "tac", id="invalid_tac"), + pytest.param(VALID_IP, "12", "", id="empty_tac"), + ], + ) + def test_given_invalid_string_format_ip_address_port_or_tac_when_set_f1_information_then_error_is_raised( # noqa: E501 + self, ip_address, port, tac + ): + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + plmns_as_string = json.dumps([plmn.asdict() for plmn in [VALID_PLMN]]) + params = {"ip-address": ip_address, "port": port, "tac": tac, "plmns": plmns_as_string} + + with pytest.raises(Exception) as e: + self.ctx.run( + self.ctx.on.action("set-f1-information-as-string", params=params), + state_in, + ) + + assert "Invalid relation data" in str(e.value) + + @pytest.mark.parametrize( + "remote_port,expected_port", + [ + pytest.param("1234", "1234", id="valid_port"), + pytest.param("invalid", "", id="invalid_port"), + ], + ) + def test_given_remote_app_filled_databag_when_get_f1_information_then_value_is_retrieved( # noqa: E501 + self, remote_port, expected_port ): fiveg_f1_relation = testing.Relation( endpoint="fiveg_f1", interface="fiveg_f1", - remote_app_data={"f1_ip_address": "1.2.3.4", "f1_port": "1234"}, - ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, + remote_app_data={"f1_port": remote_port}, ) + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + params = {"expected_port": expected_port} - self.ctx.run(self.ctx.on.relation_changed(fiveg_f1_relation), state_in) + self.ctx.run(self.ctx.on.action("get-f1-information", params=params), state_in) - assert len(self.ctx.emitted_events) == 2 - assert isinstance(self.ctx.emitted_events[1], FivegF1RequirerAvailableEvent) - assert self.ctx.emitted_events[1].f1_port == "1234" + def test_given_f1_relation_does_not_exist_when_get_f1_information_then_none_value_is_retrieved( # noqa: E501 + self, + ): + state_in = testing.State(relations=[], leader=True) + params = {"expected_port": ""} + self.ctx.run(self.ctx.on.action("get-f1-information", params=params), state_in) + + @pytest.mark.parametrize( + "mcc,mnc,sst,sd", + [ + pytest.param(None, "01", 2, 3, id="None_mcc"), + pytest.param("mcc", "01", 2, 3, id="string_mcc"), + pytest.param("01", "01", 2, 3, id="2_character_mcc"), + pytest.param("0122", "01", 2, 3, id="4_character_mcc"), + pytest.param("001", None, 2, 3, id="None_mnc"), + pytest.param("001", "mnc", 2, 3, id="string_mnc"), + pytest.param("001", "1", 2, 3, id="1_character_mnc"), + pytest.param("001", "1234", 2, 3, id="4_character_mnc"), + pytest.param("001", "01", None, 3, id="None_sst"), + pytest.param("001", "01", "sst", 3, id="string_sst"), + pytest.param("001", "01", -1, 2, id="too_small_sst"), + pytest.param("001", "01", 256, 2, id="too_big_sst"), + pytest.param("001", "01", 2, "sd", id="string_sd"), + pytest.param("001", "01", 2, -1, id="too_small_sd"), + pytest.param("001", "01", 2, 16777216, id="too_big_sd"), + ], + ) + def test_given_invalid_plmns_then_error_is_raised_at_construction(self, mcc, mnc, sst, sd): + with pytest.raises(ValidationError) as e: + PLMNConfig(mcc=mcc, mnc=mnc, sst=sst, sd=sd) + + assert "1 validation error for PLMNConfig" in str(e.value) + + @pytest.mark.parametrize( + "mcc,mnc,sst,sd", + [ + pytest.param("201", "01", 2, 3, id="2_character_nmc"), + pytest.param("405", "011", 2, 3, id="3_character_mcc"), + pytest.param("455", "123", 0, 3, id="smallest_sst"), + pytest.param("735", "255", 255, 3, id="biggest_sst"), + pytest.param("135", "123", 2, 0, id="smallest_sd"), + pytest.param("863", "01", 3, 16777215, id="biggest_sd"), + pytest.param("245", "01", 3, None, id="None_sd"), + ], + ) + def test_given_valid_plmns_then_error_is_not_raised_at_construction(self, mcc, mnc, sst, sd): + PLMNConfig(mcc=mcc, mnc=mnc, sst=sst, sd=sd) diff --git a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_requirer.py b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_requirer.py index 6fb12aa..b01223d 100644 --- a/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_requirer.py +++ b/tests/unit/lib/charms/oai_ran_cu_k8s/v0/test_fiveg_f1_requirer.py @@ -1,26 +1,21 @@ # Copyright 2024 Canonical Ltd. # See LICENSE file for licensing details. -from unittest.mock import patch + +import json import pytest -from charms.oai_ran_cu_k8s.v0.fiveg_f1 import FivegF1ProviderAvailableEvent +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import PLMNConfig from ops import testing from tests.unit.lib.charms.oai_ran_cu_k8s.v0.test_charms.test_requirer_charm.src.charm import ( WhateverCharm, ) +VALID_PLMN = PLMNConfig(mcc="123", mnc="12", sst=1, sd=12) -class TestFivegF1Requires: - @pytest.fixture(autouse=True) - def setUp(self, request): - yield - request.addfinalizer(self.tearDown) - - def tearDown(self) -> None: - patch.stopall() +class TestFivegF1Requires: @pytest.fixture(autouse=True) def context(self): self.ctx = testing.Context( @@ -29,66 +24,123 @@ def context(self): "name": "whatever-charm", "requires": {"fiveg_f1": {"interface": "fiveg_f1"}}, }, - actions={"set-f1-information": {"params": {"port": {"type": "string"}}}}, + actions={ + "set-f1-information": {"params": {"port": {"type": "string"}}}, + "get-f1-information": { + "params": { + "expected_ip_address": {"type": "string"}, + "expected_port": {"type": "string"}, + "expected_tac": {"type": "string"}, + "expected_plmns": {"type": "string"}, + } + }, + "get-f1-information-invalid": {"params": {}}, + }, ) - def test_given_fiveg_f1_relation_created_when_relation_changed_then_event_with_provider_f1_ip_address_and_port_is_emitted( # noqa: E501 + def test_given_valid_f1_port_when_set_f1_information_then_requirer_f1_port_is_pushed_to_the_relation_databag( # noqa: E501 self, ): - fiveg_f1_relation = testing.Relation( - endpoint="fiveg_f1", - interface="fiveg_f1", - remote_app_data={ - "f1_ip_address": "1.2.3.4", - "f1_port": "1234", - }, - ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, - ) + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + params = {"port": "1234"} - self.ctx.run(self.ctx.on.relation_changed(fiveg_f1_relation), state_in) + state_out = self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) - assert len(self.ctx.emitted_events) == 2 - assert isinstance(self.ctx.emitted_events[1], FivegF1ProviderAvailableEvent) - assert self.ctx.emitted_events[1].f1_ip_address == "1.2.3.4" - assert self.ctx.emitted_events[1].f1_port == "1234" + relation = state_out.get_relation(fiveg_f1_relation.id) + assert relation.local_app_data["f1_port"] == "1234" - def test_given_valid_f1_port_when_set_f1_information_then_requirer_f1_port_is_pushed_to_the_relation_databag( # noqa: E501 + def test_given_invalid_f1_port_when_fiveg_f1_provider_available_then_error_is_raised(self): + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) + params = {"port": "Not a valid port"} + + with pytest.raises(Exception) as e: + self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) + + assert "Invalid relation data" in str(e.value) + + def test_given_charm_is_not_leader_when_set_f1_information_then_error_is_raised(self): + fiveg_f1_relation = testing.Relation(endpoint="fiveg_f1", interface="fiveg_f1") + state_in = testing.State(relations=[fiveg_f1_relation], leader=False) + params = {"port": "1234"} + + with pytest.raises(Exception) as e: + self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) + + assert "Unit must be leader to set application relation data." in str(e.value) + + def test_given_f1_relation_does_not_exist_when_set_f1_information_then_error_is_raised(self): + state_in = testing.State(relations=[], leader=True) + params = {"port": "1234"} + + with pytest.raises(Exception) as e: + self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) + + assert "Relation fiveg_f1 not created yet." in str(e.value) + + def test_given_remote_app_filled_databag_when_get_f1_information_then_value_is_retrieved( self, ): + plmns_as_string = json.dumps([plmn.asdict() for plmn in [VALID_PLMN]]) fiveg_f1_relation = testing.Relation( endpoint="fiveg_f1", interface="fiveg_f1", + remote_app_data={ + "f1_ip_address": "1.2.3.4", + "f1_port": "1234", + "tac": "12", + "plmns": plmns_as_string, + }, ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, - ) + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) params = { - "port": "1234", + "expected_ip_address": "1.2.3.4", + "expected_port": "1234", + "expected_tac": "12", + "expected_plmns": plmns_as_string, } - state_out = self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) - - relation = state_out.get_relation(fiveg_f1_relation.id) - assert relation.local_app_data["f1_port"] == "1234" - - def test_given_invalid_f1_port_when_fiveg_f1_provider_available_then_error_is_raised(self): + self.ctx.run(self.ctx.on.action("get-f1-information", params=params), state_in) + + @pytest.mark.parametrize( + "remote_data", + [ + pytest.param( + { + "f1_ip_address": "1.2.3.4", + "f1_port": "port", + "tac": "22", + }, + id="invalid_port", + ), + pytest.param( + { + "f1_ip_address": "1.2.3.4", + "f1_port": "1234", + "tac": "tac", + }, + id="invalid_tac", + ), + ], + ) + def test_given_invalid_remote_databag_when_get_f1_information_then_none_is_retrieved( + self, remote_data + ): + plmns_as_string = json.dumps([plmn.asdict() for plmn in [VALID_PLMN]]) + remote_data["plmns"] = plmns_as_string fiveg_f1_relation = testing.Relation( endpoint="fiveg_f1", interface="fiveg_f1", + remote_app_data=remote_data, ) - state_in = testing.State( - relations=[fiveg_f1_relation], - leader=True, - ) - params = { - "port": "Not a valid port", - } + state_in = testing.State(relations=[fiveg_f1_relation], leader=True) - with pytest.raises(Exception) as e: - self.ctx.run(self.ctx.on.action("set-f1-information", params=params), state_in) + self.ctx.run(self.ctx.on.action("get-f1-information-invalid", params={}), state_in) - assert "Invalid relation data" in str(e.value) + def test_given_f1_relation_does_not_exist_when_get_f1_information_then_none_is_retrieved( + self, + ): + state_in = testing.State(relations=[], leader=True) + + self.ctx.run(self.ctx.on.action("get-f1-information-invalid", params={}), state_in) diff --git a/tests/unit/test_charm_configure.py b/tests/unit/test_charm_configure.py index c2002c9..ff63ca1 100644 --- a/tests/unit/test_charm_configure.py +++ b/tests/unit/test_charm_configure.py @@ -5,11 +5,15 @@ import os import tempfile +from charms.oai_ran_cu_k8s.v0.fiveg_f1 import PLMNConfig from ops import testing from ops.pebble import Layer from tests.unit.fixtures import CUCharmFixtures +HARDCODED_PLMNS = [PLMNConfig(mcc="001", mnc="01", sst=1, sd=12)] +HARDCODED_TAC = 1 + class TestCharmConfigure(CUCharmFixtures): def test_given_statefulset_is_not_patched_when_config_changed_then_statefulset_is_patched( @@ -340,6 +344,8 @@ def test_given_charm_is_configured_and_running_when_f1_relation_is_added_then_f1 self.mock_f1_set_information.assert_called_once_with( ip_address="192.168.254.7", port=2152, + tac=HARDCODED_TAC, + plmns=HARDCODED_PLMNS, ) def test_given_charm_is_active_when_config_changed_then_updated_f1_interface_ip_and_port_is_published( # noqa: E501 @@ -391,6 +397,8 @@ def test_given_charm_is_active_when_config_changed_then_updated_f1_interface_ip_ self.mock_f1_set_information.assert_called_with( ip_address=test_f1_ip_address.split("/")[0], port=3522, + tac=HARDCODED_TAC, + plmns=HARDCODED_PLMNS, ) def test_given_n3_route_not_created_when_config_changed_then_n3_route_is_created(self):