diff --git a/.gitignore b/.gitignore index 96702e8..ebbdef4 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,5 @@ tests/.env dist/ build/ *.pyc -venv \ No newline at end of file +venv +debug.py \ No newline at end of file diff --git a/README.MD b/README.MD index ee42b77..4a2f7cc 100644 --- a/README.MD +++ b/README.MD @@ -20,7 +20,7 @@ pip install cloudshell-sandbox-rest ### Basic Usage ```python -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest.api import SandboxRestApiSession # pull in api user credentials CS_SERVER = "localhost" @@ -44,7 +44,7 @@ print(f"total components in sandbox: {len(components_response)}") Using the api session with a context manager "with" statement will log out and invalidate the token for you. ```python -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest.api import SandboxRestApiSession CS_SERVER = "localhost" CS_USER = "admin" @@ -75,7 +75,7 @@ Common use case is for admin to pull user token and start a session on their beh below. ```python -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest.api import SandboxRestApiSession # admin credentials CS_SERVER = "localhost" diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..0f517bc --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +log_cli = true +log_cli_level = 10 \ No newline at end of file diff --git a/requirements-dev.txt b/requirements-dev.txt index ff6f3e3..1fc7024 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -2,4 +2,5 @@ pytest python-dotenv flake8 pylint -pre-commit \ No newline at end of file +pre-commit +cloudshell-orch-core \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 471039c..660db9b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ requests>=2,<3 -abstract-http-client>=1,<2 \ No newline at end of file +abstract-http-client>=1,<2 +pydantic +tenacity \ No newline at end of file diff --git a/src/cloudshell/sandbox_rest/api.py b/src/cloudshell/sandbox_rest/api.py new file mode 100644 index 0000000..87af005 --- /dev/null +++ b/src/cloudshell/sandbox_rest/api.py @@ -0,0 +1,625 @@ +""" +Module that handles the Sandbox Rest API session +Handles login and wraps all api methods +Returns Pydantic BaseModel responses of returned JSON +""" +import json +import logging +import time +from dataclasses import asdict, dataclass +from typing import Callable, List + +from abstract_http_client.http_clients.requests_client import RequestsClient +from tenacity import RetryError, retry, retry_if_result, stop_after_delay, wait_fixed + +from cloudshell.sandbox_rest import exceptions, model +from pydantic.main import ValidationError + + +@dataclass +class CommandInputParam: + """ + param objects passed to sandbox / component command endpoints + sandbox global inputs, commands and resource commands all follow this generic name/value convention + """ + + name: str + value: str + + +class SandboxRestApiSession(RequestsClient): + """ + Python client for CloudShell Sandbox REST api + View swagger UI at http:///api/v2/explore for raw JSON response schema + """ + + def __init__( + self, + host: str, + username="", + password="", + domain="Global", + token="", + port=82, + logger: logging.Logger = None, + use_https=False, + ssl_verify=False, + proxies: dict = None, + show_insecure_warning=False + ): + """ Login to api and store headers for future requests """ + self.logger = logger + super().__init__(host, username, password, token, logger, port, use_https, ssl_verify, proxies, show_insecure_warning) + self.domain = domain + self._base_uri = "/api" + self._v2_base_uri = f"{self._base_uri}/v2" + self.login() + + def login(self, user="", password="", token="", domain="") -> None: + """ Called from init - can also be used to refresh session with credentials """ + self.user = user or self.user + self.password = password or self.password + self.token = token or self.token + self.domain = domain or self.domain + + if not self.domain: + raise ValueError("Domain must be passed to login") + + if not self.token: + if not self.user or not self.password: + raise ValueError("Login requires Token or Username / Password") + self.token = self._get_token_with_credentials(self.user, self.password, self.domain) + + self._set_auth_header_on_session() + + def logout(self) -> None: + if not self.token: + return + self.delete_token(self.token) + self.token = None + self._remove_auth_header_from_session() + + def _get_token_with_credentials(self, user_name: str, password: str, domain: str) -> str: + """ + Get token from credentials - extraneous quotes stripped off token string + """ + uri = f"{self._base_uri}/login" + data = {"username": user_name, "password": password, "domain": domain} + response = self.rest_service.request_put(uri, data) + + login_token = response.text[1:-1] + if not login_token: + err_msg = f"Invalid token. Token response {response.text}" + raise exceptions.SandboxRestAuthException(err_msg) + + return login_token + + def _set_auth_header_on_session(self): + self.rest_service.session.headers.update({"Authorization": f"Basic {self.token}"}) + + def _remove_auth_header_from_session(self): + self.rest_service.session.headers.pop("Authorization") + + def _validate_auth_header(self) -> None: + if not self.rest_service.session.headers.get("Authorization"): + raise exceptions.SandboxRestAuthException("No Authorization header currently set for session") + + def get_token_for_target_user(self, user_name: str) -> str: + """ + Get token for target user - remove extraneous quotes + """ + self._validate_auth_header() + uri = f"{self._base_uri}/token" + data = {"username": user_name} + response = self.rest_service.request_post(uri, data) + login_token = response.text[1:-1] + return login_token + + def delete_token(self, token_id: str) -> None: + self._validate_auth_header() + uri = f"{self._base_uri}/token/{token_id}" + return self.rest_service.request_delete(uri).text + + # BLUEPRINT GET REQUESTS + def get_blueprints(self) -> List[model.BlueprintDescription]: + self._validate_auth_header() + uri = f"{self._v2_base_uri}/blueprints" + blueprints_list = self.rest_service.request_get(uri).json() + return model.BlueprintDescription.list_to_models(blueprints_list) + + def get_blueprint_details(self, blueprint_id: str) -> model.BlueprintDescription: + """ + Get details of a specific blueprint + Can pass either blueprint name OR blueprint ID + """ + self._validate_auth_header() + uri = f"{self._v2_base_uri}/blueprints/{blueprint_id}" + details_dict = self.rest_service.request_get(uri).json() + return model.BlueprintDescription.dict_to_model(details_dict) + + # SANDBOX POST REQUESTS + def _start_sandbox( + self, + blueprint_id: str, + sandbox_name: str, + duration: str = None, + bp_params: List[CommandInputParam] = None, + permitted_users: List[str] = None, + polling_setup: bool = False, + max_polling_minutes: int = 20, + polling_frequency_seconds: int = 30, + polling_log_level: int = logging.DEBUG, + ) -> model.SandboxDetails: + """ internal implementation request to handle both regular and persistent sandbox requests """ + self._validate_auth_header() + if duration: + uri = f"{self._v2_base_uri}/blueprints/{blueprint_id}/start" + else: + uri = f"{self._v2_base_uri}/blueprints/{blueprint_id}/start-persistent" + + sandbox_name = sandbox_name if sandbox_name else self.get_blueprint_details(blueprint_id).name + + payload = { + "name": sandbox_name, + "permitted_users": permitted_users if permitted_users else [], + "params": [asdict(x) for x in bp_params] if bp_params else [], + } + if duration: + payload["duration"] = duration + + response_dict = self.rest_service.request_post(uri, payload).json() + sandbox_details = model.SandboxDetails.dict_to_model(response_dict) + if polling_setup: + return self.poll_sandbox_setup( + sandbox_details.id, max_polling_minutes, polling_frequency_seconds, polling_log_level + ) + + return sandbox_details + + def start_sandbox( + self, + blueprint_id: str, + sandbox_name: str = "", + duration: str = "PT2H0M", + bp_params: List[CommandInputParam] = None, + permitted_users: List[str] = None, + polling_setup: bool = False, + max_polling_minutes: int = 20, + polling_frequency_seconds: int = 30, + polling_log_level: int = logging.DEBUG, + ) -> model.SandboxDetails: + """ + Create a sandbox from the provided blueprint id + Duration format must be a valid 'ISO 8601'. (e.g 'PT23H' or 'PT4H2M') + """ + return self._start_sandbox( + blueprint_id=blueprint_id, + sandbox_name=sandbox_name, + duration=duration, + bp_params=bp_params, + permitted_users=permitted_users, + polling_setup=polling_setup, + max_polling_minutes=max_polling_minutes, + polling_frequency_seconds=polling_frequency_seconds, + polling_log_level=polling_log_level + ) + + def start_persistent_sandbox( + self, + blueprint_id: str, + sandbox_name: str = "", + bp_params: List[CommandInputParam] = None, + permitted_users: List[str] = None, + polling_setup: bool = False, + max_polling_minutes: int = 20, + polling_frequency_seconds: int = 30, + polling_log_level: int = logging.DEBUG, + ) -> model.SandboxDetails: + """ + Create a PERSISTENT sandbox from the provided blueprint id + Duration format must be a valid 'ISO 8601'. (e.g 'PT23H' or 'PT4H2M') + """ + return self._start_sandbox( + blueprint_id=blueprint_id, + sandbox_name=sandbox_name, + duration=None, + bp_params=bp_params, + permitted_users=permitted_users, + polling_setup=polling_setup, + max_polling_minutes=max_polling_minutes, + polling_frequency_seconds=polling_frequency_seconds, + polling_log_level=polling_log_level + ) + + def run_sandbox_command( + self, + sandbox_id: str, + command_name: str, + params: List[CommandInputParam] = None, + print_output=True, + polling_execution: bool = False, + max_polling_minutes: int = 20, + polling_frequency_seconds: int = 10, + polling_log_level: int = logging.DEBUG + ) -> model.SandboxCommandExecutionDetails: + """Run a sandbox level command""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/commands/{command_name}/start" + data = {"printOutput": print_output} + params = [asdict(x) for x in params] if params else [] + data["params"] = params + response_dict = self.rest_service.request_post(uri, data).json() + start_response = model.CommandStartResponse.dict_to_model(response_dict) + model_wrapped_command_params = [model.CommandParameterNameValue(x.name, x.value) for x in params] if params else [] + command_context = model.SandboxCommandContext(sandbox_id=sandbox_id, + command_name=command_name, + command_params=model_wrapped_command_params) + if polling_execution: + execution_details = self.poll_command_execution(execution_id=start_response.executionId, + max_polling_minutes=max_polling_minutes, + polling_frequency_seconds=polling_frequency_seconds, + log_level=polling_log_level) + else: + execution_details = self.get_execution_details(start_response.executionId) + + return model.SandboxCommandExecutionDetails(id=start_response.executionId, + status=execution_details.status, + supports_cancellation=execution_details.supports_cancellation, + started=execution_details.started, + ended=execution_details.ended, + output=execution_details.output, + command_context=command_context) + + def run_component_command( + self, + sandbox_id: str, + component_id: str, + command_name: str, + params: List[CommandInputParam] = None, + print_output: bool = True, + polling_execution: bool = False, + max_polling_minutes: int = 20, + polling_frequency_seconds: int = 10, + polling_log_level: int = logging.DEBUG + ) -> model.ComponentCommandExecutionDetails: + """Start a command on sandbox component""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}/commands/{command_name}/start" + data = {"printOutput": print_output} + params_dicts = [asdict(x) for x in params] if params else [] + data["params"] = params_dicts + response_dict = self.rest_service.request_post(uri, data).json() + start_response = model.CommandStartResponse.dict_to_model(response_dict) + component_details = self.get_sandbox_component_details(sandbox_id, component_id) + model_wrapped_command_params = [model.CommandParameterNameValue(x.name, x.value) for x in params] if params else [] + command_context = model.ComponentCommandContext(sandbox_id=sandbox_id, + command_name=command_name, + command_params=model_wrapped_command_params, + component_name=component_details.name, + component_id=component_details.id) + if polling_execution: + execution_details = self.poll_command_execution(execution_id=start_response.executionId, + max_polling_minutes=max_polling_minutes, + polling_frequency_seconds=polling_frequency_seconds, + log_level=polling_log_level) + else: + execution_details = self.get_execution_details(start_response.executionId) + + return model.ComponentCommandExecutionDetails(id=start_response.executionId, + status=execution_details.status, + supports_cancellation=execution_details.supports_cancellation, + started=execution_details.started, + ended=execution_details.ended, + output=execution_details.output, + command_context=command_context) + + def extend_sandbox(self, sandbox_id: str, duration: str) -> model.ExtendResponse: + """Extend the sandbox + :param str sandbox_id: Sandbox id + :param str duration: duration in ISO 8601 format (P1Y1M1DT1H1M1S = 1year, 1month, 1day, 1hour, 1min, 1sec) + :return: + """ + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/extend" + data = {"extended_time": duration} + response_dict = self.rest_service.request_post(uri, data).json() + return model.ExtendResponse.dict_to_model(response_dict) + + def stop_sandbox( + self, + sandbox_id: str, + poll_teardown=False, + max_polling_minutes: int = 20, + polling_frequency_seconds: int = 30, + polling_log_level: str = logging.DEBUG, + ) -> model.SandboxDetails: + """Stop the sandbox given sandbox id""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/stop" + response_dict = self.rest_service.request_post(uri).json() + + stop_response = model.StopSandboxResponse.dict_to_model(response_dict) + if "success" not in stop_response.result: + raise exceptions.TeardownFailedException(f"Failed to stop sandbox. result:\n{json.dumps(response_dict)}") + + if poll_teardown: + return self.poll_sandbox_teardown(sandbox_id, max_polling_minutes, polling_frequency_seconds, polling_log_level) + + # if not polling, give teardown request chance to propagate before getting status + time.sleep(3) + return self.get_sandbox_details(sandbox_id) + + # SANDBOX GET REQUESTS + def get_sandboxes(self, show_historic=False) -> List[model.SandboxDescriptionShort]: + """Get list of sandboxes""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes" + params = {"show_historic": "true" if show_historic else "false"} + response_list = self.rest_service.request_get(uri, params=params).json() + return model.SandboxDescriptionShort.list_to_models(response_list) + + def get_sandbox_details(self, sandbox_id: str) -> model.SandboxDetails: + """Get details of the given sandbox id""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}" + response_dict = self.rest_service.request_get(uri).json() + return model.SandboxDetails.dict_to_model(response_dict) + + def get_sandbox_activity( + self, + sandbox_id: str, + error_only=False, + since="", + from_event_id: int = None, + tail: int = None, + ) -> model.ActivityEventsResponse: + """ + Get list of sandbox activity + 'since' - format must be a valid 'ISO 8601'. (e.g 'PT23H' or 'PT4H2M') + 'from_event_id' - integer id of event where to start pulling results from + 'tail' - how many of the last entries you want to pull + 'error_only' - to filter for error events only + """ + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/activity" + params = {} + + if error_only: + params["error_only"] = "true" + if since: + params["since"] = since + if from_event_id: + params["from_event_id"] = from_event_id + if tail: + params["tail"] = tail + + response_dict = self.rest_service.request_get(uri, params=params).json() + return model.ActivityEventsResponse.dict_to_model(response_dict) + + def get_sandbox_commands(self, sandbox_id: str) -> List[model.Command]: + """Get list of sandbox commands""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/commands" + response_list = self.rest_service.request_get(uri).json() + return model.Command.list_to_models(response_list) + + def get_sandbox_command_details(self, sandbox_id: str, command_name: str) -> model.Command: + """Get details of specific sandbox command""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/commands/{command_name}" + response_dict = self.rest_service.request_get(uri).json() + return model.Command.dict_to_model(response_dict) + + def get_sandbox_components(self, sandbox_id: str) -> List[model.SandboxComponentFull]: + """Get list of sandbox components""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components" + response_list = self.rest_service.request_get(uri).json() + return model.SandboxComponentFull.list_to_models(response_list) + + def get_sandbox_component_details(self, sandbox_id: str, component_id: str) -> model.SandboxComponentFull: + """Get details of components in sandbox""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}" + response_dict = self.rest_service.request_get(uri).json() + return model.SandboxComponentFull.dict_to_model(response_dict) + + def get_sandbox_component_commands(self, sandbox_id: str, component_id: str) -> List[model.Command]: + """Get list of commands for a particular component in sandbox""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}/commands" + response_list = self.rest_service.request_get(uri).json() + return model.Command.list_to_models(response_list) + + def get_sandbox_component_command_details( + self, sandbox_id: str, component_id: str, command: str + ) -> model.CommandExecutionDetails: + """Get details of a command of sandbox component""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}/commands/{command}" + response_dict = self.rest_service.request_get(uri).json() + return model.CommandExecutionDetails.dict_to_model(response_dict) + + def get_sandbox_instructions(self, sandbox_id: str) -> str: + """ Pull the instructions text of sandbox """ + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/instructions" + return self.rest_service.request_get(uri).text + + def get_sandbox_output( + self, + sandbox_id: str, + tail: int = None, + from_entry_id: int = None, + since: str = None, + ) -> model.SandboxOutput: + """Get list of sandbox output""" + self._validate_auth_header() + uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/output" + params = {} + if tail: + params["tail"] = tail + if from_entry_id: + params["from_entry_id"] = from_entry_id + if since: + params["since"] = since + + response_dict = self.rest_service.request_get(uri, params=params).json() + return model.SandboxOutput.dict_to_model(response_dict) + + # EXECUTIONS + def get_execution_details(self, execution_id: str) -> model.CommandExecutionDetails: + self._validate_auth_header() + uri = f"{self._v2_base_uri}/executions/{execution_id}" + details_dict = self.rest_service.request_get(uri).json() + try: + execution_details = model.CommandExecutionDetails.dict_to_model(details_dict) + except ValidationError as e: + err_msg = f"Validation error on execution '{execution_id}. Raw JSON:\n{json.dumps(details_dict, indent=4)}" + if self.logger: + self.logger.error(err_msg) + raise + return execution_details + + def delete_execution(self, execution_id: str) -> dict: + """ + API returns dict with single key on successful deletion of execution + {"result": "success"} + """ + self._validate_auth_header() + uri = f"{self._v2_base_uri}/executions/{execution_id}" + response_dict = self.rest_service.request_delete(uri).json() + if not response_dict["result"] == "success": + raise exceptions.SandboxRestException( + f"Failed execution deletion of id {execution_id}\n" f"{json.dumps(response_dict, indent=4)}" + ) + return response_dict + + # Polling + def _poll_orchestration_state( + self, + orchestration_type: str, + reservation_id: str, + polling_func: Callable, + max_polling_minutes: int, + polling_frequency_seconds: int, + log_level=logging.DEBUG, + ) -> model.SandboxDetails: + """ Create blocking polling process """ + + def poll_and_log(sb_details: model.SandboxDetails): + if self.logger: + polling_msg = f"Polling {orchestration_type} for sandbox '{sb_details.id}'. State: {sb_details.state}..." + self.logger.log(log_level, polling_msg) + return polling_func(sb_details) + + @retry( + retry=retry_if_result(poll_and_log), + wait=wait_fixed(polling_frequency_seconds), + stop=stop_after_delay(max_polling_minutes * 60), + ) + def retry_poll_sandbox_details(): + return self.get_sandbox_details(reservation_id) + + try: + sandbox_details = retry_poll_sandbox_details() + except RetryError: + raise exceptions.OrchestrationPollingTimeout(f"Sandbox Polling timed out after {max_polling_minutes} minutes") + return sandbox_details + + def poll_sandbox_setup( + self, reservation_id: str, max_polling_minutes=20, polling_frequency_seconds=30, log_level: int = logging.DEBUG + ) -> model.SandboxDetails: + """ poll setup until completion """ + sandbox_details = self._poll_orchestration_state( + orchestration_type="SETUP", + reservation_id=reservation_id, + polling_func=_should_keep_polling_setup, + max_polling_minutes=max_polling_minutes, + polling_frequency_seconds=polling_frequency_seconds, + log_level=log_level, + ) + setup_state = sandbox_details.state + sandbox_id = sandbox_details.id + if setup_state == model.SandboxStates.ERROR: + error_activity = self.get_sandbox_activity(sandbox_id, error_only=True) + raise exceptions.SetupFailedException( + f"Sandbox setup failed - sandbox id: '{sandbox_id}'", error_events=error_activity.events + ) + return sandbox_details + + def poll_sandbox_teardown( + self, reservation_id: str, max_polling_minutes=20, polling_frequency_seconds=30, log_level: int = logging.DEBUG + ) -> model.SandboxDetails: + """ poll teardown until completion """ + latest_event_request = self.get_sandbox_activity(reservation_id, tail=1).events + latest_event_id = latest_event_request[0].id if latest_event_request else 0 + sandbox_details = self._poll_orchestration_state( + orchestration_type="TEARDOWN", + reservation_id=reservation_id, + polling_func=_should_keep_polling_teardown, + max_polling_minutes=max_polling_minutes, + polling_frequency_seconds=polling_frequency_seconds, + log_level=log_level, + ) + teardown_error_events = self.get_sandbox_activity( + reservation_id, error_only=True, from_event_id=latest_event_id + ).events + if teardown_error_events: + raise exceptions.TeardownFailedException( + f"Failed teardown - sandbox id: '{reservation_id}'", error_events=teardown_error_events + ) + return sandbox_details + + def poll_command_execution(self, + execution_id: str, + max_polling_minutes=10, + polling_frequency_seconds=30, + log_level=logging.DEBUG, + ) -> model.CommandExecutionDetails: + """ Create blocking polling process """ + + def poll_and_log(execution_data: model.CommandExecutionDetails) -> bool: + if self.logger: + polling_msg = f"Polling execution '{execution_data.id}'. Status: '{execution_data.status}'..." + self.logger.log(log_level, polling_msg) + return _should_keep_polling_execution(execution_data) + + @retry( + retry=retry_if_result(poll_and_log), + wait=wait_fixed(polling_frequency_seconds), + stop=stop_after_delay(max_polling_minutes * 60), + ) + def retry_poll_execution(): + return self.get_execution_details(execution_id) + + pre_poll_execution_details = self.get_execution_details(execution_id) + try: + execution_details = retry_poll_execution() + except RetryError: + err_msg = f"Execution '{pre_poll_execution_details.id}' timed out after {max_polling_minutes} minutes" + raise exceptions.CommandPollingTimeout(err_msg) + return execution_details + + +# Polling Helpers +def _should_keep_polling_setup(sandbox_details: model.SandboxDetails) -> bool: + """ if still in setup keep polling """ + post_setup_states = model.SandboxStates.get_post_setup_states() + if sandbox_details.state not in post_setup_states: + return True + return False + + +def _should_keep_polling_teardown(sandbox_details: model.SandboxDetails) -> bool: + """ if in teardown keep polling """ + if sandbox_details.state == model.SandboxStates.TEARDOWN: + return True + return False + + +def _should_keep_polling_execution(execution_data: model.CommandExecutionDetails) -> bool: + unfinished_states = model.CommandExecutionStates.get_incomplete_execution_states() + if execution_data.status in unfinished_states: + return True + return False diff --git a/src/cloudshell/sandbox_rest/async_commands.py b/src/cloudshell/sandbox_rest/async_commands.py new file mode 100644 index 0000000..53e1673 --- /dev/null +++ b/src/cloudshell/sandbox_rest/async_commands.py @@ -0,0 +1,7 @@ +""" +module to run sandbox actions in parallel using asyncio module +""" + + +class AsyncCommandExecutor: + pass diff --git a/src/cloudshell/sandbox_rest/components.py b/src/cloudshell/sandbox_rest/components.py new file mode 100644 index 0000000..6b42032 --- /dev/null +++ b/src/cloudshell/sandbox_rest/components.py @@ -0,0 +1,84 @@ +""" +Component helpers for filtering and sorting the sandbox components +""" +from enum import Enum +from typing import List + +from cloudshell.sandbox_rest import model +from cloudshell.sandbox_rest.api import SandboxRestApiSession + + +class ComponentTypes(str, Enum): + app_type = "Application" + resource_type = "Resource" + service_type = "Service" + + +class AppLifeCycleTypes(str, Enum): + deployed = "Deployed" + un_deployed = "Undeployed" + + +class AttributeTypes(str, Enum): + boolean = "boolean" + password = "password" + string = "string" + numeric = "numeric" + + +class SandboxComponents: + def __init__(self, components: List[model.SandboxComponentFull] = None): + self.all_components = components or [] + + def refresh_components(self, api: SandboxRestApiSession, sandbox_id: str) -> None: + self.all_components = api.get_sandbox_components(sandbox_id) + + def _filter_components_by_type(self, component_type: ComponentTypes) -> List[model.SandboxComponentFull]: + """ accepts both short info components and full info """ + return [component for component in self.all_components if component.component_type == component_type] + + def _filter_app_by_lifecycle(self, lifecycle_type: AppLifeCycleTypes) -> List[model.SandboxComponentFull]: + return [component for component in self.all_components if component.component_type == ComponentTypes.app_type + and component.app_lifecycle == lifecycle_type] + + @property + def resources(self): + return self._filter_components_by_type(ComponentTypes.service_type) + + @property + def services(self): + return self._filter_components_by_type(ComponentTypes.service_type) + + @property + def deployed_apps(self): + return self._filter_app_by_lifecycle(AppLifeCycleTypes.deployed) + + @property + def un_deployed_apps(self): + return self._filter_app_by_lifecycle(AppLifeCycleTypes.un_deployed) + + def filter_by_model(self, component_model: str) -> List[dict]: + """ + Component Model / Shell Template + ex: 'Juniper JunOS Switch Shell 2G' + """ + return [component for component in self.all_components if component.component_type == component_model] + + def filter_by_attr(self, attr_name: str, attr_val: str) -> List[model.SandboxComponentFull]: + """ attr name is shell agnostic, no namespacing """ + components = [] + for component in self.all_components: + for attr in component.attributes: + if attr.name.endswith(attr_name) and attr.value == attr_val: + components.append(component) + return components + + def filter_by_boolean_attr(self, attr_name: str) -> List[dict]: + """ attr name is shell agnostic, no namespacing """ + components = [] + for component in self.all_components: + for attr in component.attributes: + if attr.type == AttributeTypes.boolean and attr.name.endswith(attr_name) and attr.value == "True": + components.append(component) + return components + diff --git a/src/cloudshell/sandbox_rest/default_logger.py b/src/cloudshell/sandbox_rest/default_logger.py new file mode 100644 index 0000000..59049ec --- /dev/null +++ b/src/cloudshell/sandbox_rest/default_logger.py @@ -0,0 +1,29 @@ +import logging + + +DEFAULT_LOGGER_NAME = "Sandbox Rest Logger" +DEFAULT_FORMAT = '%(asctime)s:%(filename)s:%(lineno)d %(message)s' + + +def set_up_default_logger(logger_name=DEFAULT_LOGGER_NAME, + log_level=logging.INFO, + log_to_file=False, + log_file_path=".", + log_format=DEFAULT_FORMAT): + logger = logging.getLogger(logger_name) + logger.setLevel(log_level) + + formatter = logging.Formatter(log_format) + + stream_handler = logging.StreamHandler() + stream_handler.setLevel(logging.INFO) + stream_handler.setFormatter(formatter) + logger.addHandler(stream_handler) + + if log_to_file: + file_handler = logging.FileHandler(log_file_path) + file_handler.setLevel(logging.INFO) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + return logger diff --git a/src/cloudshell/sandbox_rest/exceptions.py b/src/cloudshell/sandbox_rest/exceptions.py index befbf51..443cafb 100644 --- a/src/cloudshell/sandbox_rest/exceptions.py +++ b/src/cloudshell/sandbox_rest/exceptions.py @@ -1,6 +1,46 @@ +from typing import List + +from cloudshell.sandbox_rest import model + + class SandboxRestException(Exception): """ Base Exception Class inside Rest client class """ class SandboxRestAuthException(SandboxRestException): """ Failed login action """ + + +class OrchestrationPollingTimeout(SandboxRestException): + pass + + +class CommandPollingTimeout(SandboxRestException): + pass + + +class FailedOrchestrationException(SandboxRestException): + def __init__(self, message: str, error_events: List[model.SandboxEvent] = None): + self.message = message + self.error_events = error_events or [] + super().__init__(message) + + def events_to_json(self): + return model.models_to_json(self.error_events) + + def __str__(self): + return f"{self.message}\n{self.events_to_json()}" + + +class SetupFailedException(FailedOrchestrationException): + def __init__(self, message: str, error_events: List[model.SandboxEvent] = None): + super().__init__(message, error_events) + + +class TeardownFailedException(FailedOrchestrationException): + def __init__(self, message: str, error_events: List[model.SandboxEvent] = None): + super().__init__(message, error_events) + + +class CommandExecutionFailed(SandboxRestException): + pass diff --git a/src/cloudshell/sandbox_rest/model.py b/src/cloudshell/sandbox_rest/model.py new file mode 100644 index 0000000..9eaf8f3 --- /dev/null +++ b/src/cloudshell/sandbox_rest/model.py @@ -0,0 +1,336 @@ +""" +All pydantic BaseModel class representations of Sandbox API responses +""" +from __future__ import annotations + +import json +from enum import Enum +from typing import TYPE_CHECKING, List, Optional + +from pydantic import BaseModel + +RESPONSE_DICT_KEY = "response_dict" + +# added for dev intellisense: https://stackoverflow.com/a/71257588 +if TYPE_CHECKING: + from dataclasses import dataclass as _basemodel_decorator +else: + + def _basemodel_decorator(func): + return func + + +class SandboxApiBaseModel(BaseModel): + """ Base Model for all other classes. Defines useful helper methods """ + + class Config: + use_enum_values = True + + response_dict: dict = None + """ this attribute will cache the original response before loading into model """ + + @classmethod + def dict_to_model(cls, response_dict: dict) -> SandboxApiBaseModel: + """ calls parse_obj, but additionally caches the response dict """ + wrapped_item = cls.parse_obj(response_dict) + wrapped_item.response_dict = response_dict + return wrapped_item + + @classmethod + def list_to_models(cls, response_list: List[dict]) -> List[SandboxApiBaseModel]: + return [cls.dict_to_model(dict_item) for dict_item in response_list] + + def pretty_json(self, indent=4, exclude_response=True): + excluded_key_set = set() + if exclude_response: + excluded_key_set.add(RESPONSE_DICT_KEY) + return self.json(indent=indent, exclude=excluded_key_set) + + +def models_to_json(model_list: List[SandboxApiBaseModel]) -> str: + """ helper method to be used to convert a list of models to a list of dicts and dump to json """ + list_of_dicts = [x.dict() for x in model_list] + excluded_keys = [RESPONSE_DICT_KEY] + updated_list = [{k: v for k, v in curr_dict if k not in excluded_keys} for curr_dict in list_of_dicts] + return json.dumps(updated_list, indent=4) + + +class BlueprintAvailabilityStates(str, Enum): + AVAILABLE = "Available Now" + NOT_AVAILABLE = "Not Available" + + +class SandboxStates(str, Enum): + PENDING = "Pending" + BEFORE_SETUP = "BeforeSetup" + PENDING_SETUP = "Pending Setup" + RUNNING_SETUP = "Setup" + READY = "Ready" + ERROR = "Error" + TEARDOWN = "Teardown" + ENDED = "Ended" + + @classmethod + def get_post_setup_states(cls) -> List: + return [cls.READY, cls.ERROR, cls.TEARDOWN, cls.ENDED] + + @classmethod + def get_active_states(cls) -> List: + return [cls.READY, cls.ERROR] + + @classmethod + def get_post_active_states(cls) -> List: + return [cls.TEARDOWN, cls.ENDED] + + +class SetupStages(str, Enum): + NONE = "None" + PROVISIONING = "Provisioning" + CONNECTIVITY = "Connectivity" + CONFIGURATION = "Configuration" + ENDED = "Ended" + + +class SandboxEventTypes(str, Enum): + SUCCESS = "success" + ERROR = "error" + + +class CommandParameterTypes(str, Enum): + STRING = "String" + NUMERIC = "Numeric" + LOOKUP = "Lookup" + + +class CommandExecutionStates(str, Enum): + PENDING = "Pending" + RUNNING = "Running" + STOPPING = "Stopping" + CANCELLED = "Cancelled" + COMPLETE = "Completed" + FAILED = "Failed" + + @classmethod + def get_incomplete_execution_states(cls) -> List: + return [cls.PENDING, cls.RUNNING] + + +@_basemodel_decorator +class BlueprintInput(SandboxApiBaseModel): + name: Optional[str] + type: Optional[str] + possible_values: Optional[List[str]] + default_value: Optional[str] + + +@_basemodel_decorator +class BlueprintDescription(SandboxApiBaseModel): + id: Optional[str] + name: Optional[str] + categories: Optional[List[str]] + description: Optional[str] + params: Optional[List[BlueprintInput]] + availability: Optional[BlueprintAvailabilityStates] + """ The availability of blueprint: ['Available Now', 'Not Available'] """ + estimated_setup_duration: Optional[str] + """ Estimated blueprint setup duration. Uses'ISO 8601' Standard. (e.g 'PT23H' or 'PT4H2M') """ + + +@_basemodel_decorator +class SandboxComponentBasic(SandboxApiBaseModel): + id: Optional[str] + name: Optional[str] + type: Optional[str] + component_type: Optional[str] + description: Optional[str] + address: Optional[str] + app_lifecycle: Optional[str] + + +@_basemodel_decorator +class SandboxDetails(SandboxApiBaseModel): + id: Optional[str] + blueprint_id: Optional[str] + type: Optional[str] + name: Optional[str] + permitted_users: Optional[List[str]] + description: Optional[str] + start_time: Optional[str] + end_time: Optional[str] + params: Optional[List[BlueprintInput]] + components: Optional[List[SandboxComponentBasic]] + state: Optional[SandboxStates] + setup_stage: Optional[SetupStages] + + def components_to_json(self): + return models_to_json(self.components) + + +@_basemodel_decorator +class BlueprintReference(SandboxApiBaseModel): + id: Optional[str] + name: Optional[str] + + +@_basemodel_decorator +class SandboxDescriptionShort(SandboxApiBaseModel): + id: Optional[str] + name: Optional[str] + blueprint: Optional[BlueprintReference] + description: Optional[str] + state: Optional[SandboxStates] + + +@_basemodel_decorator +class SandboxEvent(SandboxApiBaseModel): + id: Optional[int] + event_type: Optional[SandboxEventTypes] + event_text: Optional[str] + output: Optional[str] + time: Optional[str] + """ Event time in 'ISO 8601' Standard. (e.g '2000-12-31T23:59:60Z') """ + + +@_basemodel_decorator +class ActivityEventsResponse(SandboxApiBaseModel): + num_returned_events: Optional[int] + more_pages: Optional[bool] + next_event_id: Optional[int] + events: Optional[List[SandboxEvent]] + + def events_to_json(self) -> str: + return models_to_json(self.events) + + +@_basemodel_decorator +class CommandParameterDetails(SandboxApiBaseModel): + name: Optional[str] + description: Optional[str] + type: Optional[CommandParameterTypes] + possibleValues: Optional[List[str]] + defaultValue: Optional[str] + mandatory: Optional[bool] + + +@_basemodel_decorator +class CommandParameterNameValue(SandboxApiBaseModel): + name: str + value: str + + +@_basemodel_decorator +class CommandExecution(SandboxApiBaseModel): + id: Optional[str] + status: Optional[CommandExecutionStates] + supports_cancellation: Optional[bool] + + +@_basemodel_decorator +class Command(SandboxApiBaseModel): + name: Optional[str] + description: Optional[str] + params: Optional[List[CommandParameterDetails]] + executions: Optional[List[CommandExecution]] + + +@_basemodel_decorator +class CommandStartResponse(SandboxApiBaseModel): + executionId: Optional[str] + supports_cancellation: Optional[bool] + + +@_basemodel_decorator +class CommandExecutionDetails(SandboxApiBaseModel): + id: Optional[str] + status: Optional[CommandExecutionStates] + supports_cancellation: Optional[bool] + started: Optional[str] + ended: Optional[str] + output: Optional[str] + + +@_basemodel_decorator +class ComponentAttribute(SandboxApiBaseModel): + type: Optional[str] + name: Optional[str] + value: Optional[str] + + +@_basemodel_decorator +class ConnectionInterface(SandboxApiBaseModel): + name: Optional[str] + url: Optional[str] + + +@_basemodel_decorator +class LinkMetadata(SandboxApiBaseModel): + href: Optional[str] + method: Optional[str] + + +@_basemodel_decorator +class SandboxComponentFull(SandboxApiBaseModel): + id: Optional[str] + name: Optional[str] + type: Optional[str] + component_type: Optional[str] + description: Optional[str] + address: Optional[str] + app_lifecycle: Optional[str] + attributes: Optional[List[ComponentAttribute]] + connection_interfaces: Optional[List[ConnectionInterface]] + _links: Optional[List[LinkMetadata]] + + +@_basemodel_decorator +class ExtendResponse(SandboxApiBaseModel): + id: Optional[str] + name: Optional[str] + start_time: Optional[str] + end_time: Optional[str] + remaining_time: Optional[str] + + +@_basemodel_decorator +class SandboxOutputEntry(SandboxApiBaseModel): + id: Optional[str] + text: Optional[str] + time: Optional[str] + """ Event time in 'ISO 8601' Standard. (e.g '2000-12-31T23:59:60Z' """ + + +@_basemodel_decorator +class SandboxOutput(SandboxApiBaseModel): + number_of_returned_entries: Optional[int] + next_entry_id: Optional[str] + more_pages: Optional[bool] + entries: Optional[List[SandboxOutputEntry]] + + +# ===== Custom Data Models NOT returned from endpoint responses +@_basemodel_decorator +class StopSandboxResponse(SandboxApiBaseModel): + result: Optional[str] = "success" + + +@_basemodel_decorator +class SandboxCommandContext(SandboxApiBaseModel): + sandbox_id: Optional[str] + command_name: Optional[str] + command_params: Optional[List[CommandParameterNameValue]] + + +@_basemodel_decorator +class ComponentCommandContext(SandboxCommandContext): + component_name: Optional[str] + component_id: Optional[str] + + +@_basemodel_decorator +class SandboxCommandExecutionDetails(CommandExecutionDetails): + command_context: Optional[SandboxCommandContext] + + +@_basemodel_decorator +class ComponentCommandExecutionDetails(CommandExecutionDetails): + command_context: Optional[ComponentCommandContext] diff --git a/src/cloudshell/sandbox_rest/sandbox_api.py b/src/cloudshell/sandbox_rest/sandbox_api.py deleted file mode 100644 index 63a4970..0000000 --- a/src/cloudshell/sandbox_rest/sandbox_api.py +++ /dev/null @@ -1,346 +0,0 @@ -import json -import logging -from dataclasses import asdict, dataclass -from typing import List - -from abstract_http_client.http_clients.requests_client import RequestsClient - -from cloudshell.sandbox_rest.exceptions import SandboxRestAuthException, SandboxRestException - - -@dataclass -class InputParam: - """ - param objects passed to sandbox / component command endpoints - sandbox global inputs, commands and resource commands all follow this generic name/value convention - """ - - name: str - value: str - - -class SandboxRestApiSession(RequestsClient): - """ - Python wrapper for CloudShell Sandbox API - View http:///api/v2/explore to see schemas of return json values - """ - - def __init__( - self, - host: str, - username="", - password="", - domain="Global", - token="", - port=82, - logger: logging.Logger = None, - use_https=False, - ssl_verify=False, - proxies: dict = None, - show_insecure_warning=False, - ): - """ Login to api and store headers for future requests """ - super().__init__(host, username, password, token, logger, port, use_https, ssl_verify, proxies, show_insecure_warning) - self._base_uri = "/api" - self._v2_base_uri = f"{self._base_uri}/v2" - self.domain = domain - self.login() - - def login(self, user="", password="", token="", domain="") -> None: - """ Called from init - can also be used to refresh session with credentials """ - self.user = user or self.user - self.password = password or self.password - self.token = token or self.token - self.domain = domain or self.domain - - if not self.domain: - raise ValueError("Domain must be passed to login") - - if not self.token: - if not self.user or not self.password: - raise ValueError("Login requires Token or Username / Password") - self.token = self._get_token_with_credentials(self.user, self.password, self.domain) - - self._set_auth_header_on_session() - - def logout(self) -> None: - if not self.token: - return - self.delete_token(self.token) - self.token = None - self._remove_auth_header_from_session() - - def _get_token_with_credentials(self, user_name: str, password: str, domain: str) -> str: - """ - Get token from credentials - extraneous quotes stripped off token string - """ - uri = f"{self._base_uri}/login" - data = {"username": user_name, "password": password, "domain": domain} - response = self.rest_service.request_put(uri, data) - - login_token = response.text[1:-1] - if not login_token: - raise SandboxRestAuthException(f"Invalid token. Token response {response.text}") - - return login_token - - def _set_auth_header_on_session(self): - self.rest_service.session.headers.update({"Authorization": f"Basic {self.token}"}) - - def _remove_auth_header_from_session(self): - self.rest_service.session.headers.pop("Authorization") - - def _validate_auth_header(self) -> None: - if not self.rest_service.session.headers.get("Authorization"): - raise SandboxRestAuthException("No Authorization header currently set for session") - - def get_token_for_target_user(self, user_name: str) -> str: - """ - Get token for target user - remove extraneous quotes - """ - self._validate_auth_header() - uri = f"{self._base_uri}/token" - data = {"username": user_name} - response = self.rest_service.request_post(uri, data) - login_token = response.text[1:-1] - return login_token - - def delete_token(self, token_id: str) -> None: - self._validate_auth_header() - uri = f"{self._base_uri}/token/{token_id}" - return self.rest_service.request_delete(uri).text - - # SANDBOX POST REQUESTS - def start_sandbox( - self, - blueprint_id: str, - sandbox_name="", - duration="PT2H0M", - bp_params: List[InputParam] = None, - permitted_users: List[str] = None, - ) -> dict: - """ - Create a sandbox from the provided blueprint id - Duration format must be a valid 'ISO 8601'. (e.g 'PT23H' or 'PT4H2M') - """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/blueprints/{blueprint_id}/start" - sandbox_name = sandbox_name if sandbox_name else self.get_blueprint_details(blueprint_id)["name"] - - data = { - "duration": duration, - "name": sandbox_name, - "permitted_users": permitted_users if permitted_users else [], - "params": [asdict(x) for x in bp_params] if bp_params else [], - } - - return self.rest_service.request_post(uri, data).json() - - def start_persistent_sandbox( - self, - blueprint_id: str, - sandbox_name="", - bp_params: List[InputParam] = None, - permitted_users: List[str] = None, - ) -> dict: - """ Create a persistent sandbox from the provided blueprint id """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/blueprints/{blueprint_id}/start-persistent" - - sandbox_name = sandbox_name if sandbox_name else self.get_blueprint_details(blueprint_id)["name"] - data = { - "name": sandbox_name, - "permitted_users": permitted_users if permitted_users else [], - "params": [asdict(x) for x in bp_params] if bp_params else [], - } - - return self.rest_service.request_post(uri, data).json() - - def run_sandbox_command( - self, - sandbox_id: str, - command_name: str, - params: List[InputParam] = None, - print_output=True, - ) -> dict: - """Run a sandbox level command""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/commands/{command_name}/start" - data = {"printOutput": print_output} - params = [asdict(x) for x in params] if params else [] - data["params"] = params - return self.rest_service.request_post(uri, data).json() - - def run_component_command( - self, - sandbox_id: str, - component_id: str, - command_name: str, - params: List[InputParam] = None, - print_output: bool = True, - ) -> dict: - """Start a command on sandbox component""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}/commands/{command_name}/start" - data = {"printOutput": print_output} - params = [asdict(x) for x in params] if params else [] - data["params"] = params - return self.rest_service.request_post(uri, data).json() - - def extend_sandbox(self, sandbox_id: str, duration: str) -> dict: - """Extend the sandbox - :param str sandbox_id: Sandbox id - :param str duration: duration in ISO 8601 format (P1Y1M1DT1H1M1S = 1year, 1month, 1day, 1hour, 1min, 1sec) - :return: - """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/extend" - data = {"extended_time": duration} - return self.rest_service.request_post(uri, data).json() - - def stop_sandbox(self, sandbox_id: str) -> None: - """Stop the sandbox given sandbox id""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/stop" - return self.rest_service.request_post(uri).json() - - # SANDBOX GET REQUESTS - def get_sandboxes(self, show_historic=False) -> list: - """Get list of sandboxes""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes" - params = {"show_historic": "true" if show_historic else "false"} - return self.rest_service.request_get(uri, params=params).json() - - def get_sandbox_details(self, sandbox_id: str) -> dict: - """Get details of the given sandbox id""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}" - return self.rest_service.request_get(uri).json() - - def get_sandbox_activity( - self, - sandbox_id: str, - error_only=False, - since="", - from_event_id: int = None, - tail: int = None, - ) -> dict: - """ - Get list of sandbox activity - 'since' - format must be a valid 'ISO 8601'. (e.g 'PT23H' or 'PT4H2M') - 'from_event_id' - integer id of event where to start pulling results from - 'tail' - how many of the last entries you want to pull - 'error_only' - to filter for error events only - """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/activity" - params = {} - - if error_only: - params["error_only"] = error_only - if since: - params["since"] = since - if from_event_id: - params["from_event_id"] = from_event_id - if tail: - params["tail"] = tail - - return self.rest_service.request_get(uri, params=params).json() - - def get_sandbox_commands(self, sandbox_id: str) -> list: - """Get list of sandbox commands""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/commands" - return self.rest_service.request_get(uri).json() - - def get_sandbox_command_details(self, sandbox_id: str, command_name: str) -> dict: - """Get details of specific sandbox command""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/commands/{command_name}" - return self.rest_service.request_get(uri).json() - - def get_sandbox_components(self, sandbox_id: str) -> list: - """Get list of sandbox components""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components" - return self.rest_service.request_get(uri).json() - - def get_sandbox_component_details(self, sandbox_id: str, component_id: str) -> dict: - """Get details of components in sandbox""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}" - return self.rest_service.request_get(uri).json() - - def get_sandbox_component_commands(self, sandbox_id: str, component_id: str) -> list: - """Get list of commands for a particular component in sandbox""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}/commands" - return self.rest_service.request_get(uri).json() - - def get_sandbox_component_command_details(self, sandbox_id: str, component_id: str, command: str) -> dict: - """Get details of a command of sandbox component""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/components/{component_id}/commands/{command}" - return self.rest_service.request_get(uri).json() - - def get_sandbox_instructions(self, sandbox_id: str) -> str: - """ Pull the instructions text of sandbox """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/instructions" - return self.rest_service.request_get(uri).json() - - def get_sandbox_output( - self, - sandbox_id: str, - tail: int = None, - from_entry_id: int = None, - since: str = None, - ) -> dict: - """Get list of sandbox output""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/sandboxes/{sandbox_id}/output" - params = {} - if tail: - params["tail"] = tail - if from_entry_id: - params["from_entry_id"] = from_entry_id - if since: - params["since"] = since - - return self.rest_service.request_get(uri, params=params).json() - - # BLUEPRINT GET REQUESTS - def get_blueprints(self) -> list: - """Get list of blueprints""" - self._validate_auth_header() - uri = f"{self._v2_base_uri}/blueprints" - return self.rest_service.request_get(uri).json() - - def get_blueprint_details(self, blueprint_id: str) -> dict: - """ - Get details of a specific blueprint - Can pass either blueprint name OR blueprint ID - """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/blueprints/{blueprint_id}" - return self.rest_service.request_get(uri).json() - - # EXECUTIONS - def get_execution_details(self, execution_id: str) -> dict: - self._validate_auth_header() - uri = f"{self._v2_base_uri}/executions/{execution_id}" - return self.rest_service.request_get(uri).json() - - def delete_execution(self, execution_id: str) -> None: - """ - API returns dict with single key on successful deletion of execution - {"result": "success"} - """ - self._validate_auth_header() - uri = f"{self._v2_base_uri}/executions/{execution_id}" - response_dict = self.rest_service.request_delete(uri).json() - if not response_dict["result"] == "success": - raise SandboxRestException( - f"Failed execution deletion of id {execution_id}\n" f"{json.dumps(response_dict, indent=4)}" - ) diff --git a/src/cloudshell/sandbox_rest/sandbox_context.py b/src/cloudshell/sandbox_rest/sandbox_context.py new file mode 100644 index 0000000..9817e1c --- /dev/null +++ b/src/cloudshell/sandbox_rest/sandbox_context.py @@ -0,0 +1,162 @@ +""" +Sandbox controller context manager to manage the lifecyle of setup and teardown +- start sandbox on context enter, end sandbox on context exit +- cache setup / teardown duration and errors +- components object member +- async commands executor member +""" +import logging +from typing import List + +from cloudshell.sandbox_rest.api import SandboxRestApiSession, CommandInputParam +from cloudshell.sandbox_rest import exceptions +from cloudshell.sandbox_rest import model +from cloudshell.sandbox_rest.async_commands import AsyncCommandExecutor +from cloudshell.sandbox_rest.components import SandboxComponents +from dataclasses import dataclass +from timeit import default_timer + + +@dataclass +class SandboxStartRequest: + blueprint_id: str + sandbox_name: str = "" + duration: str = "PT2H0M" + bp_params: List[CommandInputParam] = None + permitted_users: List[str] = None + max_polling_minutes: int = 20 + polling_frequency_seconds: int = 30 + polling_log_level: int = logging.DEBUG + + +@dataclass +class TeardownSettings: + max_polling_minutes: int = 20 + polling_frequency_seconds: int = 30 + polling_log_level: int = logging.DEBUG + + +@dataclass +class OrchMetadata: + setup_duration_seconds: int = None + teardown_duration_seconds: int = None + setup_errors: List[model.SandboxEvent] = None + teardown_errors: List[model.SandboxEvent] = None + + +class SandboxControllerContext: + def __init__(self, api: SandboxRestApiSession, sandbox_id: str = None, sandbox_request: SandboxStartRequest = None, + teardown_settings = TeardownSettings(), logger: logging.Logger = None): + self.api = api + self.sandbox_id = sandbox_id + self.sandbox_request = sandbox_request + self.teardown_settings = teardown_settings + self.logger = logger + self.components = SandboxComponents() + self.async_executor = AsyncCommandExecutor() + self.orch_metadata = OrchMetadata() + self._handle_init() + + def _handle_init(self): + if not self.sandbox_request and not self.sandbox_id: + raise ValueError("Must supply either an existing sandbox id or a SandboxStartRequest object") + if self.sandbox_id and self.sandbox_request: + raise ValueError("Pass either existing sandbox id, or SandboxStartRequest object to init, not both") + if self.sandbox_id: + self._info_log(f"Existing sandbox id '{self.sandbox_id}' passed. Refreshing components") + self.refresh_components() + + def _info_log(self, msg: str): + if self.logger: + self.logger.info(msg) + + def _debug_log(self, msg: str): + if self.logger: + self.logger.debug(msg) + + def _error_log(self, msg: str): + if self.logger: + self.logger.error(msg) + + def refresh_components(self): + self.components.refresh_components(self.api, self.sandbox_id) + + def launch_sandbox(self): + if self.sandbox_id: + self._debug_log(f"launch sandbox called for existing sandbox '{self.sandbox_id}'. Returning") + return + + if not self.sandbox_request: + raise ValueError("No StartSandboxRequest object passed to init") + + start_response = self.api.start_sandbox(blueprint_id=self.sandbox_request.blueprint_id, + sandbox_name=self.sandbox_request.sandbox_name, + duration=self.sandbox_request.duration, + bp_params=self.sandbox_request.bp_params, + permitted_users=self.sandbox_request.permitted_users) + self.sandbox_id = start_response.id + self._info_log(f"Sandbox '{self.sandbox_id}' LAUNCHED.") + self._info_log("Starting blocking sandbox and polling...") + start = default_timer() + try: + self.api.poll_sandbox_setup(reservation_id=self.sandbox_id, + max_polling_minutes=self.sandbox_request.max_polling_minutes, + polling_frequency_seconds=self.sandbox_request.polling_frequency_seconds, + log_level=self.sandbox_request.polling_log_level) + except exceptions.SetupFailedException as e: + self.orch_metadata.setup_errors = e.error_events + setup_duration_seconds = default_timer() - start + self._error_log(f"Sandbox '{self.sandbox_id}' setup FAILED after {setup_duration_seconds} seconds") + self._error_log(str(e)) + raise + + setup_duration_seconds = default_timer() - start + self._info_log(f"Sandbox '{self.sandbox_id}' setup completed after {setup_duration_seconds} seconds") + self.orch_metadata.setup_duration_seconds = setup_duration_seconds + self.refresh_components() + + def teardown_sandbox(self): + if not self.sandbox_id: + self._debug_log("Trying to start teardown, but no sandbox ID found. Returning") + return + + if self.orch_metadata.teardown_duration_seconds: + self._debug_log(f"Teardown has already ran for sandbox '{self.sandbox_id}'. Returning") + + self._info_log(f"starting blocking teardown of sandbox '{self.sandbox_id}'") + start = default_timer() + try: + self.api.stop_sandbox(sandbox_id=self.sandbox_id, + poll_teardown=True, + max_polling_minutes=self.teardown_settings.max_polling_minutes, + polling_frequency_seconds=self.teardown_settings.polling_frequency_seconds, + polling_log_level=self.teardown_settings.polling_log_level) + except exceptions.TeardownFailedException as e: + self.orch_metadata.teardown_errors = e.error_events + teardown_duration_seconds = default_timer() - start + self._error_log(f"Sandbox '{self.sandbox_id}' setup FAILED after {teardown_duration_seconds} seconds") + self._error_log(str(e)) + raise + + teardown_duration_seconds = default_timer() - start + self._info_log(f"Sandbox '{self.sandbox_id}' teardown completed after {teardown_duration_seconds} seconds") + self.orch_metadata.teardown_duration_seconds = teardown_duration_seconds + self.refresh_components() + + def __enter__(self): + """ start sandbox """ + if not self.sandbox_id: + self.launch_sandbox() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """ end sandbox """ + self.teardown_sandbox() + return self + + + + +if __name__ == "__main__": + controller = SandboxControllerContext() + controller.async_executor diff --git a/tests/assets/polling-orchestrations/setup_with_sleep/__main__.py b/tests/assets/polling-orchestrations/setup_with_sleep/__main__.py new file mode 100644 index 0000000..e4a4baf --- /dev/null +++ b/tests/assets/polling-orchestrations/setup_with_sleep/__main__.py @@ -0,0 +1,24 @@ +from cloudshell.workflow.orchestration.sandbox import Sandbox +from cloudshell.workflow.orchestration.setup.default_setup_orchestrator import DefaultSetupWorkflow +import time + + +SLEEP_SECONDS = 30 + + +def fake_config(sandbox, components=None): + """ + pretend to configure but then go to sleep + :param Sandbox sandbox: + :param components: + :return: + """ + sandbox.automation_api.WriteMessageToReservationOutput(sandbox.id, f"starting fake config for {SLEEP_SECONDS} seconds") + time.sleep(SLEEP_SECONDS) + + +sandbox = Sandbox() + +DefaultSetupWorkflow().register(sandbox) +sandbox.workflow.add_to_configuration(fake_config) +sandbox.execute_setup() diff --git a/tests/assets/polling-orchestrations/setup_with_sleep/requirements.txt b/tests/assets/polling-orchestrations/setup_with_sleep/requirements.txt new file mode 100644 index 0000000..9b4a25e --- /dev/null +++ b/tests/assets/polling-orchestrations/setup_with_sleep/requirements.txt @@ -0,0 +1 @@ +cloudshell-orch-core>=4.0.0.0,<4.1.0.0 \ No newline at end of file diff --git a/tests/assets/polling-orchestrations/setup_with_sleep/version.txt b/tests/assets/polling-orchestrations/setup_with_sleep/version.txt new file mode 100644 index 0000000..eb85010 --- /dev/null +++ b/tests/assets/polling-orchestrations/setup_with_sleep/version.txt @@ -0,0 +1 @@ +4.0.0.250 diff --git a/tests/assets/polling-orchestrations/teardown_with_sleep/__main__.py b/tests/assets/polling-orchestrations/teardown_with_sleep/__main__.py new file mode 100644 index 0000000..5d6a0c3 --- /dev/null +++ b/tests/assets/polling-orchestrations/teardown_with_sleep/__main__.py @@ -0,0 +1,24 @@ +from cloudshell.workflow.orchestration.sandbox import Sandbox +from cloudshell.workflow.orchestration.teardown.default_teardown_orchestrator import DefaultTeardownWorkflow +import time + +SLEEP_SECONDS = 30 + + +def fake_config(sandbox, components=None): + """ + pretend to configure but then go to sleep + :param Sandbox sandbox: + :param components: + :return: + """ + sandbox.automation_api.WriteMessageToReservationOutput(sandbox.id, f"starting fake config for {SLEEP_SECONDS} seconds") + time.sleep(SLEEP_SECONDS) + + +sandbox = Sandbox() + +DefaultTeardownWorkflow().register(sandbox) +sandbox.workflow.before_teardown_started(fake_config) + +sandbox.execute_teardown() diff --git a/tests/assets/polling-orchestrations/teardown_with_sleep/requirements.txt b/tests/assets/polling-orchestrations/teardown_with_sleep/requirements.txt new file mode 100644 index 0000000..9b4a25e --- /dev/null +++ b/tests/assets/polling-orchestrations/teardown_with_sleep/requirements.txt @@ -0,0 +1 @@ +cloudshell-orch-core>=4.0.0.0,<4.1.0.0 \ No newline at end of file diff --git a/tests/assets/polling-orchestrations/teardown_with_sleep/version.txt b/tests/assets/polling-orchestrations/teardown_with_sleep/version.txt new file mode 100644 index 0000000..eb85010 --- /dev/null +++ b/tests/assets/polling-orchestrations/teardown_with_sleep/version.txt @@ -0,0 +1 @@ +4.0.0.250 diff --git a/tests/common.py b/tests/common.py index e6d6678..4556e01 100644 --- a/tests/common.py +++ b/tests/common.py @@ -2,7 +2,7 @@ from random import randint from time import sleep -from src.cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from src.cloudshell.sandbox_rest.api import SandboxRestApiSession def pretty_print_response(dict_response): @@ -21,6 +21,7 @@ def fixed_sleep(): sleep(3) -def get_blueprint_id_from_name(api: SandboxRestApiSession, bp_name: str): +def get_blueprint_id_from_name(api: SandboxRestApiSession, bp_name: str) -> str: res = api.get_blueprint_details(bp_name) - return res["id"] + return res.id + diff --git a/tests/conftest.py b/tests/conftest.py index 2a1dbf4..6f0d282 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,17 +1,20 @@ import time import pytest -from constants import * -from env_settings import * +import constants +import env_settings -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest.api import SandboxRestApiSession +from cloudshell.sandbox_rest.default_logger import set_up_default_logger @pytest.fixture(scope="session") def admin_session() -> SandboxRestApiSession: - admin_api = SandboxRestApiSession( - host=CLOUDSHELL_SERVER, username=CLOUDSHELL_ADMIN_USER, password=CLOUDSHELL_ADMIN_PASSWORD, domain=CLOUDSHELL_DOMAIN - ) + admin_api = SandboxRestApiSession(host=env_settings.CLOUDSHELL_SERVER, + username=env_settings.CLOUDSHELL_ADMIN_USER, + password=env_settings.CLOUDSHELL_ADMIN_PASSWORD, + domain=env_settings.CLOUDSHELL_DOMAIN, + logger=set_up_default_logger()) print(f"Admin session started. Token: {admin_api.token}") with admin_api: yield admin_api @@ -23,9 +26,14 @@ def admin_session() -> SandboxRestApiSession: @pytest.fixture(scope="session") def empty_blueprint(): - return DEFAULT_EMPTY_BLUEPRINT + return constants.DEFAULT_EMPTY_BLUEPRINT @pytest.fixture(scope="session") def dut_blueprint(): - return DUT_BLUEPRINT + return constants.DUT_BLUEPRINT + + +@pytest.fixture(scope="session") +def sleep_orch_blueprint(): + return constants.SLEEP_ORCH_BLUEPRINT diff --git a/tests/constants.py b/tests/constants.py index 3d37424..2784ec9 100644 --- a/tests/constants.py +++ b/tests/constants.py @@ -4,6 +4,7 @@ # this blueprint has a Putshell Resource in it # https://community.quali.com/repos/3318/put-shell-mock DUT_BLUEPRINT = "DUT Blueprint Test" +SLEEP_ORCH_BLUEPRINT = "DUT Blueprint Sleep Orch" # DUT blueprint constants DUT_MODEL = "Putshell" diff --git a/tests/test_api_dut_sandbox.py b/tests/test_api_dut_sandbox.py index 9b6bc85..4d4d834 100644 --- a/tests/test_api_dut_sandbox.py +++ b/tests/test_api_dut_sandbox.py @@ -9,7 +9,8 @@ import constants import pytest -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest import model +from cloudshell.sandbox_rest.api import SandboxRestApiSession @pytest.fixture(scope="module") @@ -23,7 +24,7 @@ def blueprint_id(admin_session: SandboxRestApiSession, dut_blueprint): def sandbox_id(admin_session: SandboxRestApiSession, blueprint_id): # start sandbox start_res = admin_session.start_sandbox(blueprint_id=blueprint_id, sandbox_name="Pytest DUT blueprint test") - sandbox_id = start_res["id"] + sandbox_id = start_res.id print(f"Sandbox started: {sandbox_id}") common.fixed_sleep() yield sandbox_id @@ -35,9 +36,9 @@ def sandbox_id(admin_session: SandboxRestApiSession, blueprint_id): def component_id(admin_session: SandboxRestApiSession, sandbox_id: str): components = admin_session.get_sandbox_components(sandbox_id) common.fixed_sleep() - component_filter = [x for x in components if x["component_type"] == constants.DUT_MODEL] + component_filter = [x for x in components if x.component_type == constants.DUT_MODEL] assert component_filter - return component_filter[0]["id"] + return component_filter[0].id @pytest.fixture(scope="module") @@ -47,10 +48,9 @@ def execution_id(admin_session: SandboxRestApiSession, sandbox_id: str, componen sandbox_id=sandbox_id, component_id=component_id, command_name=constants.DUT_COMMAND ) common.fixed_sleep() - assert isinstance(res, dict) - print("Started execution response") - common.pretty_print_response(res) - execution_id = res["executionId"] + assert isinstance(res, model.CommandStartResponse) + print(f"Start execution response: {res.pretty_json()}") + execution_id = res.executionId return execution_id @@ -58,14 +58,13 @@ def execution_id(admin_session: SandboxRestApiSession, sandbox_id: str, componen def test_get_execution_details(admin_session, execution_id): res = admin_session.get_execution_details(execution_id) common.fixed_sleep() - assert isinstance(res, dict) + assert isinstance(res, model.CommandExecutionDetails) return res def test_delete_execution(admin_session, execution_id, test_get_execution_details): - print("Execution Details") - common.pretty_print_response(test_get_execution_details) - is_supports_cancellation = test_get_execution_details["supports_cancellation"] + print(f"Execution Details: {test_get_execution_details.pretty_json()}") + is_supports_cancellation = test_get_execution_details.supports_cancellation if not is_supports_cancellation: print("Can't cancel this command. Returning") return diff --git a/tests/test_api_empty_sandbox.py b/tests/test_api_empty_sandbox.py index e224e2b..e19cb44 100644 --- a/tests/test_api_empty_sandbox.py +++ b/tests/test_api_empty_sandbox.py @@ -5,7 +5,8 @@ import common import pytest -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest.api import SandboxRestApiSession +from cloudshell.sandbox_rest.api import model @pytest.fixture(scope="module") @@ -19,10 +20,11 @@ def blueprint_id(admin_session: SandboxRestApiSession, empty_blueprint): def sandbox_id(admin_session: SandboxRestApiSession, blueprint_id): # start sandbox start_res = admin_session.start_sandbox(blueprint_id=blueprint_id, sandbox_name="Pytest empty blueprint test") - sandbox_id = start_res["id"] + sandbox_id = start_res.id print(f"Sandbox started: {sandbox_id}") yield sandbox_id - admin_session.stop_sandbox(sandbox_id) + stop_response = admin_session.stop_sandbox(sandbox_id) + print(stop_response) print(f"\nSandbox ended: {sandbox_id}") @@ -34,8 +36,8 @@ def test_start_stop(sandbox_id): def test_get_sandbox_details(admin_session, sandbox_id): common.random_sleep() details_res = admin_session.get_sandbox_details(sandbox_id) - assert isinstance(details_res, dict) - sb_name = details_res["name"] + sb_name = details_res.name + assert isinstance(details_res, model.SandboxDetails) print(f"Pulled details for sandbox '{sb_name}'") @@ -51,24 +53,24 @@ def test_get_sandbox_commands(admin_session, sandbox_id): common.random_sleep() commands_res = admin_session.get_sandbox_commands(sandbox_id) assert isinstance(commands_res, list) - print(f"Sandbox commands: {[x['name'] for x in commands_res]}") - first_sb_command = admin_session.get_sandbox_command_details(sandbox_id, commands_res[0]["name"]) - print(f"SB command name: {first_sb_command['name']}\n" f"description: {first_sb_command['description']}") + print(f"Sandbox commands: {[x.name for x in commands_res]}") + first_sb_command_details = admin_session.get_sandbox_command_details(sandbox_id, commands_res[0].name) + print(f"SB command name: {first_sb_command_details.name}\n" f"description: {first_sb_command_details.description}") def test_get_sandbox_events(admin_session, sandbox_id): common.random_sleep() activity_res = admin_session.get_sandbox_activity(sandbox_id) - assert isinstance(activity_res, dict) and "events" in activity_res - events = activity_res["events"] + assert isinstance(activity_res, model.ActivityEventsResponse) + events = activity_res.events print(f"activity events count: {len(events)}") def test_get_console_output(admin_session, sandbox_id): common.random_sleep() output_res = admin_session.get_sandbox_output(sandbox_id) - assert isinstance(output_res, dict) and "entries" in output_res - entries = output_res["entries"] + assert isinstance(output_res, model.SandboxOutput) + entries = output_res.entries print(f"Sandbox output entries count: {len(entries)}") @@ -82,5 +84,5 @@ def test_get_instructions(admin_session, sandbox_id): def test_extend_sandbox(admin_session, sandbox_id): common.random_sleep() extend_response = admin_session.extend_sandbox(sandbox_id, "PT0H10M") - assert isinstance(extend_response, dict) and "remaining_time" in extend_response - print(f"extended sandbox. Remaining time: {extend_response['remaining_time']}") + assert isinstance(extend_response, model.ExtendResponse) + print(f"Extended sandbox - Remaining time: {extend_response.remaining_time}") diff --git a/tests/test_api_no_sandbox.py b/tests/test_api_no_sandbox.py index 521d024..c1e1ed5 100644 --- a/tests/test_api_no_sandbox.py +++ b/tests/test_api_no_sandbox.py @@ -7,7 +7,8 @@ import env_settings import pytest -from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession +from cloudshell.sandbox_rest import model +from cloudshell.sandbox_rest.api import SandboxRestApiSession @pytest.fixture(scope="module") @@ -37,12 +38,19 @@ def test_get_blueprints(admin_session: SandboxRestApiSession): bp_res = admin_session.get_blueprints() common.random_sleep() assert isinstance(bp_res, list) - print(f"Blueprint count found in system: '{len(bp_res)}'") + if bp_res: + print(f"Blueprint count found in system: '{len(bp_res)}'") + else: + print("no blueprints found in system") + return + first_bp = bp_res[0] + print(f"blueprint name: {first_bp.name}") + print(f"pretty printed:\n{first_bp.pretty_json()}") def test_get_default_blueprint(admin_session: SandboxRestApiSession): bp_res = admin_session.get_blueprint_details(constants.DEFAULT_EMPTY_BLUEPRINT) common.random_sleep() - assert isinstance(bp_res, dict) - bp_name = bp_res["name"] - print(f"Pulled details for '{bp_name}'") + bp_name = bp_res.name + print(f"Pulled details for '{bp_name}'\n{bp_res.pretty_json()}") + assert isinstance(bp_res, model.BlueprintDescription) diff --git a/tests/test_api_polling.py b/tests/test_api_polling.py new file mode 100644 index 0000000..2f803c5 --- /dev/null +++ b/tests/test_api_polling.py @@ -0,0 +1,70 @@ +""" +Test the api methods against blueprint with a resource containing a command. + +- Putshell mock can be used - https://community.quali.com/repos/3318/put-shell-mock +- DUT model / command can be referenced in constants.py (Putshell / health_check) +- Assumed that only one DUT is in blueprint +""" +import logging + +import common +import constants +import pytest + +from cloudshell.sandbox_rest import model +from cloudshell.sandbox_rest.api import SandboxRestApiSession +from timeit import default_timer + + +@pytest.fixture(scope="module") +def sandbox_id(admin_session: SandboxRestApiSession, sleep_orch_blueprint): + # start sandbox + start = default_timer() + start_res = admin_session.start_sandbox(blueprint_id=sleep_orch_blueprint, + sandbox_name="Pytest POLLING blueprint test", + polling_setup=True, + max_polling_minutes=5, + polling_frequency_seconds=10, + polling_log_level=logging.INFO) + sandbox_id = start_res.id + print(f"Sandbox started after {default_timer() - start:.2f} seconds. Sandbox id: {sandbox_id}. State: {start_res.state}") + yield sandbox_id + common.fixed_sleep() + + print("cleaning up sandbox") + # stop sandbox + stop_timer = default_timer() + stop_res = admin_session.stop_sandbox(sandbox_id, + poll_teardown=True, + max_polling_minutes=5, + polling_frequency_seconds=10, + polling_log_level=logging.INFO) + print(f"\nSandbox ended after {default_timer() - stop_timer:.2f} seconds. ID: {sandbox_id}. State: {stop_res.state}") + + +@pytest.fixture(scope="module") +def component_id(admin_session: SandboxRestApiSession, sandbox_id: str): + components = admin_session.get_sandbox_components(sandbox_id) + common.fixed_sleep() + component_filter = [x for x in components if x.component_type == constants.DUT_MODEL] + assert component_filter + return component_filter[0].id + + +def test_start_stop(sandbox_id): + print(f"\nstart stop test got sandbox id '{sandbox_id}'") + assert sandbox_id + + +def test_component_command_blocking(admin_session: SandboxRestApiSession, sandbox_id: str, component_id: str): + print("\nStarting blocking DUT Command...") + start = default_timer() + response = admin_session.run_component_command(sandbox_id=sandbox_id, + component_id=component_id, + command_name=constants.DUT_COMMAND, + polling_execution=True, + polling_log_level=logging.INFO) + common.fixed_sleep() + assert isinstance(response, model.ComponentCommandExecutionDetails) + print(f"Resource command finished after {default_timer() - start:.2f} seconds.\n" + f"Execution response: {response.pretty_json()}")