diff --git a/README.md b/README.md index 7e98f73c4..25e289994 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,26 @@ except UserAlreadyExistsError as error: ``` +## API usage examples + +### AdminTechAPI + +
+ Python (click to expand) + +```Python +from vmngclient.session import create_vManageSession +from vmngclient.api.admin_tech_api import AdminTechAPI + +session = create_vManageSession(url=..., username=..., password=...) +admintech = AdminTechAPI(session) +filename = admintech.generate("172.16.255.11") +admintech.download(filename) +admintech.delete(filename) +``` + +
+ ## Contributing, reporting issues, seeking support Please contact authors direcly or via Issues Github page. diff --git a/vmngclient/__init__.py b/vmngclient/__init__.py index ef31ea9f9..cbe6b1d96 100644 --- a/vmngclient/__init__.py +++ b/vmngclient/__init__.py @@ -5,4 +5,7 @@ LOGGING_CONF_DIR: Final[str] = str(Path(__file__).parents[0] / 'logging.conf') -logging.config.fileConfig(LOGGING_CONF_DIR, disable_existing_loggers=False) +vmngclient_logger = logging.getLogger(__name__) + +if not vmngclient_logger.handlers: + logging.config.fileConfig(LOGGING_CONF_DIR, disable_existing_loggers=False) diff --git a/vmngclient/api/admin_tech_api.py b/vmngclient/api/admin_tech_api.py index eb810a0fa..b0a63f39f 100644 --- a/vmngclient/api/admin_tech_api.py +++ b/vmngclient/api/admin_tech_api.py @@ -1,23 +1,29 @@ """ Module for handling admintech logs for a device """ -import json +import logging import time from pathlib import Path -from typing import List, Optional, cast -from urllib.error import HTTPError +from typing import List, Optional from requests import Response +from requests.exceptions import HTTPError -from vmngclient.dataclasses import AdminTech +from vmngclient.dataclasses import AdminTech, DeviceAdminTech from vmngclient.session import vManageSession from vmngclient.utils.creation_tools import create_dataclass +logger = logging.getLogger(__name__) + class GenerateAdminTechLogError(Exception): pass +class DownloadAdminTechLogError(Exception): + pass + + class RequestTokenIdNotFound(Exception): pass @@ -35,17 +41,18 @@ def __init__(self, session: vManageSession) -> None: def __str__(self) -> str: return str(self.session) - def get(self, device_id: str) -> AdminTech: + def get(self, device_id: str) -> List[DeviceAdminTech]: """Gets admintech log information for a device. Args: device_id: device ID (usually system-ip) Returns: - AdminTech object for given device + AdminTech object list for given device """ body = {'deviceIP': device_id} - response = self.session.post(url='/dataservice/device/tools/admintechlist', data=body).json() - return create_dataclass(AdminTech, response[0]) + response = self.session.post(url='/dataservice/device/tools/admintechlist', json=body) + items = response.json()["data"] + return [create_dataclass(DeviceAdminTech, item) for item in items] def get_all(self) -> List[AdminTech]: """Gets admintech log information for all devices. @@ -53,72 +60,98 @@ def get_all(self) -> List[AdminTech]: Returns: AdminTech objects list for all devices """ - response = self.session.get_data('/dataservice/device/tools/admintechs') - return [create_dataclass(AdminTech, dev_data) for dev_data in response] + response = self.session.get('/dataservice/device/tools/admintechs') + items = response.json()["data"] + return [create_dataclass(AdminTech, item) for item in items] def generate( - self, device_id: str, request_timeout: int = 3600, polling_timeout: int = 1200, polling_interval: int = 30 + self, + device_id: str, + exclude_cores: bool = True, + exclude_tech: bool = False, + exclude_logs: bool = True, + request_timeout: int = 3600, + polling_timeout: int = 1200, + polling_interval: int = 30, ) -> str: """Generates admintech log for a device. - Args: device_id: device ID (usually system-ip) - request_timeout: wait time in seconds to generate admin tech after request + exclude_cores: exclude core in generated admintech log file + exclude_tech: exclude tech in generated admintech log file + exclude_logs: exclude logs in generated admintech log file + request_timeout: wait time in seconds to generate admintech after request polling_timeout: retry period in seconds for successfull request polling_interval: polling interval in seconds between request attempts Returns: filename of generated admintech log """ - create_admin_tech_error_msgs = 'Admin tech creation already in progress' - body = {'deviceIP': device_id, 'exclude-cores': True, 'exclude-tech': False, 'exclude-logs': True} + create_admin_tech_error_msgs = "Admin tech creation already in progress" + body = { + "deviceIP": device_id, + "exclude-cores": exclude_cores, + "exclude-tech": exclude_tech, + "exclude-logs": exclude_logs, + } polling_timer = polling_timeout while polling_timer > 0: + logger.info( + f"Starting AdminTech log creation for {device_id}, waiting up to {request_timeout} seconds to complete" + ) try: - response = self.session.post(url='/dataservice/device/tools/admintech', data=body) - return cast(dict, response)['fileName'] - except HTTPError as error: - error_details = error.read().decode() - if error.code != 400 and create_admin_tech_error_msgs not in json.loads(error_details).get('error').get( - 'details' - ): - raise GenerateAdminTechLogError(f'It is not possible to generate admintech log for {device_id}') - time.sleep(polling_interval) - polling_timer -= polling_interval + response = self.session.post( + url="/dataservice/device/tools/admintech", json=body, timeout=request_timeout + ) + except HTTPError as http_error: + response = http_error.response + if response.status_code == 200: + return response.json()["fileName"] + if response.status_code == 400 and create_admin_tech_error_msgs in response.json().get("error", {}).get( + "details", "" + ): + logger.warning(f"Admin tech creation already in progress, retrying in {polling_interval} seconds") + else: + raise GenerateAdminTechLogError(f"It is not possible to generate admintech log for {device_id}") + time.sleep(polling_interval) + polling_timer -= polling_interval raise GenerateAdminTechLogError(f'It is not possible to generate admintech log for {device_id}') - def _get_token_id(self, device_id) -> str: - admin_tech_filename = self.generate(device_id) + def _get_token_id(self, filename: str) -> str: admin_techs = self.get_all() for admin_tech in admin_techs: - if admin_tech_filename == admin_tech.filename: + if filename == admin_tech.filename: return admin_tech.token_id - raise RequestTokenIdNotFound(f'Request Id of admin tech generation request not found for device: {device_id}') + raise RequestTokenIdNotFound( + f"requestTokenId of admin tech generation request not found for file name: {filename}" + ) - def delete(self, device_id: str) -> Response: + def delete(self, filename: str) -> Response: """Deletes admin tech logs for a device. - Args: - device_id: device ID (usually system-ip) + filename: name of admin_tech file Returns: response: http response for delete operation """ - token_id = self._get_token_id(device_id) - response = self.session.delete(f'/dataservice/device/tools/admintech/{token_id}') + token_id = self._get_token_id(filename) + response = self.session.delete(f"/dataservice/device/tools/admintech/{token_id}") + if response.status_code == 200: + logger.info(f"Deleted AdminTech file {filename} on remote") return response - def download(self, admin_tech_name: str, download_dir: Optional[Path] = None) -> Path: + def download(self, filename: str, download_dir: Optional[Path] = None) -> Path: """Downloads admintech log for a device. - Args: - admin_tech_name: name of admin_tech file + filename: name of admin_tech file download_dir: download directory (defaults to current working directory) Returns: path to downloaded admin_tech file """ if not download_dir: download_dir = Path.cwd() - download_path = download_dir / admin_tech_name - url = f'/dataservice/device/tools/admintech/download/{admin_tech_name}' - self.session.get_file(url, download_path) + download_path = download_dir / filename + url = f"/dataservice/device/tools/admintech/download/{filename}" + if self.session.get_file(url=url, filename=download_path).status_code != 200: + raise DownloadAdminTechLogError(f"Cannot download admin tech file: {filename} from remote") + logger.info(f"Downloaded AdminTech file to: {download_path}") return download_path diff --git a/vmngclient/api/administration.py b/vmngclient/api/administration.py index 0daac67a4..071516951 100644 --- a/vmngclient/api/administration.py +++ b/vmngclient/api/administration.py @@ -35,7 +35,7 @@ def create_user(self, user: User) -> None: url_path = "/dataservice/admin/user" data = asdict(user) # type: ignore - response = self.session.post(url=url_path, data=data) + response = self.session.post(url=url_path, json=data) logger.info(response) def delete_user(self, username: str) -> bool: @@ -101,7 +101,7 @@ def enable_sdavc_cloud_connector(self, cloud_connector: CloudConnectorData) -> b """Enables SD-AVC Cloud Connector on vManage.""" url_path = "/dataservice/sdavc/cloudconnector" data = asdict(cloud_connector) # type: ignore - response = self.session.post(url_path, data) + response = self.session.post(url_path, json=data) return True if response.status_code == 200 else False def disable_sdavc_cloud_connector(self) -> bool: diff --git a/vmngclient/api/alarms_api.py b/vmngclient/api/alarms_api.py index a3f153bda..a49357c1d 100644 --- a/vmngclient/api/alarms_api.py +++ b/vmngclient/api/alarms_api.py @@ -52,9 +52,9 @@ def get_alarms( {"value": [value], "field": "acknowledged", "type": "bool", "operator": "equal"} ) - alarms = self.session.post(url=AlarmsAPI.URL, data=query).json() + alarms = self.session.post(url=AlarmsAPI.URL, json=query).json()["data"] - logger.info("Actualas alarms collected successfuly.") + logger.info("Current alarms collected successfully.") return [create_dataclass(AlarmData, flatten_dict(alarm)) for alarm in alarms] @@ -136,7 +136,7 @@ def __check(self, expected: AlarmData) -> bool: return any(checked) def verify(self, expected: Set[AlarmData], timeout_seconds: int, sleep_seconds: int): - """The verifing if the expected alarms is included in the actuals alarms set. + """The verifying if the expected alarms is included in the actual alarms set. Args: expected(Set[AlarmData]): The set expected alarms. @@ -149,7 +149,7 @@ def verify(self, expected: Set[AlarmData], timeout_seconds: int, sleep_seconds: def _log_exception(retry_state): self.logger.error(f"Cannot found alarms in {timeout_seconds}.") - self.logger.error(f"Orignial exception: {retry_state.outcome.exception()}.") + self.logger.error(f"Original exception: {retry_state.outcome.exception()}.") def check(founds): return expected != founds diff --git a/vmngclient/api/basic_api.py b/vmngclient/api/basic_api.py index 32e109996..58ca11897 100644 --- a/vmngclient/api/basic_api.py +++ b/vmngclient/api/basic_api.py @@ -293,11 +293,11 @@ def enable_data_stream(self) -> Iterator: # TODO check "vpn": "0", } url_path = "/dataservice/settings/configuration/vmanagedatastream" - self.session.post(url=url_path, params=query) + self.session.post(url=url_path, json=query) yield None finally: url_path = "/dataservice/settings/configuration/vmanagedatastream" - self.session.post(url=url_path, data=data_stream_status) + self.session.post(url=url_path, json=data_stream_status) def get_bfd_sessions(self, device_id: str) -> List[BfdSessionData]: items = self.session.get_data(f'/dataservice/device/bfd/sessions?deviceId={device_id}') diff --git a/vmngclient/api/device_action_api.py b/vmngclient/api/device_action_api.py index 965435867..54e6e9653 100644 --- a/vmngclient/api/device_action_api.py +++ b/vmngclient/api/device_action_api.py @@ -55,7 +55,7 @@ def execute(self): "deviceType": "controller", "devices": [{"deviceIP": self.dev.id, "deviceId": self.dev.uuid}], } - response = self.session.post_json('/dataservice/device/action/reboot', data=body) + response = self.session.post('/dataservice/device/action/reboot', json=body).json() if response.get('id'): self.action_id = response['id'] else: @@ -110,7 +110,7 @@ def execute(self, valid: bool = True): "validity": "valid" if valid else "invalid", } - response = self.session.post(url='/dataservice/certificate/save/vedge/list', data=body).json() + response = self.session.post(url='/dataservice/certificate/save/vedge/list', json=body).json() if response.get('id'): self.action_id = response['id'] else: diff --git a/vmngclient/api/packet_capture_api.py b/vmngclient/api/packet_capture_api.py index e3bb5ab78..a48b40267 100644 --- a/vmngclient/api/packet_capture_api.py +++ b/vmngclient/api/packet_capture_api.py @@ -78,7 +78,7 @@ def channel(self, device: Device) -> Iterator: try: url_path = r"/dataservice/stream/device/capture" - packet_setup = self.session.post(url=url_path, params=query).json() # TODO check + packet_setup = self.session.post(url=url_path, json=query).json() # TODO check self.packet_channel = create_dataclass(PacketSetup, packet_setup) if self.packet_channel.is_new_session is True: yield self.packet_channel diff --git a/vmngclient/api/speedtest_api.py b/vmngclient/api/speedtest_api.py index 8874eb640..4ceb35174 100644 --- a/vmngclient/api/speedtest_api.py +++ b/vmngclient/api/speedtest_api.py @@ -64,7 +64,7 @@ def perform( "port": "80", } url_path = "/dataservice/stream/device/speed" - setup_speedtest = self.session.post(url_path, start_query).json() + setup_speedtest = self.session.post(url_path, json=start_query).json() speedtest_session = setup_speedtest["sessionId"] @@ -95,7 +95,7 @@ def perform( "size": 10000, } url_path = "/dataservice/statistics/speedtest" - post_speedtest = self.session.post(url_path, end_query).json() + post_speedtest = self.session.post(url_path, json=end_query).json() try: self.speedtest_output.status = disable_speedtest["status"] diff --git a/vmngclient/api/templates.py b/vmngclient/api/templates.py index 07d4dd763..f270cefc6 100644 --- a/vmngclient/api/templates.py +++ b/vmngclient/api/templates.py @@ -1,7 +1,11 @@ +import json import logging +from difflib import Differ +from enum import Enum from typing import List, cast from ciscoconfparse import CiscoConfParse # type: ignore +from requests.exceptions import HTTPError from tenacity import retry, retry_if_result, stop_after_attempt, wait_fixed # type: ignore from vmngclient.dataclasses import Device, Template @@ -13,6 +17,11 @@ logger = logging.getLogger(__name__) +class TemplateType(Enum): + CLI = 'file' + FEATURE = 'template' + + class NotFoundError(Exception): """Used when a template item is not found.""" @@ -34,6 +43,13 @@ def __init__(self, template): self.message = f"Template: {template} is attached to device." +class TemplateTypeError(Exception): + """Used when wrong type template.""" + + def __init__(self, name): + self.message = f"Template: {name} - wrong template type." + + class TemplateAPI: def __init__(self, session: vManageSession) -> None: self.session = session @@ -91,7 +107,7 @@ def wait_for_complete(self, operation_id: str, timeout_seconds: int = 300, sleep def _log_exception(retry_state): logger.error( - f"Operatrion status not achieved in the given time, exception: {retry_state.outcome.exception()}" + f"Operation status not achieved in the given time, exception: {retry_state.outcome.exception()}." ) return False @@ -111,7 +127,12 @@ def check_status(action_data): def wait_for_status(): return self.get_operation_status(operation_id) - return True if wait_for_status() else False + if wait_for_status(): + logger.info(f"The action: {operation_id} - successful") + return True + else: + logger.info(f"The action: {operation_id} - failed") + return False def attach(self, name: str, device: Device) -> bool: """ @@ -125,7 +146,13 @@ def attach(self, name: str, device: Device) -> bool: """ try: template_id = self.get_id(name) + self.template_validation(template_id, device=device) except NotFoundError: + logger.error(f"Error, Template with name {name} not found on {device}.") + return False + except HTTPError as error: + error_details = json.loads(error.response.text) + logger.error(f"Error in config: {error_details['error']['details']}.") return False payload = { "deviceTemplateList": [ @@ -144,7 +171,8 @@ def attach(self, name: str, device: Device) -> bool: ] } endpoint = "/dataservice/template/device/config/attachcli" - response = self.session.post(url=endpoint, data=payload).json() + response = self.session.post(url=endpoint, json=payload).json() + logger.info(f"Attaching a template: {name} to the device: {device.hostname}.") return self.wait_for_complete(response['id']) def device_to_cli(self, device: Device) -> bool: @@ -161,7 +189,8 @@ def device_to_cli(self, device: Device) -> bool: "devices": [{"deviceId": device.uuid, "deviceIP": device.id}], } endpoint = "/dataservice/template/config/device/mode/cli" - response = self.session.post(url=endpoint, data=payload).json() + response = self.session.post(url=endpoint, json=payload).json() + logger.info(f"Changing mode to cli mode for {device.hostname}.") return self.wait_for_complete(response['id']) def get_operation_status(self, operation_id: str) -> List[OperationStatus]: @@ -193,10 +222,12 @@ def delete(self, name: str) -> bool: endpoint = f"/dataservice/template/device/{template.id}" if template.devices_attached == 0: response = self.session.delete(url=endpoint) - return response.status_code == 200 + logger.info(f"Template with name: {name} - deleted.") + return response.ok + logger.info(f"Template: {template} is attached to device - cannot be deleted.") raise AttachedError(template.name) - def create(self, device_model: DeviceModel, name: str, description: str, config: CiscoConfParse) -> str: + def create(self, device_model: DeviceModel, name: str, description: str, config: CiscoConfParse) -> bool: """ Args: @@ -205,19 +236,142 @@ def create(self, device_model: DeviceModel, name: str, description: str, config: description (str): Description template to create. config (CiscoConfParse): The config to device. + Raises: + NameAlreadyExistError: If such template name already exists. + Returns: - str: Id of the created template. + bool: True if create template is successful, otherwise - False. """ try: self.get(name) + logger.error(f"Error, Template with name: {name} exists.") raise NameAlreadyExistError(name) except NotFoundError: - cli_template = CliTemplate(self.session, device_model, name, description) + cli_template = CLITemplate(self.session, device_model, name, description) cli_template.config = config + logger.info(f"Template with name: {name} - created.") return cli_template.send_to_device() + def template_validation(self, id: str, device: Device) -> str: + """Checking the template of the configuration on the machine. + + Args: + id (str): template id to check. + device (Device): The device on which the configuration is to be validate. + + Returns: + str: Validated config. + """ + payload = { + "templateId": id, + "device": { + "csv-status": "complete", + "csv-deviceId": device.uuid, + "csv-deviceIP": device.id, + "csv-host-name": device.hostname, + "csv-templateId": id, + }, + "isEdited": False, + "isMasterEdited": False, + "isRFSRequired": True, + } + endpoint = "/dataservice/template/device/config/config/" + response = self.session.post(url=endpoint, json=payload) + return response.text + + @staticmethod + def compare_template(first: CiscoConfParse, second: CiscoConfParse, full: bool = False, debug: bool = False) -> str: + """ + + Args: + first: First template for comparison. + second: Second template for comparison. + full: Return a full comparison if True, otherwise only the lines that differ. + debug: Adding debug to the logger. Defaults to False. + + Returns: + str: The compared templates. + + Code Meaning + '- ' line unique to sequence 1 + '+ ' line unique to sequence 2 + ' ' line common to both sequences + '? ' line not present in either input sequence + + Example: + >>> a = "!\n tacacs\n server 192.168.1.1\n vpn 2\n secret-key a\n auth-port 151\n exit".splitlines() + >>> b = "!\n tacacs\n server 192.168.1.1\n vpn 3\n secret-key a\n auth-port 151\n exit".splitlines() + >>> a_conf = CiscoConfParse(a) + >>> b_conf = CiscoConfParse(b) + >>> compare = TemplateAPI.compare_template(a_conf, b_conf full=True) + >>> print(compare) + ! + tacacs + server 192.168.1.1 + - vpn 2 + ? ^ + + vpn 3 + ? ^ + secret-key a + auth-port 151 + exit + """ + first_n = list(map(lambda x: x + "\n", first.ioscfg)) + second_n = list(map(lambda x: x + "\n", second.ioscfg)) + compare = list(Differ().compare(first_n, second_n)) + if not full: + compare = [x for x in compare if x[0] in ["?", "-", "+"]] + if debug: + logger.debug("".join(compare)) + return "".join(compare) + + def compare_with_running( + self, template: CiscoConfParse, device: Device, full: bool = False, debug: bool = False + ) -> str: + """The comparison of the config with the one running on the machine. + + Args: + template: The template to compare. + device: The device on which to compare config. + full: Return a full comparison if True, otherwise only the lines that differ. + debug: Adding debug to the logger. Defaults to False. -class CliTemplate: + Returns: + str: The compared templates. + + Example: + >>> a = "!\n tacacs\n server 192.168.1.1\n vpn 512\n secret-key a\n auth-port 151\n exit".splitlines() + >>> a_conf = CiscoConfParse(a) + >>> device = DevicesAPI(API_SESSION).get(DeviceField.HOSTNAME, device_name) + >>> compare = TemplateAPI.compare_template(a_conf, device, full=True) + >>> print(compare) + . + . + . + zbfw-udp-idle-time 30 + ! + ! + + ! + + tacacs + + server 192.168.1.1 + + vpn vpn 512 + + secret-key a + + auth-port 151 + + exit + omp + no shutdown + ecmp-limit 6 + . + . + . + """ + running_config = CLITemplate( + self.session, DeviceModel(device.model), 'running_conf', 'running_conf' + ).load_running(device) + return self.compare_template(running_config, template, debug) + + +class CLITemplate: def __init__(self, session: vManageSession, device_model: DeviceModel, name: str, description: str) -> None: self.session = session self.device_model = device_model @@ -225,31 +379,47 @@ def __init__(self, session: vManageSession, device_model: DeviceModel, name: str self.description = description self.config: CiscoConfParse = CiscoConfParse([]) - def load(self, id: str) -> None: - """Load config from template. + def load(self, id: str) -> CiscoConfParse: + """Load CLI config from template. Args: id (str): The template id from which load config. + + Raises: + TemplateTypeError: wrong template type - CLI required. + + Returns: + CiscoConfParse: Loaded template. """ endpoint = f"/dataservice/template/device/object/{id}" - config = cast(dict, self.session.get_json(endpoint)) + config = self.session.get_json(endpoint) + if TemplateType(config['configType']) == TemplateType.FEATURE: + raise TemplateTypeError(config['templateName']) self.config = CiscoConfParse(config['templateConfiguration'].splitlines()) + return self.config - def load_running(self, device: Device) -> None: + def load_running(self, device: Device) -> CiscoConfParse: """Load running config from device. Args: device (Device): The device from which load config. + + Returns: + CiscoConfParse: A working configuration on the machine. """ endpoint = f"/dataservice/template/config/running/{device.uuid}" - config = cast(dict, self.session.get_json(endpoint)) + config = self.session.get_json(endpoint) self.config = CiscoConfParse(config['config'].splitlines()) + logger.debug(f"Template loaded from {device.hostname}.") + return self.config - def send_to_device(self) -> str: + def send_to_device(self) -> bool: """ Returns: - str: Template id. + bool: True if send template to device is successful, otherwise - False. + + The payload differs depending on the type of machine - for physical machines it has two more attributes. """ config_str = "\n".join(self.config.ioscfg) payload = { @@ -260,18 +430,29 @@ def send_to_device(self) -> str: "factoryDefault": False, "configType": "file", } + if self.device_model not in [DeviceModel.VEDGE, DeviceModel.VSMART, DeviceModel.VMANAGE, DeviceModel.VBOND]: + payload["cliType"] = "device" + payload["draftMode"] = False + endpoint = "/dataservice/template/device/cli/" - response = self.session.post(url=endpoint, data=payload).json() - return response['templateId'] + try: + self.session.post(url=endpoint, json=payload).json() + except HTTPError as error: + response = json.loads(error.response.text)['error'] + logger.error(response['message']) + logger.error(response['details']) + return False + logger.info(f"Template with name: {self.name} - sent to the device.") + return True - def update(self, id: str) -> None: + def update(self, id: str) -> bool: """ Args: id (str): Template id to update. Returns: - str: Process id. + bool: True if update template is successful, otherwise - False. """ config_str = "\n".join(self.config.ioscfg) payload = { @@ -285,7 +466,15 @@ def update(self, id: str) -> None: "draftMode": False, } endpoint = f"/dataservice/template/device/{id}" - self.session.put(url=endpoint, data=payload).json() + try: + self.session.put(url=endpoint, json=payload) + except HTTPError as error: + response = json.loads(error.response.text)['error'] + logger.error(response['message']) + logger.error(response['details']) + return False + logger.info(f"Template with name: {self.name} - updated.") + return True def add_to_config(self, add_config: CiscoConfParse, add_before: str) -> None: """Add config to existing config before provided value. diff --git a/vmngclient/dataclasses.py b/vmngclient/dataclasses.py index d2eec2c2b..c9982133b 100644 --- a/vmngclient/dataclasses.py +++ b/vmngclient/dataclasses.py @@ -20,11 +20,23 @@ def __str__(self): @define(frozen=True, field_transformer=convert_attributes) class AdminTech(DataclassBase): - state: str + creation_time: dt.datetime = field(metadata={FIELD_NAME: "creationTime"}) + size: int filename: str = field(metadata={FIELD_NAME: "fileName"}) - token_id: str = field(metadata={FIELD_NAME: "requestTokenId"}) + state: str + tac_state: Optional[str] device_ip: str = field(metadata={FIELD_NAME: "deviceIP"}) system_ip: str = field(metadata={FIELD_NAME: "local-system-ip"}) + token_id: str = field(metadata={FIELD_NAME: "requestTokenId"}) + + +@define(frozen=True, field_transformer=convert_attributes) +class DeviceAdminTech(DataclassBase): + filename: str = field(metadata={FIELD_NAME: "fileName"}) + creation_time: dt.datetime = field(metadata={FIELD_NAME: "creationTime"}) + size: int + state: str + token_id: Optional[str] = field(default=None, metadata={FIELD_NAME: "requestTokenId"}) @define(frozen=True, field_transformer=convert_attributes) diff --git a/vmngclient/logging.conf b/vmngclient/logging.conf index 16a3f94ee..146a11439 100644 --- a/vmngclient/logging.conf +++ b/vmngclient/logging.conf @@ -27,7 +27,7 @@ args=(sys.stdout,) class=FileHandler level=DEBUG formatter=simpleFormatter -args=('test.log', 'w') +args=('vmngclient.log', 'w') [formatter_simpleFormatter] format=%(asctime)s - %(name)s - %(levelname)s - %(message)s diff --git a/vmngclient/session.py b/vmngclient/session.py index a7a8b3ed6..70155ea85 100644 --- a/vmngclient/session.py +++ b/vmngclient/session.py @@ -263,7 +263,6 @@ def get_virtual_session_id(self, tenant_id: str) -> str: def __prepare_session(self, verify: bool, auth: Optional[AuthBase]) -> None: self.auth = auth self.verify = verify - self.headers.update({"content-type": "application/json"}) def __str__(self) -> str: return f"{self.username}@{self.base_url}" diff --git a/vmngclient/tests/test_admin_tech_api.py b/vmngclient/tests/test_admin_tech_api.py new file mode 100644 index 000000000..117a0ccb7 --- /dev/null +++ b/vmngclient/tests/test_admin_tech_api.py @@ -0,0 +1,185 @@ +import io +import tempfile +import unittest +from pathlib import Path +from unittest.mock import ANY, patch + +from vmngclient.api.admin_tech_api import ( + AdminTechAPI, + DownloadAdminTechLogError, + GenerateAdminTechLogError, + RequestTokenIdNotFound, +) +from vmngclient.dataclasses import DeviceAdminTech + + +class TestAdminTechAPI(unittest.TestCase): + def setUp(self): + self.maxDiff = None + self.device_ip = "169.254.10.4" + self.admin_tech_generate_response = { + "size": 1479740, + "fileName": "172.16.253.129-vm129-20221116-065219-admin-tech.tar.gz", + } + self.device_admin_tech_infos = { + "data": [ + { + "fileName": "172.16.255.200-vm200-20221116-060922-admin-tech.tar.gz", + "creationTime": 1668578961380, + "size": 31934859, + "state": "done", + "requestTokenId": "null", + } + ] + } + self.admin_tech_infos = { + "data": [ + { + "creationTime": 1668581537503, + "size": self.admin_tech_generate_response["size"], + "fileName": self.admin_tech_generate_response["fileName"], + "state": "done", + "tac_state": "notStarted", + "deviceIP": self.device_ip, + "local-system-ip": "172.16.253.129", + "requestTokenId": "aace1605-ba9a-40fa-990b-c694a011a866", + }, + { + "creationTime": 1668578961380, + "size": 31934859, + "fileName": "172.16.255.200-vm200-20221116-060922-admin-tech.tar.gz", + "state": "done", + "tac_state": "notStarted", + "deviceIP": "169.254.10.1", + "local-system-ip": "172.16.255.200", + "requestTokenId": "8b7f9a3a-e137-4af4-b2f9-b04808e1e2eb", + }, + ] + } + self.admin_tech_info = self.admin_tech_infos["data"][0] + self.download_file_content = "Downloaded file content" + self.download_file = io.BytesIO(self.download_file_content.encode()) + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_get(self, mock_session, mock_response): + # Arrange + mock_session.post.return_value = mock_response + mock_response.json.return_value = self.device_admin_tech_infos + # Act + admintechs = AdminTechAPI(mock_session).get(self.device_ip) + # Assert + mock_session.post.assert_called_once_with( + url="/dataservice/device/tools/admintechlist", json={"deviceIP": self.device_ip} + ) + self.assertIsInstance(admintechs[0], DeviceAdminTech) + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_get_all(self, mock_session, mock_response): + # Arrange + mock_session.get.return_value = mock_response + mock_response.json.return_value = self.admin_tech_infos + # Act + admintechs = AdminTechAPI(mock_session).get_all() + # Assert + mock_session.get.assert_called_once_with("/dataservice/device/tools/admintechs") + self.assertEqual(len(admintechs), len(self.admin_tech_infos["data"])) + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_generate(self, mock_session, mock_response): + # Arrange + mock_session.post.return_value = mock_response + mock_response.status_code = 200 + mock_response.json.return_value = self.admin_tech_generate_response + # Act + filename = AdminTechAPI(mock_session).generate( + device_id=self.device_ip, polling_timeout=0.01, polling_interval=0.01 + ) + print(filename) + # Assert + mock_session.post.assert_called_once_with(url="/dataservice/device/tools/admintech", json=ANY, timeout=ANY) + self.assertEqual(filename, self.admin_tech_generate_response["fileName"]) + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_generate_in_progress_error_retry(self, mock_session, mock_response): + # Arrange + mock_session.post.return_value = mock_response + mock_response.status_code = 400 + mock_response.json.return_value = {"error": {"details": "Admin tech creation already in progress"}} + interval = 0.01 + count = 2 + # Act/Assert + with self.assertRaises(GenerateAdminTechLogError): + AdminTechAPI(mock_session).generate( + device_id=self.device_ip, polling_timeout=interval * count, polling_interval=interval + ) + self.assertEqual(mock_session.post.call_count, count) + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_generate_error(self, mock_session, mock_response): + # Arrange + mock_session.post.return_value = mock_response + mock_response.status_code = 500 + mock_response.json.return_value = {"error": {"details": "Server Error"}} + interval = 0.01 + count = 3 + # Act/Assert + with self.assertRaises(GenerateAdminTechLogError): + AdminTechAPI(mock_session).generate( + device_id=self.device_ip, polling_timeout=interval * count, polling_interval=interval + ) + mock_session.post.assert_called_once() + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_delete(self, mock_session, mock_response): + # Arrange + filename = self.admin_tech_generate_response["fileName"] + token_id = self.admin_tech_info["requestTokenId"] + mock_session.get.return_value = mock_response + mock_response.json.return_value = self.admin_tech_infos + # Act + AdminTechAPI(mock_session).delete(filename) + # Assert + mock_session.delete.assert_called_once_with(f"/dataservice/device/tools/admintech/{token_id}") + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_delete_token_not_found(self, mock_session, mock_response): + # Arrange + mock_session.get.return_value = mock_response + mock_response.json.return_value = self.admin_tech_infos + # Act/Assert + with self.assertRaises(RequestTokenIdNotFound): + AdminTechAPI(mock_session).delete("fake-filename.tar.gz") + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_download(self, mock_session, mock_response): + # Arrange + filename = self.admin_tech_generate_response["fileName"] + mock_session.get_file.return_value = mock_response + mock_session.get.return_value = mock_response + mock_response.status_code = 200 + mock_response.content = self.download_file_content + with tempfile.TemporaryDirectory() as tmpdir: + # Act + download_path = AdminTechAPI(mock_session).download(filename, Path(tmpdir)) + # Assert + self.assertEqual(download_path, Path(tmpdir) / filename) + + @patch("vmngclient.session.vManageSession") + @patch("requests.Response") + def test_download_error(self, mock_session, mock_response): + # Arrange + mock_session.get.return_value = mock_response + mock_response.status_code = 500 + mock_response.json.return_value = {"error": {"details": "Server Error"}} + with tempfile.TemporaryDirectory() as tmpdir: + # Act/Assert + with self.assertRaises(DownloadAdminTechLogError): + AdminTechAPI(mock_session).download("fake-filename.tar.gz", Path(tmpdir)) diff --git a/vmngclient/tests/test_devices_api.py b/vmngclient/tests/test_devices_api.py new file mode 100644 index 000000000..67281ed77 --- /dev/null +++ b/vmngclient/tests/test_devices_api.py @@ -0,0 +1,673 @@ +from unittest import TestCase +from unittest.mock import Mock, patch + +from parameterized import parameterized # type: ignore +from tenacity import RetryError + +from vmngclient.api.basic_api import DeviceField, DevicesAPI, DeviceStateAPI, FailedSend +from vmngclient.dataclasses import BfdSessionData, Connection, Device, Reboot, WanInterface +from vmngclient.utils.creation_tools import create_dataclass +from vmngclient.utils.personality import Personality + + +class TestDevicesAPI(TestCase): + def setUp(self) -> None: + self.devices = [ + { # vmanage + "device-model": "vmanage", + "deviceId": "1.1.1.1", + "uuid": "aaaaaaaa-6169-445c-8e49-c0bdaaaaaaa", + "cpuLoad": 4.71, + "state_description": "All daemons up", + "status": "normal", + "memState": "normal", + "local-system-ip": "1.1.1.1", + "board-serial": "11122233", + "personality": "vmanage", + "memUsage": 57.0, + "reachability": "reachable", + "connectedVManages": ["1.1.1.1"], + "host-name": "vm200", + "cpuState": "normal", + "chassis-number": "aaaaaaaa-6aa9-445c-8e49-c0aaaaaaaaa9", + }, + { # vsmart + "device-model": "vsmart", + "deviceId": "1.1.1.3", + "uuid": "bbcccccc-6169-445c-8e49-c0bdccccccc", + "cpuLoad": 2.76, + "state_description": "All daemons up", + "status": "normal", + "memState": "normal", + "local-system-ip": "1.1.1.3", + "board-serial": "11223399", + "personality": "vsmart", + "memUsage": 28.0, + "reachability": "reachable", + "connectedVManages": ["1.1.1.1"], + "host-name": "vm129", + "cpuState": "normal", + "chassis-number": "abbbbbbb-6169-445c-8e49-c0bccccccccc", + }, + { # vbond + "device-model": "vedge-cloud", + "deviceId": "1.1.1.2", + "uuid": "bbbbbbbb-6169-445c-8e49-c0bdaaaaaaa", + "cpuLoad": 1.76, + "state_description": "All daemons up", + "status": "normal", + "memState": "normal", + "local-system-ip": "1.1.1.2", + "board-serial": "11223344", + "personality": "vbond", + "memUsage": 26.0, + "reachability": "reachable", + "connectedVManages": ["1.1.1.1"], + "host-name": "vm128", + "cpuState": "normal", + "chassis-number": "abbbbbbb-6169-445c-8e49-c0bbbbbbbbb9", + }, + { # vedge + "device-model": "vedge-cloud", + "deviceId": "169.254.10.10", + "uuid": "bc2a78ac-a06e-40fd-b2b7-1b1e062f3f9e", + "cpuLoad": 1.46, + "state_description": "All daemons up", + "status": "normal", + "memState": "normal", + "local-system-ip": "172.16.254.2", + "board-serial": "12345708", + "personality": "vedge", + "memUsage": 24.1, + "reachability": "reachable", + "connectedVManages": ["1.1.1.1"], + "host-name": "vm1", + "cpuState": "normal", + "chassis-number": "bc2a78ac-a06e-40fd-b2b7-1b1e062f3f9e", + }, + ] + self.devices_dataclass = [create_dataclass(Device, device) for device in self.devices] + self.controllers_dataclass = [create_dataclass(Device, device) for device in self.devices[:2]] + self.orchestrators_dataclass = [create_dataclass(Device, self.devices[2])] + self.edges_dataclass = [create_dataclass(Device, self.devices[3])] + self.vsmarts_dataclass = [create_dataclass(Device, self.devices[1])] + self.system_ips_list = [device["local-system-ip"] for device in self.devices] + self.ips_list = [device["deviceId"] for device in self.devices] + self.tenants = [ + { + 'flakeId': 11111, + 'orgName': 'my-org Inc', + 'samlSpInfo': '', + 'subDomain': 'sub1.domain.com', + 'vBondAddress': '1.1.1.1', + 'oldIdpMetadata': '', + 'configDBClusterServiceName': '', + 'vSmarts': ['aaaaaaaa-aaac-42f4-b2c7-1aaaaaaaaaaaa', 'bbbbbbbb-9d8a-453c-841f-4bbbbbbbbbbb'], + 'mode': 'off', + 'idpMetadata': '', + 'createdAt': 1111111111111, + '@rid': 1111, + 'tenantId': 'aaaaaaaaaa-9ca8-42fd-8290-67aaaaaaaaaa', + 'name': 'sub1', + 'wanEdgeForecast': '100', + 'spMetadata': '', + 'state': 'READY', + 'wanEdgePresent': 1, + 'desc': 'This is sub1', + }, + { + 'flakeId': 22222, + 'orgName': 'my-org Inc', + 'samlSpInfo': '', + 'subDomain': 'sub2.domain.com', + 'vBondAddress': '1.1.1.2', + 'oldIdpMetadata': '', + 'configDBClusterServiceName': '', + 'vSmarts': ['ccccccccccccccc-42f4-b2c7-1cccccccccc', 'ddddddddd-9d8a-453c-841f-4dddddddddd'], + 'mode': 'off', + 'idpMetadata': '', + 'createdAt': 2222222222222, + '@rid': 2222, + 'tenantId': 'ccccccccca-9ca8-42fd-8290-67aaaccccccc', + 'name': 'sub2', + 'wanEdgeForecast': '100', + 'spMetadata': '', + 'state': 'READY', + 'wanEdgePresent': 1, + 'desc': 'This is sub2', + }, + ] + self.devices_vbonds = [self.devices[2]] + self.devices_vsmatrs = [self.devices[1]] + self.devices_edges = [self.devices[3]] + + @patch.object(DevicesAPI, 'devices') + def test_controllers(self, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.controllers + # Assert + self.assertEqual(answer, self.controllers_dataclass) + + @patch.object(DevicesAPI, 'devices') + def test_orchestrators(self, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.orchestrators + # Assert + self.assertEqual(answer, self.orchestrators_dataclass) + + @patch.object(DevicesAPI, 'devices') + def test_edges(self, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.edges + # Assert + self.assertEqual(answer, self.edges_dataclass) + + @patch.object(DevicesAPI, 'devices') + def test_vsmarts(self, mock_vsmarts): + # Arrange + MockDevices = Mock() + mock_vsmarts.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.vsmarts + # Assert + self.assertEqual(answer, self.vsmarts_dataclass) + + @patch.object(DevicesAPI, 'devices') + def test_system_ips(self, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.system_ips + # Assert + self.assertEqual(answer, self.system_ips_list) + + @patch.object(DevicesAPI, 'devices') + def test_ips(self, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.ips + # Assert + self.assertEqual(answer, self.ips_list) + + @patch('vmngclient.session.vManageSession') + def test_devices(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.devices + # Act + answer = DevicesAPI(mock_session).devices + # Assert + self.assertEqual(answer, self.devices_dataclass) + + def test_get_device_details(self): + pass # TODO fix method before test + + def test_count_devices(self): + pass # TODO fix method before test + + @patch('vmngclient.session.vManageSession') + def test_get_tenants(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.tenants + # Act + answer = DevicesAPI(mock_session).get_tenants() + # Assert + self.assertEqual(answer, self.tenants) + + @patch('vmngclient.session.vManageSession') + def test_get_reachable_devices_for_vsmatrs(self, mock_session, personality=Personality.VSMART): + # Arrange + mock_session.get_data.return_value = [ + device for device in self.devices if device["personality"] == personality.value + ] + # Act + answer = DevicesAPI(mock_session).get_reachable_devices(personality) + # Assert + self.assertEqual(answer, self.devices_vsmatrs) + + @patch('vmngclient.session.vManageSession') + def test_get_reachable_devices_for_vbonds(self, mock_session, personality=Personality.VBOND): + # Arrange + mock_session.get_data.return_value = [ + device for device in self.devices if device["personality"] == personality.value + ] + # Act + answer = DevicesAPI(mock_session).get_reachable_devices(personality) + # Assert + self.assertEqual(answer, self.devices_vbonds) + + @patch('vmngclient.session.vManageSession') + def test_get_reachable_devices_for_edges(self, mock_session, personality=Personality.EDGE): + # Arrange + mock_session.get_data.return_value = [ + device for device in self.devices if device["personality"] == personality.value + ] + # Act + answer = DevicesAPI(mock_session).get_reachable_devices(personality) + # Assert + self.assertEqual(answer, self.devices_edges) + + @patch('vmngclient.session.vManageSession') + def test_get_reachable_devices_for_vmanage(self, mock_session, personality=Personality.VMANAGE): + # Arrange + mock_session.get_data.return_value = [ + device for device in self.devices if device["personality"] == personality.value + ] + + # Act + def answer(): + return DevicesAPI(mock_session).get_reachable_devices(personality) + + # Assert + self.assertRaises(AssertionError, answer) + + @patch('vmngclient.session.vManageSession') + def test_send_certificate_state_to_controllers(self, mock_session): + # Arrange + mock_session.post().json.return_value = {'id': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'} + + # Act + answer = DevicesAPI(mock_session).send_certificate_state_to_controllers() + + # Assert + self.assertTrue(answer) + + @patch('vmngclient.session.vManageSession') + def test_send_certificate_state_to_controllers_error(self, mock_session): + # Arrange + mock_session.post().json.return_value = {} + + # Act + def answer(): + return DevicesAPI(mock_session).send_certificate_state_to_controllers() + + # Assert + self.assertRaises(FailedSend, answer) + + @parameterized.expand([["vm200", 0], ["vm129", 1], ["vm128", 2], ["vm1", 3]]) + @patch.object(DevicesAPI, 'devices') + def test_get_by_hostname(self, hostname, device_number, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.get(DeviceField.HOSTNAME, hostname) + # Assert + self.assertEqual(answer, self.devices_dataclass[device_number]) + + @parameterized.expand([["1.1.1.1", 0], ["1.1.1.3", 1], ["1.1.1.2", 2], ["169.254.10.10", 3]]) + @patch.object(DevicesAPI, 'devices') + def test_get_by_id(self, device_id, device_number, mock_devices): + # Arrange + MockDevices = Mock() + mock_devices.return_value = MockDevices + session = Mock() + test_object = DevicesAPI(session) + test_object.devices = self.devices_dataclass + # Act + answer = test_object.get(DeviceField.ID, device_id) + # Assert + self.assertEqual(answer, self.devices_dataclass[device_number]) + + +class TestDevicesStateAPI(TestCase): + def setUp(self) -> None: + self.crash_info = [ + { + "core-time": "Tue Feb 15 04:14:38 UTC 2022", + "vdevice-dataKey": "169.254.10.12-0", + "vdevice-name": "169.254.10.12", + "index": 0, + "lastupdated": 1645064726542, + "core-filename": "vm5_htx_19219_20220215-041430-UTC.core.gz", + "core-time-date": 1644898478000, + "vdevice-host-name": "vm5", + }, + { + "core-time": "Tue Feb 15 04:15:24 UTC 2022", + "vdevice-dataKey": "169.254.10.12-1", + "vdevice-name": "169.254.10.12", + "index": 1, + "lastupdated": 1645064726542, + "core-filename": "vm5_htx_19539_20220215-041515-UTC.core.gz", + "core-time-date": 1644898524000, + "vdevice-host-name": "vm5", + }, + ] + self.connections_info = [ + { + 'system-ip': '1.1.1.1', + 'peer-type': 'vsmart', + 'state': 'up', + }, + { + 'system-ip': '1.1.1.2', + 'peer-type': 'vedge', + 'state': 'up', + }, + ] + self.connections_info_dataclass = [create_dataclass(Connection, item) for item in self.connections_info] + self.reboot_history = [ + { + "reboot_date_time-date": 1642651714000, + "lastupdated": 1642656685355, + "vdevice-dataKey": "172.16.255.11-2022-01-20T04:08:34+00:00", + "reboot_date_time": "2022-01-20T04:08:34+00:00", + "reboot_reason": "Initiated by user - Reboot issued via NETCONF", + "vdevice-host-name": "vm1", + "vdevice-name": "172.16.255.11", + }, + { + "reboot_date_time-date": 164265171400, + "lastupdated": 1642656685356, + "vdevice-dataKey": "172.16.255.11-2022-01-20T04:08:34+00:00", + "reboot_date_time": "2022-01-20T04:08:34+00:00", + "reboot_reason": "Initiated by user - Reboot issued via NETCONF", + "vdevice-host-name": "vm2", + "vdevice-name": "172.16.255.12", + }, + ] + self.reboot_history_dataclass = [create_dataclass(Reboot, item) for item in self.reboot_history] + self.device = [ + { # vmanage + "device-model": "vmanage", + "deviceId": "1.1.1.1", + "uuid": "aaaaaaaa-6169-445c-8e49-c0bdaaaaaaa", + "cpuLoad": 4.71, + "state_description": "All daemons up", + "status": "normal", + "memState": "normal", + "local-system-ip": "1.1.1.1", + "board-serial": "11122233", + "personality": "vmanage", + "memUsage": 57.0, + "reachability": "reachable", + "connectedVManages": ["1.1.1.1"], + "host-name": "vm200", + "cpuState": "normal", + "chassis-number": "aaaaaaaa-6aa9-445c-8e49-c0aaaaaaaaa9", + } + ] + self.device_dataclass = create_dataclass(Device, self.device[0]) + self.wan_interfaces = [ + { + 'color': 'default', + 'vdevice-name': '1.1.1.1', + 'admin-state': 'up', + 'interface': 'eth1', + 'private-port': 12345, + 'vdevice-host-name': 'vm1', + 'public-ip': '1.1.1.1', + 'operation-state': 'up', + 'public-port': 12345, + 'private-ip': '1.1.1.1', + }, + { + 'color': 'default', + 'vdevice-name': '1.1.1.1', + 'admin-state': 'up', + 'interface': 'eth1', + 'private-port': 11111, + 'vdevice-host-name': 'vm2', + 'public-ip': '1.1.1.1', + 'operation-state': 'up', + 'public-port': 11111, + 'private-ip': '1.1.1.1', + }, + ] + self.wan_interfaces_dataclass = [create_dataclass(WanInterface, item) for item in self.wan_interfaces] + self.bfd_session = [ + { + 'src-ip': '1.1.1.1', + 'dst-ip': '1.1.1.2', + 'color': '3g', + 'system-ip': '1.1.1.1', + 'site-id': 1, + 'local-color': '3g', + 'state': 'up', + }, + { + 'src-ip': '1.1.1.1', + 'dst-ip': '1.1.1.3', + 'color': '3g', + 'system-ip': '1.1.1.1', + 'site-id': 1, + 'local-color': '3g', + 'state': 'up', + }, + ] + self.bfd_session_dataclass = [create_dataclass(BfdSessionData, item) for item in self.bfd_session] + self.bfd_session_down = [ + { + 'src-ip': '1.1.1.1', + 'dst-ip': '1.1.1.2', + 'color': '3g', + 'system-ip': '1.1.1.1', + 'site-id': 1, + 'local-color': '3g', + 'state': 'down', + } + ] + self.device_unreachable = [ + { # vmanage + "device-model": "vmanage", + "deviceId": "1.1.1.1", + "uuid": "zzzzzzzz-6169-445c-8e49-c0bdaaaaaaa", + "cpuLoad": 4.71, + "state_description": "All daemons up", + "status": "normal", + "memState": "normal", + "local-system-ip": "1.1.1.1", + "board-serial": "11122233", + "personality": "vmanage", + "memUsage": 57.0, + "reachability": "unreachable", + "connectedVManages": ["1.1.1.1"], + "host-name": "vm200", + "cpuState": "normal", + "chassis-number": "aaaaaaaa-6aa9-445c-8e49-c0aaaaaaaaa9", + } + ] + + @patch('vmngclient.session.vManageSession') + def test_get_device_crash_info(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.crash_info + # Act + answer = DeviceStateAPI(mock_session).get_device_crash_info(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, self.crash_info) + + @patch('vmngclient.session.vManageSession') + def test_get_device_crash_info_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + # Act + answer = DeviceStateAPI(mock_session).get_device_crash_info(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, []) + + @patch('vmngclient.session.vManageSession') + def test_get_device_control_connections_info(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.connections_info + # Act + answer = DeviceStateAPI(mock_session).get_device_control_connections_info(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, self.connections_info_dataclass) + + @patch('vmngclient.session.vManageSession') + def test_get_device_control_connections_info_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + # Act + answer = DeviceStateAPI(mock_session).get_device_control_connections_info(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, []) + + @patch('vmngclient.session.vManageSession') + def test_get_device_orchestrator_connections_info(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.connections_info + # Act + answer = DeviceStateAPI(mock_session).get_device_orchestrator_connections_info(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, self.connections_info_dataclass) + + @patch('vmngclient.session.vManageSession') + def test_get_device_orchestrator_connections_info_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + # Act + answer = DeviceStateAPI(mock_session).get_device_orchestrator_connections_info(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, []) + + @patch('vmngclient.session.vManageSession') + def test_get_device_reboot_history(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.reboot_history + # Act + answer = DeviceStateAPI(mock_session).get_device_reboot_history(device_id="1.1.1.11") + # Assert + self.assertEqual(answer, self.reboot_history_dataclass) + + @patch('vmngclient.session.vManageSession') + def test_get_device_reboot_history_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + # Act + answer = DeviceStateAPI(mock_session).get_device_reboot_history(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, []) + + @patch('vmngclient.session.vManageSession') + def test_get_system_status(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.device + # Act + answer = DeviceStateAPI(mock_session).get_system_status(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, self.device_dataclass) + + @patch('vmngclient.session.vManageSession') + def test_get_system_status_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + + # Act + def answer(): + return DeviceStateAPI(mock_session).get_system_status(device_id="1.1.1.1") + + # Assert + self.assertRaises(AssertionError, answer) + + @patch('vmngclient.session.vManageSession') + def test_get_device_wan_interfaces(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.wan_interfaces + # Act + answer = DeviceStateAPI(mock_session).get_device_wan_interfaces(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, self.wan_interfaces_dataclass) + + @patch('vmngclient.session.vManageSession') + def test_get_device_wan_interfaces_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + # Act + answer = DeviceStateAPI(mock_session).get_device_wan_interfaces(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, []) + + def test_get_colors(self): + pass # TODO fix method before test + + def test_enable_data_stream(self): + pass # TODO fix method before test + + @patch('vmngclient.session.vManageSession') + def test_get_bfd_sessions(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.bfd_session + # Act + answer = DeviceStateAPI(mock_session).get_bfd_sessions(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, self.bfd_session_dataclass) + + @patch('vmngclient.session.vManageSession') + def test_get_bfd_sessions_empty(self, mock_session): + # Arrange + mock_session.get_data.return_value = [] + # Act + answer = DeviceStateAPI(mock_session).get_bfd_sessions(device_id="1.1.1.1") + # Assert + self.assertEqual(answer, []) + + @patch('vmngclient.session.vManageSession') + def test_wait_for_bfd_session_up(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.bfd_session + # Act + answer = DeviceStateAPI(mock_session).wait_for_bfd_session_up(system_ip="1.1.1.1") + # Assert + self.assertIsNone(answer) + + @patch('vmngclient.session.vManageSession') + def test_wait_for_bfd_session_up_timeout(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.bfd_session_down + + # Act + def answer(): + return DeviceStateAPI(mock_session).wait_for_bfd_session_up( + system_ip="1.1.1.1", sleep_seconds=1, timeout_seconds=1 + ) + + # Assert + self.assertRaises(RetryError, answer) + + @patch('vmngclient.session.vManageSession') + def test_wait_for_device_state(self, mock_session): + # Arrange + mock_session.get_data.return_value = self.device + # Act + answer = DeviceStateAPI(mock_session).wait_for_device_state(device_id="1.1.1.1") + # Assert + self.assertTrue(answer) + + @patch('vmngclient.session.vManageSession') + def test_wait_for_device_state_unreachable(self, mock_session): + pass # TODO fix method before test diff --git a/vmngclient/tests/test_templates.py b/vmngclient/tests/test_templates.py index 8044f1353..927bda086 100644 --- a/vmngclient/tests/test_templates.py +++ b/vmngclient/tests/test_templates.py @@ -315,7 +315,7 @@ def test_delete_wrong_status(self, mock_session, mock_templates): # Arrage MockResponse = MagicMock() - MockResponse.status = 404 + MockResponse.ok = False mock_session.delete.return_value = MockResponse test_object = TemplateAPI(mock_session) diff --git a/vmngclient/utils/response.py b/vmngclient/utils/response.py index 8ebb32dd8..b9bbfa597 100644 --- a/vmngclient/utils/response.py +++ b/vmngclient/utils/response.py @@ -31,8 +31,12 @@ def response_debug(response: Response, headers: bool = False) -> str: request_body = str(request_body, encoding="utf-8") info = VManageResponseDebugInfo( request={"method": response.request.method, "url": response.request.url, "body": request_body}, - response={"status": response.status_code, "reason": response.reason, "text": response.text}, + response={"status": response.status_code, "reason": response.reason}, ) + if len(response.text) <= 1024: + info.response.update({"text": response.text}) + else: + info.response.update({"text(trimmed)": response.text[:128]}) if headers: info.request.update({"headers": dict(response.request.headers.items())}) info.response.update({"headers": dict(response.headers.items())})