diff --git a/tests/integration/juju_helper.py b/tests/integration/juju_helper.py index 4616ec25..5d4172d5 100644 --- a/tests/integration/juju_helper.py +++ b/tests/integration/juju_helper.py @@ -7,7 +7,7 @@ import time from contextlib import contextmanager from subprocess import CalledProcessError, call, check_output -from typing import Optional +from typing import Optional, Tuple logger = logging.getLogger(__name__) @@ -177,6 +177,91 @@ def set_application_config(model_name: str, application_name: str, config: dict) juju_wait_for_active_idle(model_name, 60) +def _get_juju_secret_id(model_name: str, juju_secret_label: str) -> Optional[str]: + """Get Juju secret ID. + + The `juju secrets` command returns a JSON object with the following format: + { + "csuci57mp25c7993rgeg": { + "revision": 1, + "owner": "nms", + "label": "NMS_LOGIN", + "created": "2024-11-19T17:21:25Z", + "updated": "2024-11-19T17:21:25Z" + } + } + + Args: + model_name(str): Juju model name + juju_secret_label(str): Juju secret label + """ + with juju_context(model_name): + cmd_out = check_output(["juju", "secrets", "--format=json"]).decode() + for key, value in json.loads(cmd_out).items(): + if value["label"] == juju_secret_label: + return key + return None + + +def wait_for_nms_credentials( + model_name, juju_secret_label, timeout=300 +) -> Tuple[Optional[str], Optional[str]]: + """Wait for NMS credentials to be available in Juju secret. + + Args: + model_name(str): Juju model name + juju_secret_label(str): Juju secret label + timeout(int): Time to wait for the credentials to be available + + Raises: + TimeoutError: Raised if NMS credentials are not available within given time + """ + now = time.time() + while time.time() - now <= timeout: + username, password = get_nms_credentials(model_name, juju_secret_label) + if username and password: + return username, password + time.sleep(5) + raise TimeoutError( + f"Timed out waiting for NMS credentials in juju secret after {timeout} seconds!" + ) + + +def get_nms_credentials( + model_name: str, juju_secret_label: str +) -> Tuple[Optional[str], Optional[str]]: + """Get NMS credentials from Juju secret. + + The `juju show-secret --reveal` command returns a JSON object with the following format: + { + "csuci57mp25c7993rgeg": { + "revision": 1, + "owner": "nms", + "label": "NMS_LOGIN", + "created": "2024-11-19T17:21:25Z", + "updated": "2024-11-19T17:21:25Z", + "content": { + "Data": { + "password": "whatever-password", + "token": "whatever-token", + "username": "whatever-username" + } + } + } + } + """ + secret_id = _get_juju_secret_id(model_name, juju_secret_label) + if not secret_id: + logger.warning("could not find secret with label %s", juju_secret_label) + return None, None + with juju_context(model_name): + cmd_out = check_output( + ["juju", "show-secret", "--reveal", "--format=json", secret_id] + ).decode() + secret_content = json.loads(cmd_out)[secret_id]["content"]["Data"] + return secret_content.get("username"), secret_content.get("password") + + @contextmanager def juju_context(model_name: str): """Allow changing currently active Juju model. diff --git a/tests/integration/nms_helper.py b/tests/integration/nms_helper.py index 8a533429..3082d664 100644 --- a/tests/integration/nms_helper.py +++ b/tests/integration/nms_helper.py @@ -1,15 +1,23 @@ #!/usr/bin/env python3 -# Copyright 2023 Canonical Ltd. +# Copyright 2024 Canonical Ltd. # See LICENSE file for licensing details. +"""Module use to handle NMS API calls.""" + import json import logging import time +from dataclasses import asdict, dataclass +from typing import Any, List, Optional import requests logger = logging.getLogger(__name__) +ACCOUNTS_URL = "config/v1/account" + +JSON_HEADER = {"Content-Type": "application/json"} + SUBSCRIBER_CONFIG = { "UeId": "PLACEHOLDER", "plmnId": "00101", @@ -17,6 +25,7 @@ "key": "5122250214c33e723a5dd523fc145fc0", "sequenceNumber": "16f3b3f70fc2", } + DEVICE_GROUP_CONFIG = { "imsis": [], "site-info": "demo", @@ -34,6 +43,8 @@ }, }, } + + NETWORK_SLICE_CONFIG = { "slice-id": {"sst": "1", "sd": "102030"}, "site-device-group": [], @@ -46,65 +57,144 @@ } -class Nms: - def __init__(self, nms_ip: str) -> None: - """Construct the NMS class. +@dataclass +class StatusResponse: + """Response from NMS when checking the status.""" - Args: - nms_ip (str): IP address of the NMS application unit - """ - self.nms_ip = nms_ip + initialized: bool - def create_subscriber(self, imsi: str) -> None: - """Create a subscriber. - Args: - imsi (str): Subscriber's IMSI - """ - SUBSCRIBER_CONFIG["UeId"] = imsi - url = f"https://{self.nms_ip}:5000/api/subscriber/imsi-{imsi}" - response = requests.post(url=url, data=json.dumps(SUBSCRIBER_CONFIG), verify=False) - response.raise_for_status() - logger.info(f"Created subscriber with IMSI {imsi}.") +@dataclass +class LoginParams: + """Parameters to login to NMS.""" - def create_device_group(self, device_group_name: str, imsis: list) -> None: - """Create a device group. + username: str + password: str - Args: - device_group_name (str): Device group name - imsis (list): List of IMSIs to be included in the device group - """ - DEVICE_GROUP_CONFIG["imsis"] = imsis - url = f"https://{self.nms_ip}:5000/config/v1/device-group/{device_group_name}" - response = requests.post(url, json=DEVICE_GROUP_CONFIG, verify=False) - response.raise_for_status() - now = time.time() - timeout = 5 - while time.time() - now <= timeout: - if requests.get(url, verify=False).json(): - logger.info(f"Created device group {device_group_name}.") + +@dataclass +class LoginResponse: + """Response from NMS when logging in.""" + + token: str + + +class NMS: + """Handle NMS API calls.""" + + def __init__(self, url: str): + if url.endswith("/"): + url = url[:-1] + self.url = url + + def _make_request( + self, + method: str, + endpoint: str, + token: Optional[str] = None, + data: any = None, # type: ignore[reportGeneralTypeIssues] + ) -> Any | None: + """Make an HTTP request and handle common error patterns.""" + headers = JSON_HEADER + if token: + headers["Authorization"] = f"Bearer {token}" + url = f"{self.url}{endpoint}" + try: + response = requests.request( + method=method, + url=url, + headers=headers, + json=data, + verify=False, + ) + except requests.exceptions.SSLError as e: + logger.error("SSL error: %s", e) + return None + except requests.RequestException as e: + logger.error("HTTP request failed: %s", e) + return None + except OSError as e: + logger.error("couldn't complete HTTP request: %s", e) + return None + try: + response.raise_for_status() + except requests.HTTPError: + logger.error( + "Request failed: code %s", + response.status_code, + ) + return None + try: + json_response = response.json() + except json.JSONDecodeError: + return None + return json_response + + def is_initialized(self) -> bool: + """Return if NMS is initialized.""" + status = self.get_status() + return status.initialized if status else False + + def is_api_available(self) -> bool: + """Return if NMS is reachable.""" + status = self.get_status() + return status is not None + + def get_status(self) -> StatusResponse | None: + """Return if NMS is initialized.""" + response = self._make_request("GET", "/status") + if response: + return StatusResponse( + initialized=response.get("initialized"), + ) + return None + + def wait_for_api_to_be_available(self, timeout: int = 300) -> None: + """Wait for NMS API to be available.""" + t0 = time.time() + while time.time() - t0 < timeout: + if self.is_api_available(): + return + time.sleep(5) + raise TimeoutError(f"NMS API is not available after {timeout} seconds.") + + def wait_for_initialized(self, timeout: int = 300) -> None: + """Wait for NMS to be initialized.""" + t0 = time.time() + while time.time() - t0 < timeout: + if self.is_initialized(): return - else: - time.sleep(1) - raise TimeoutError("Timed out creating device group.") + time.sleep(5) + raise TimeoutError(f"NMS is not initialized after {timeout} seconds.") + + def login(self, username: str, password: str) -> LoginResponse | None: + """Login to NMS by sending the username and password and return a Token.""" + login_params = LoginParams(username=username, password=password) + response = self._make_request("POST", "/login", data=asdict(login_params)) + if response: + return LoginResponse( + token=response.get("token"), + ) + return None + + def create_subscriber(self, imsi: str, token: str) -> None: + """Create a subscriber.""" + url = f"/api/subscriber/imsi-{imsi}" + data = SUBSCRIBER_CONFIG.copy() + data["UeId"] = imsi + self._make_request("POST", url, token=token, data=data) + logger.info(f"Created subscriber with IMSI {imsi}.") - def create_network_slice(self, network_slice_name: str, device_groups: list) -> None: - """Create a network slice. + def create_device_group(self, name: str, imsis: List[str], token: str) -> None: + """Create a device group.""" + DEVICE_GROUP_CONFIG["imsis"] = imsis + url = f"/config/v1/device-group/{name}" + self._make_request("POST", url, token=token, data=DEVICE_GROUP_CONFIG) + logger.info(f"Created device group {name}.") - Args: - network_slice_name (str): Network slice name - device_groups (list): List of device groups to be included in the network slice - """ + def create_network_slice(self, name: str, device_groups: List[str], token: str) -> None: + """Create a network slice.""" NETWORK_SLICE_CONFIG["site-device-group"] = device_groups - url = f"https://{self.nms_ip}:5000/config/v1/network-slice/{network_slice_name}" - response = requests.post(url, json=NETWORK_SLICE_CONFIG, verify=False) - response.raise_for_status() - now = time.time() - timeout = 5 - while time.time() - now <= timeout: - if requests.get(url, verify=False).json(): - logger.info(f"Created network slice {network_slice_name}.") - return - else: - time.sleep(1) - raise TimeoutError("Timed out creating network slice.") + url = f"/config/v1/network-slice/{name}" + self._make_request("POST", url, token=token, data=NETWORK_SLICE_CONFIG) + logger.info(f"Created network slice {name}.") diff --git a/tests/integration/test_integration.py b/tests/integration/test_integration.py index d397ea93..424c02c0 100644 --- a/tests/integration/test_integration.py +++ b/tests/integration/test_integration.py @@ -14,7 +14,7 @@ from requests.auth import HTTPBasicAuth from tests.integration import juju_helper -from tests.integration.nms_helper import Nms +from tests.integration.nms_helper import NMS from tests.integration.terraform_helper import TerraformClient logger = logging.getLogger(__name__) @@ -27,6 +27,7 @@ TEST_DEVICE_GROUP_NAME = "default-default" TEST_IMSI = "001010100007487" TEST_NETWORK_SLICE_NAME = "default" +NMS_CREDENTIALS_LABEL = "NMS_LOGIN" class TestSDCoreBundle: @@ -45,8 +46,15 @@ async def test_given_sdcore_terraform_module_when_deploy_then_status_is_active(s @pytest.mark.abort_on_fail async def test_given_sdcore_bundle_and_gnbsim_deployed_when_start_simulation_then_simulation_success_status_is_true( # noqa: E501 - self, configure_sdcore + self, ): + username, password = juju_helper.wait_for_nms_credentials( + model_name=SDCORE_MODEL_NAME, + juju_secret_label=NMS_CREDENTIALS_LABEL, + ) + if not username or not password: + raise Exception("NMS credentials not found.") + configure_sdcore(username, password) for _ in range(5): action_output = juju_helper.juju_run_action( model_name=RAN_MODEL_NAME, @@ -146,27 +154,41 @@ async def _get_grafana_url_and_admin_password() -> Tuple[str, str]: return action_output["url"], action_output["admin-password"] -@pytest.fixture(scope="module") @pytest.mark.abort_on_fail -def configure_sdcore(): +def configure_sdcore(username: str, password: str) -> None: """Configure Charmed SD-Core. Configuration includes: - subscriber creation - device group creation - network slice creation + + Args: + username (str): NMS username + password (str): NMS password """ nms_ip_address = juju_helper.get_unit_address( model_name=SDCORE_MODEL_NAME, application_name="nms", unit_number=0, ) - nms_client = Nms(nms_ip_address) - nms_client.create_subscriber(TEST_IMSI) - nms_client.create_device_group(TEST_DEVICE_GROUP_NAME, [TEST_IMSI]) - nms_client.create_network_slice(TEST_NETWORK_SLICE_NAME, [TEST_DEVICE_GROUP_NAME]) - # 5 seconds for the config to propagate - time.sleep(5) + nms_client = NMS(url=f"https://{nms_ip_address}:5000") + nms_client.wait_for_api_to_be_available() + nms_client.wait_for_initialized() + login_response = nms_client.login(username=username, password=password) + if not login_response or not login_response.token: + raise Exception("Failed to login to NMS.") + nms_client.create_subscriber(imsi=TEST_IMSI, token=login_response.token) + nms_client.create_device_group( + name=TEST_DEVICE_GROUP_NAME, imsis=[TEST_IMSI], token=login_response.token + ) + nms_client.create_network_slice( + name=TEST_NETWORK_SLICE_NAME, + device_groups=[TEST_DEVICE_GROUP_NAME], + token=login_response.token, + ) + # 60 seconds for the config to propagate + time.sleep(60) @pytest.fixture(scope="module")