XR Driver add consistency unit tests

class CmConnection:

    CONSISTENCY_WAIT_LOG_INTERVAL = 1.0

    def __init__(self, address: str, port: int, username: str, password: str, timeout=30, tls_verify=True, consistency_mode: ConsistencyMode = ConsistencyMode.asynchronous, retry_interval: float=0.2) -> None:

    def __init__(self, address: str, port: int, username: str, password: str, timeout=30, tls_verify=True, consistency_mode: ConsistencyMode = ConsistencyMode.asynchronous, retry_interval: float=0.2, max_consistency_tries:int = 100_000) -> None:

        self.__tls_verify = tls_verify

        if not tls_verify:

            urllib3.disable_warnings()

        self.__consistency_mode = consistency_mode

        self.__timeout = timeout

        self.__retry_interval = retry_interval if retry_interval > 0.01 else 0.01

        # Consistency tries limit is mostly useful for testing where it can be use to make

        # test cases faster without timing dependency

        self.__max_consistency_tries = max_consistency_tries

        self.__username = username

        self.__password = password

        self.__cm_root = 'https://' + address + ':' + str(port)

        log_ts = ts_start

        get_result = get_fn()

        valid = False

        limit = self.__max_consistency_tries

        while True:

            if get_result:

                if self.__consistency_mode == ConsistencyMode.synchronous:

                if ts - log_ts >= self.CONSISTENCY_WAIT_LOG_INTERVAL:

                    log_ts = ts

                    LOGGER.info(f"apply_create_consistency(): waiting for REST API object for {obj}, ellapsed time {ts-ts_start} seconds")

            if time.perf_counter() - ts_start > self.__timeout:

            limit -= 1

            if limit < 0 or time.perf_counter() - ts_start > self.__timeout:

                break

            time.sleep(self.__retry_interval)

            get_result = get_fn()

                LOGGER.info(f"Failed to apply create consistency for {get_result}, insufficient life-cycle-state progress ({str(get_result.life_cycle_info)}), duration {duration} seconds")

            else:

                LOGGER.info(f"Failed to apply create consistency for {obj}, REST object did not appear, duration {duration} seconds")

            return None

        else:

            LOGGER.info(f"Applied create consistency for {get_result}, final life-cycle-state {str(get_result.life_cycle_info)}, duration {duration} seconds")

        log_ts = ts_start

        get_result = get_fn()

        valid = False

        limit = self.__max_consistency_tries

        while True:

            if not get_result:

                # Object no longer exist, so this is completely successful operation

                        else:

                            LOGGER.info(f"apply_delete_consistency(): waiting for life cycle state progress for {get_result}, current: {str(get_result.life_cycle_info)}, ellapsed time {ts-ts_start} seconds")

            if time.perf_counter() - ts_start > self.__timeout:

            limit -= 1

            if limit < 0 or time.perf_counter() - ts_start > self.__timeout:

                break

            time.sleep(self.__retry_interval)

            get_result = get_fn()

import inspect

import os

import json

import requests_mock

import traceback

import copy

import requests_mock

from ..cm_connection import CmConnection

from ..cm_connection import CmConnection, ConsistencyMode

from ..tf import set_config_for_service

access_token = r'{"access_token":"eyI3...","expires_in":3600,"refresh_expires_in":0,"refresh_token":"ey...","token_type":"Bearer","not-before-policy":0,"session_state":"f6e235c4-4ca4-4258-bede-4f2b7125adfb","scope":"profile email offline_access"}'

        ]

        assert called_mocks == expected_mocks

# In life cycle tests, multiple queries are performed by the driver to check life cycle progress.

# Extend expected mock to match called mock length by replicating the last item (the repeated GET)

def repeat_last_expected(expected: list[tuple], called: list[tuple]) -> list[tuple]:

    diff = len(called) - len(expected)

    if diff > 0:

        expected = list(expected) # Don't modify the original list

        expected.extend([expected[-1]] * diff)

    return expected

def test_xr_set_config_consistency_lifecycle():

    with mock_cm() as m:

        cm = CmConnection("127.0.0.1", 9999, "xr-user", "xr-password", tls_verify=False, consistency_mode=ConsistencyMode.lifecycle, retry_interval=0, timeout=1, max_consistency_tries=3)

        assert cm.Connect()

        constellation = cm.get_constellation_by_hub_name("XR HUB 1")

        assert constellation

        # Note that JSON here is for different object, but we are not inspecting fields where it would matter (e.g. IDs).

        json_terminal = res_connection_by_name_json[0]

        json_non_terminal = copy.deepcopy(json_terminal)

        json_non_terminal["state"]["lifecycleState"] = "pendingConfiguration"

        # We go trough 404 and non-terminal lstate first and then terminal state.

        m.get("https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432",

              [{'text': '', 'status_code': 404},

               { 'json': json_non_terminal, 'status_code': 200 },

               {'json': json_terminal, 'status_code': 200  }])

        result = set_config_for_service(cm, constellation, uuid, config)

        _validate_result(result, True)

        called_mocks = [(r._request.method, r._request.url) for r in m._adapter.request_history]

        expected_mocks = [

            ('POST', 'https://127.0.0.1:9999/realms/xr-cm/protocol/openid-connect/token'), # Authentication

            ('GET',  'https://127.0.0.1:9999/api/v1/ns/xr-networks?content=expanded&content=expanded&q=%7B%22hubModule.state.module.moduleName%22%3A+%22XR+HUB+1%22%7D'),  # Hub module by name

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections?content=expanded&q=%7B%22state.name%22%3A+%22TF%3A12345ABCDEFGHIJKLMN%22%7D'),  # Get by name, determine update or create

            ('POST', 'https://127.0.0.1:9999/api/v1/ncs/network-connections'), # Create

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432?content=expanded'), # Life cycle state check --> no REST API object

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432?content=expanded'), # Life cycle state check --> non-terminal

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432?content=expanded') # Life cycle state check --> terminal

        ]

        assert called_mocks == expected_mocks

        ################################################################################

        # Same as before, but without life cycle progress

        m.reset_mock()

        m.get("https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432",

              [{'text': '', 'status_code': 401},

               { 'json': json_non_terminal, 'status_code': 200 }])

        result = set_config_for_service(cm, constellation, uuid, config)

        _validate_result(result, False) # Service creation failure due to insufficient progress

        called_mocks = [(r._request.method, r._request.url) for r in m._adapter.request_history]

        expected_mocks_no_connect = [

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections?content=expanded&q=%7B%22state.name%22%3A+%22TF%3A12345ABCDEFGHIJKLMN%22%7D'),  # Get by name, determine update or create

            ('POST', 'https://127.0.0.1:9999/api/v1/ncs/network-connections'), # Create

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432?content=expanded'), # Life cycle state check --> no REST API object

            ('GET',  'https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432?content=expanded'), # Life cycle state check --> non-terminal

        ]

        assert called_mocks == repeat_last_expected(expected_mocks_no_connect, called_mocks)

        ################################################################################

        # Same as before, but CmConnection no longer requiring lifcycle progress

        m.reset_mock()

        cm = CmConnection("127.0.0.1", 9999, "xr-user", "xr-password", tls_verify=False, consistency_mode=ConsistencyMode.synchronous, retry_interval=0, timeout=1, max_consistency_tries=3)

        assert cm.Connect()

        constellation = cm.get_constellation_by_hub_name("XR HUB 1")

        assert constellation

        m.get("https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432",

              [{'text': '', 'status_code': 401},

               { 'json': json_non_terminal, 'status_code': 200 }])

        result = set_config_for_service(cm, constellation, uuid, config)

        _validate_result(result, True)

        called_mocks = [(r._request.method, r._request.url) for r in m._adapter.request_history]

        assert called_mocks == expected_mocks[:2] + expected_mocks_no_connect

        ################################################################################

        # Same as above, but without REST object appearing

        m.reset_mock()

        cm = CmConnection("127.0.0.1", 9999, "xr-user", "xr-password", tls_verify=False, consistency_mode=ConsistencyMode.synchronous, retry_interval=0, timeout=1, max_consistency_tries=3)

        assert cm.Connect()

        constellation = cm.get_constellation_by_hub_name("XR HUB 1")

        assert constellation

        m.get("https://127.0.0.1:9999/api/v1/ncs/network-connections/c3b31608-0bb7-4a4f-9f9a-88b24a059432",

              [{'text': '', 'status_code': 401}])

        result = set_config_for_service(cm, constellation, uuid, config)

        _validate_result(result, False)

        called_mocks = [(r._request.method, r._request.url) for r in m._adapter.request_history]

        assert called_mocks == repeat_last_expected(expected_mocks[:2] + expected_mocks_no_connect, called_mocks)

def test_xr_set_config_update_case():

    with mock_cm() as m:

        cm = CmConnection("127.0.0.1", 9999, "xr-user", "xr-password", tls_verify=False)