diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..96702e8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.idea/ +.pytest_cache/ +*.egg +*.egg-info +tests/.env +*.zip +dist/ +build/ +*.pyc +venv \ No newline at end of file diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..f6badf9 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,44 @@ +repos: + - repo: https://github.com/timothycrosley/isort + rev: 5.4.2 + hooks: + - id: isort + language_version: python3.7 + args: [ --line-length=127 ] + - repo: https://github.com/python/black + rev: 20.8b1 + hooks: + - id: black + language_version: python3.7 + args: [ --line-length=127 ] + - repo: https://gitlab.com/pycqa/flake8 + rev: 3.8.3 + hooks: + - id: flake8 + additional_dependencies: [ + flake8-docstrings, + flake8-builtins, + flake8-comprehensions, + flake8-print, + flake8-eradicate, + ] + language_version: python3.7 + args: [ + --max-line-length=127, + '--ignore=D100,D101,D102,D103,D104,D105,D106,D107,D200,D205,D210,D400,D401,W503,E203,F403,F405,T001' + ] + # See https://stackoverflow.com/questions/61238318/pylint-and-pre-commit-hook-unable-to-import/61238571#61238571 + - repo: local + hooks: + - id: pylint + name: pylint + entry: pylint + language: system + types: [ python ] + args: [ + --max-line-length=127, + --max-public-methods=32, + --max-args=13, + '--disable=too-few-public-methods,logging-fstring-interpolation,too-many-instance-attributes,no-else-return,too-many-locals,no-self-use,duplicate-code,broad-except,logging-not-lazy,unspecified-encoding, unused-wildcard-import,missing-function-docstring,missing-module-docstring,import-error,wildcard-import,invalid-name,redefined-outer-name,no-name-in-module, arguments-differ', + '--good-names=ip,rc,eval' + ] \ No newline at end of file diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..93ee38a --- /dev/null +++ b/LICENSE @@ -0,0 +1,22 @@ + +MIT License + +Copyright (c) 2021 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. \ No newline at end of file diff --git a/README.MD b/README.MD new file mode 100644 index 0000000..e64d887 --- /dev/null +++ b/README.MD @@ -0,0 +1,20 @@ +[![Python 3.7](https://img.shields.io/badge/python-3.7-blue.svg)](https://www.python.org/downloads/release/python/) +[![PyPI version](https://badge.fury.io/py/cloudshell-sandboxapi-wrapper.svg)](https://badge.fury.io/py/cloudshell-sandboxapi-wrapper) + +# Cloudshell Sandbox API Wrapper +A python client implementation of the [cloudshell sandbox api](https://help.quali.com/Online%20Help/0.0/Portal/Content/API/CS-Snbx-API-Topic.htm?Highlight=sandbox%20api). + +This client provides an object-oriented interface to instantiating an api session and interacting with the endpoints. +All methods return a loaded python dictionary/list of the [documented](https://help.quali.com/Online%20Help/0.0/Portal/Content/API/RefGuides/Sndbx-REST-API/REST-API-V2-Ref-Guide.htm?tocpath=CloudShell%20API%20Guide%7CCloudShell%20Sandbox%20API%7C_____3) json responses. +No additional library object wrapping implemented. + + +### Installation + +``` +pip install cloudshell_sandboxapi_wrapper +``` +## Usage +```python +pass +``` diff --git a/README.rst b/README.rst deleted file mode 100644 index 306cc41..0000000 --- a/README.rst +++ /dev/null @@ -1,27 +0,0 @@ -CloudShell Sandbox API Wrapper -============================== - -Installation -************* -:: - - pip install cloudshell_sandboxapi_wrapper - -Example Usage -************** -:: - - from cloudshell_sandboxapi_wrapper.SandboxAPI import SandboxAPI - sandbox = SandboxAPI(host=SERVER_NAME, username=USERNAME, password=PASSWORD, domain=DOMAIN, port=SERVER_PORT) - blueprints = sandbox.get_blueprints() - blueprint_id = sandbox.get_blueprint_details(blueprint_id=BLUEPRINT_NAME)['id'] - sandbox_id = sandbox.start_sandbox(BLUEPRINT_NAME, PT23H, SANDBOX_NAME) - sandbox.stop_sandbox(sandbox_id) - -| - -:Note: - Tested on cloudshell 9.3 with Python 2.7/3.7/3.8. - For API details, please refer to CloudShell Sandbox API help: `CloudShell Sandbox API `_ - -| diff --git a/Sandbox-API/Sandbox.py b/Sandbox-API/Sandbox.py deleted file mode 100644 index ee72f02..0000000 --- a/Sandbox-API/Sandbox.py +++ /dev/null @@ -1,357 +0,0 @@ -import json -import requests - - -class SandboxAPI: - """ Python wrapper for CloudShell Sandbox API - """ - def __init__(self, host, username, password, domain='Global', port=82): - """Initializes and logs in Sandbox - :param str host: hostname of IP address of sandbox API server - :param str username: CloudShell username - :param str password: CloudShell password - :param str domain: CloudShell domain (default=Global) - :param int port: Sandbox API port number(default=82) - """ - self._server_url = 'http://{}:{}/api'.format(host, port) - response = requests.put('{}/login'.format(self._server_url), - json={'username': username, 'password': password, 'domain': domain}) - - self._headers = {"Authorization": "Basic " + response.content[1:-1].decode('utf-8'), - "Content-Type": "application/json"} - - def get_blueprints(self): - """Get list of blueprints - :return: - """ - response = requests.get('{}/v2/blueprints'.format(self._server_url), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_blueprint_details(self, blueprint_id): - """Get details of a specific blueprint - :param blueprint_id: Blueprint name or id - :return: - """ - response = requests.get('{}/v2/blueprints/{}'.format(self._server_url, blueprint_id), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def start_sandbox(self, blueprint_id, duration, sandbox_name=None, parameters=None, permitted_users=None): - """Create a sandbox from the provided blueprint id - :param list permitted_users: list of permitted users ex: ['user1', 'user2'] - :param list parameters: List of dicts, input parameters in the format ex: [{"name": "Version", - "value": "3.0"}, {"name": "Build Number", "value": "5"}] - :param str blueprint_id: blueprint_id or name - :param str duration: duration in ISO 8601 format (P1Y1M1DT1H1M1S = 1year, 1month, 1day, 1hour, 1min, 1sec) - :param str sandbox_name: name of the sandbox, same as blueprint if name='' - :return: - """ - if not sandbox_name: - sandbox_name = self.get_blueprint_details(blueprint_id)['name'] - data_dict = {"duration": duration, "name": sandbox_name} - if permitted_users: - data_dict['permitted_users'] = permitted_users - if parameters: - data_dict["params"] = parameters - - response = requests.post('{}/v2/blueprints/{}/start'.format(self._server_url, blueprint_id), - headers=self._headers, - data=json.dumps(data_dict)) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandboxes(self): - """Get list of sandboxes - :return: - """ - response = requests.get('{}/v2/sandboxes'.format(self._server_url), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_details(self, sandbox_id): - """Get details of the given sandbox id - :param sandbox_id: Sandbox id - :return: - """ - response = requests.get('{}/v2/sandboxes/{}'.format(self._server_url, sandbox_id), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_activity(self, sandbox_id): - """Get list of sandbox activity - :param str sandbox_id: Sandbox id - :return: - """ - response = requests.get('{}/v2/sandboxes/{}/{}'.format(self._server_url, sandbox_id, 'activity'), - headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_commands(self, sandbox_id): - """Get list of sandbox commands - :param str sandbox_id: Sandbox id - :return: - """ - response = requests.get('{}/v2/sandboxes/{}/commands'.format(self._server_url, sandbox_id), - headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_command_details(self, sandbox_id, command_name): - """Get details of specific sandbox command - :param str sandbox_id: Sandbox id - :param str command_name: Sandbox command to be executed - :return: - """ - response = requests.get('{}/v2/sandboxes/{}/commands/{}'.format( - self._server_url, sandbox_id, command_name), - headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def sandbox_command_start(self, sandbox_id, command_name, params=None): - """Start a sandbox command - :param str sandbox_id: Sandbox id - :param str command_name: Sandbox command to be executed - :param dict params: parameters to be passed to the command ex: {"params": [{"name": "WaitTime", "value": "1"}], - "printOutput": True} - :return: - """ - if params: - response = requests.post('{}/v2/sandboxes/{}/commands/{}/start'.format( - self._server_url, sandbox_id, command_name), - data=json.dumps(params), - headers=self._headers) - else: - response = requests.post('{}/v2/sandboxes/{}/commands/{}/start'.format( - self._server_url, sandbox_id, command_name), - headers=self._headers) - - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_components(self, sandbox_id): - """Get list of sandbox components - :param str sandbox_id: Sandbox id - :return: Returns a list of tuples of component type, name and id - """ - response = requests.get('{}/v2/sandboxes/{}/components'.format( - self._server_url, sandbox_id), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_component_details(self, sandbox_id, component_id): - """Get details of components in sandbox - :param str sandbox_id: Sandbox id - :param str component_id: Component id - :return: Returns a tuple of component type, name, id, address, description - """ - response = requests.get('{}/v2/sandboxes/{}/components/{}'.format(self._server_url, sandbox_id, component_id), - headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_component_commands(self, sandbox_id, component_id): - """Get list of commands for a particular component in sandbox - :param str sandbox_id: Sandbox id - :param str component_id: Component id - :return: - """ - response = requests.get('{}/v2/sandboxes/{}/components/{}/commands'.format( - self._server_url, sandbox_id, component_id), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_component_command_details(self, sandbox_id, component_id, command): - """Get details of a command of sandbox component - :param str sandbox_id: Sandbox id - :param str component_id: Component id - :param str command: Command name - :return: - """ - response = requests.get('{}/v2/sandboxes/{}/components/{}/commands/{}'.format( - self._server_url, sandbox_id, component_id, command), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def sandbox_component_command_start(self, sandbox_id, component_id, command_name, params=None): - """Start a command on sandbox component - :param str sandbox_id: Sandbox id - :param str component_id: Component id - :param str command_name: Command name - :param dict params: parameters to be passed to the command ex: - {"params": [{"name": "Duration", "value": "Sandbox tester"}], "printOutput": True} - :return: - """ - if params: - response = requests.post('{}/v2/sandboxes/{}/components/{}/commands/{}/start'.format( - self._server_url, sandbox_id, component_id, command_name), data=json.dumps(params), - headers=self._headers) - else: - response = requests.post('{}/v2/sandboxes/{}/components/{}/commands/{}/start'.format( - self._server_url, sandbox_id, component_id, command_name), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def extend_sandbox(self, sandbox_id, duration): - """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: - """ - data_dict = json.loads('{"extended_time": "' + duration + '"}') - response = requests.post('{}/v2/sandboxes/{}/extend'.format(self._server_url, sandbox_id), - data=json.dumps(data_dict), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def get_sandbox_output(self, sandbox_id): - """Get list of sandbox output - :param str sandbox_id: Sandbox id - :return: - """ - response = requests.get('{}/v2/sandboxes/{}/{}'.format(self._server_url, sandbox_id, 'output'), - headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - def stop_sandbox(self, sandbox_id): - """Stop the sandbox given sandbox id - :param sandbox_id: Sandbox id - :return: - """ - response = requests.post('{}/v2/sandboxes/{}/stop'.format(self._server_url, sandbox_id), headers=self._headers) - if response.ok: - return json.loads(response.content) - return response.reason - - -usage = """ - bp_name = 'Environment1' - bp_id = '9a3ad040-14ff-4078-9b0a-6ce7986daaaa' - sb_duration = 'PT10M' - - sb_cmd = 'blueprint_power_cycle' - sb_cmd_params = {"params": [{"name": "WaitTime", "value": "1"}], "printOutput": True} - sb_component = 'Demo Resource 1' - sb_component_cmd = 'Hello' - sb_comp_cmd_params = {"params": [{"name": "Duration", "value": "Sandbox tester"}], "printOutput": True} - - sandbox_api = SandboxAPI('localhost', 'admin', 'admin', 'Global') - r = sandbox_api.get_blueprints() - print(r) - r = sandbox_api.get_blueprint_details(bp_name) - print(r) - r = sandbox_api.get_blueprint_details(bp_id) - print(r) - r = sandbox_api.start_sandbox(bp_name, sb_duration) - print(r) - sb_id = r['id'] - r = sandbox_api.get_sandboxes() - print(r) - r = sandbox_api.get_sandbox_details(sb_id) - print(r) - r = sandbox_api.get_sandbox_commands(sb_id) - print(r) - r = sandbox_api.get_sandbox_command_details(sb_id, 'Setup') - print(r) - r = sandbox_api.sandbox_command_start(sb_id, sb_cmd, sb_cmd_params) - print(r) - r = sandbox_api.get_sandbox_components(sb_id) - print(r) - comp_id = [c.get('id') for c in r if c.get('name') == sb_component][0] - - r = sandbox_api.get_sandbox_component_details(sb_id, comp_id) - print(r) - r = sandbox_api.get_sandbox_component_commands(sb_id, comp_id) - print(r) - r = sandbox_api.get_sandbox_component_command_details(sb_id, comp_id, sb_component_cmd) - print(r) - r = sandbox_api.sandbox_component_command_start( - sb_id, comp_id, sb_component_cmd, - params=sb_comp_cmd_params) - print(r) - r = sandbox_api.extend_sandbox(sb_id, sb_duration) - print(r) - r = sandbox_api.get_sandbox_activity(sb_id) - print(r) - r = sandbox_api.get_sandbox_output(sb_id) - print(r) - print(sandbox_api.stop_sandbox(sb_id)) -""" - - -def main(): - bp_name = 'Environment1' - bp_id = '9a3ad040-14ff-4078-9b0a-6ce7986daaaa' - sb_duration = 'PT10M' - - sb_cmd = 'blueprint_power_cycle' - sb_cmd_params = {"params": [{"name": "WaitTime", "value": "1"}], "printOutput": True} - sb_component = 'Demo Resource 1' - sb_component_cmd = 'Hello' - sb_comp_cmd_params = {"params": [{"name": "Duration", "value": "Sandbox tester"}], "printOutput": True} - - sandbox_api = SandboxAPI('localhost', 'admin', 'admin', 'Global') - r = sandbox_api.get_blueprints() - print(r) - r = sandbox_api.get_blueprint_details(bp_name) - print(r) - r = sandbox_api.get_blueprint_details(bp_id) - print(r) - r = sandbox_api.start_sandbox(bp_name, sb_duration) - print(r) - sb_id = r['id'] - r = sandbox_api.get_sandboxes() - print(r) - r = sandbox_api.get_sandbox_details(sb_id) - print(r) - r = sandbox_api.get_sandbox_commands(sb_id) - print(r) - r = sandbox_api.get_sandbox_command_details(sb_id, 'Setup') - print(r) - r = sandbox_api.sandbox_command_start(sb_id, sb_cmd, sb_cmd_params) - print(r) - r = sandbox_api.get_sandbox_components(sb_id) - print(r) - comp_id = [c.get('id') for c in r if c.get('name') == sb_component][0] - - r = sandbox_api.get_sandbox_component_details(sb_id, comp_id) - print(r) - r = sandbox_api.get_sandbox_component_commands(sb_id, comp_id) - print(r) - r = sandbox_api.get_sandbox_component_command_details(sb_id, comp_id, sb_component_cmd) - print(r) - r = sandbox_api.sandbox_component_command_start( - sb_id, comp_id, sb_component_cmd, - params=sb_comp_cmd_params) - print(r) - r = sandbox_api.extend_sandbox(sb_id, sb_duration) - print(r) - r = sandbox_api.get_sandbox_activity(sb_id) - print(r) - r = sandbox_api.get_sandbox_output(sb_id) - print(r) - print(sandbox_api.stop_sandbox(sb_id)) - - -if __name__ == '__main__': - print(usage) - main() diff --git a/Sandbox-API/__init__.py b/Sandbox-API/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/Sandbox-API/quali_config.json b/Sandbox-API/quali_config.json deleted file mode 100644 index e1e47f4..0000000 --- a/Sandbox-API/quali_config.json +++ /dev/null @@ -1,7 +0,0 @@ -{ - "server_name": "SERVER_NAME_OR_IP", - "server_port": "82", - "username": "USERNAME", - "password": "PASSWORD", - "domain": "DOMAIN" -} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1b68d94 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["setuptools>=42", "wheel"] +build-backend = "setuptools.build_meta" \ No newline at end of file diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..ff6f3e3 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,5 @@ +pytest +python-dotenv +flake8 +pylint +pre-commit \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9bcc92b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +requests>=2,<3 +abstract-http-client \ No newline at end of file diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..06172dc --- /dev/null +++ b/setup.cfg @@ -0,0 +1,27 @@ +[metadata] +name = cloudshell_sandboxapi_wrapper +version = file: version.txt +author = QualiLab +author_email = support@qualisystems.com +description = Python client for CloudShell Sandbox REST api - consume sandboxes via CI +long_description = file: README.MD +long_description_content_type = text/markdown +url = https://github.com/QualiSystemsLab/Sandbox-API-Python +classifiers = + Programming Language :: Python :: 3.7 + License :: OSI Approved :: MIT License + Operating System :: OS Independent +license = MIT +license_file = LICENSE + + +[options] +package_dir = + = src +packages = find: +python_requires = >=3.7 +install_requires = + requests>=2 + +[options.packages.find] +where = src \ No newline at end of file diff --git a/setup.py b/setup.py index 3740fdc..da0fa5e 100644 --- a/setup.py +++ b/setup.py @@ -1,27 +1,4 @@ from setuptools import setup - -def readme(): - with open('README.rst') as f: - return f.read() - - -setup( - name='cloudshell_sandboxapi_wrapper', - version='1.0.0', - packages=['cloudshell_sandboxapi_wrapper'], - url='http://www.quali.com', - license='Apache 2.0', - author='sadanand.s', - author_email='sadanand.s@quali.com', - description='Python wrapper for CloudShell Sandbox API', - long_description=readme(), - classifiers=[ - 'License :: OSI Approved :: Apache Software License', - 'Natural Language :: English', - 'Programming Language :: Python :: 2.7', - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', - ], - install_requires=['requests'] -) +if __name__ == "__main__": + setup() diff --git a/src/cloudshell/__init__.py b/src/cloudshell/__init__.py new file mode 100644 index 0000000..b36383a --- /dev/null +++ b/src/cloudshell/__init__.py @@ -0,0 +1,3 @@ +from pkgutil import extend_path + +__path__ = extend_path(__path__, __name__) diff --git a/src/cloudshell/sandbox_rest/__init__.py b/src/cloudshell/sandbox_rest/__init__.py new file mode 100644 index 0000000..b36383a --- /dev/null +++ b/src/cloudshell/sandbox_rest/__init__.py @@ -0,0 +1,3 @@ +from pkgutil import extend_path + +__path__ = extend_path(__path__, __name__) diff --git a/src/cloudshell/sandbox_rest/exceptions.py b/src/cloudshell/sandbox_rest/exceptions.py new file mode 100644 index 0000000..befbf51 --- /dev/null +++ b/src/cloudshell/sandbox_rest/exceptions.py @@ -0,0 +1,6 @@ +class SandboxRestException(Exception): + """ Base Exception Class inside Rest client class """ + + +class SandboxRestAuthException(SandboxRestException): + """ Failed login action """ diff --git a/src/cloudshell/sandbox_rest/sandbox_api.py b/src/cloudshell/sandbox_rest/sandbox_api.py new file mode 100644 index 0000000..203b9fa --- /dev/null +++ b/src/cloudshell/sandbox_rest/sandbox_api.py @@ -0,0 +1,345 @@ +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: + 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 and self.user and self.password: + self.token = self._get_token_with_credentials(self.user, self.password, self.domain) + else: + raise ValueError("Login requires Token or Username / Password") + + 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/tests/common.py b/tests/common.py new file mode 100644 index 0000000..e6d6678 --- /dev/null +++ b/tests/common.py @@ -0,0 +1,26 @@ +import json +from random import randint +from time import sleep + +from src.cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession + + +def pretty_print_response(dict_response): + json_str = json.dumps(dict_response, indent=4) + print(f"\n{json_str}") + + +def random_sleep(): + """ To offset the api calls and avoid rate limit quota """ + random = randint(1, 3) + print(f"sleeping {random} seconds") + sleep(random) + + +def fixed_sleep(): + sleep(3) + + +def get_blueprint_id_from_name(api: SandboxRestApiSession, bp_name: str): + res = api.get_blueprint_details(bp_name) + return res["id"] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..0bbf68f --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,29 @@ +import time + +import pytest +from constants import * +from env_settings import * + +from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession + + +@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 + ) + with admin_api: + yield admin_api + time.sleep(3) + print("admin session token revoked") + print(f"total requests: {admin_api.rest_service.request_counter}") + + +@pytest.fixture(scope="session") +def empty_blueprint(): + return DEFAULT_EMPTY_BLUEPRINT + + +@pytest.fixture(scope="session") +def dut_blueprint(): + return DUT_BLUEPRINT diff --git a/tests/constants.py b/tests/constants.py new file mode 100644 index 0000000..77becad --- /dev/null +++ b/tests/constants.py @@ -0,0 +1,7 @@ +DEFAULT_EMPTY_BLUEPRINT = "CloudShell Sandbox Template" +BLUEPRINT_SETUP_COMMAND = "Setup" + +# DUT blueprint constants +DUT_MODEL = "Putshell" +DUT_COMMAND = "health_check" +DUT_BLUEPRINT = "DUT Blueprint Test" diff --git a/tests/env_settings.py b/tests/env_settings.py new file mode 100644 index 0000000..306daa0 --- /dev/null +++ b/tests/env_settings.py @@ -0,0 +1,19 @@ +""" +create .env file in tests directory with following keys: +CLOUDSHELL_ADMIN_USER=admin +CLOUDSHELL_ADMIN_PASSWORD=admin +CLOUDSHELL_SERVER=localhost +CLOUDSHELL_DOMAIN=Global +""" + +import os + +from dotenv import load_dotenv + +load_dotenv() + +# server credentials from .env +CLOUDSHELL_ADMIN_USER = os.environ.get("CLOUDSHELL_ADMIN_USER") +CLOUDSHELL_ADMIN_PASSWORD = os.environ.get("CLOUDSHELL_ADMIN_PASSWORD") +CLOUDSHELL_SERVER = os.environ.get("CLOUDSHELL_SERVER") +CLOUDSHELL_DOMAIN = os.environ.get("CLOUDSHELL_DOMAIN") diff --git a/tests/test_api_dut_sandbox.py b/tests/test_api_dut_sandbox.py new file mode 100644 index 0000000..35bf6a9 --- /dev/null +++ b/tests/test_api_dut_sandbox.py @@ -0,0 +1,73 @@ +""" +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 pytest +from common import * +from constants import * + +from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession + + +@pytest.fixture(scope="module") +def blueprint_id(admin_session: SandboxRestApiSession, dut_blueprint): + res_id = get_blueprint_id_from_name(admin_session, dut_blueprint) + assert isinstance(res_id, str) + return res_id + + +@pytest.fixture(scope="module") +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"] + print(f"Sandbox started: {sandbox_id}") + fixed_sleep() + yield sandbox_id + admin_session.stop_sandbox(sandbox_id) + print(f"\nSandbox ended: {sandbox_id}") + + +@pytest.fixture(scope="module") +def component_id(admin_session: SandboxRestApiSession, sandbox_id: str): + components = admin_session.get_sandbox_components(sandbox_id) + fixed_sleep() + component_filter = [x for x in components if x["component_type"] == DUT_MODEL] + assert component_filter + return component_filter[0]["id"] + + +@pytest.fixture(scope="module") +def execution_id(admin_session: SandboxRestApiSession, sandbox_id: str, component_id: str): + print("Starting DUT Command...") + res = admin_session.run_component_command(sandbox_id=sandbox_id, component_id=component_id, command_name=DUT_COMMAND) + fixed_sleep() + assert isinstance(res, dict) + print("Started execution response") + pretty_print_response(res) + execution_id = res["executionId"] + return execution_id + + +@pytest.fixture(scope="module") +def test_get_execution_details(admin_session, execution_id): + res = admin_session.get_execution_details(execution_id) + fixed_sleep() + assert isinstance(res, dict) + return res + + +def test_delete_execution(admin_session, execution_id, test_get_execution_details): + print("Execution Details") + pretty_print_response(test_get_execution_details) + is_supports_cancellation = test_get_execution_details["supports_cancellation"] + if not is_supports_cancellation: + print("Can't cancel this command. Returning") + return + print("Stopping execution...") + admin_session.delete_execution(execution_id) + fixed_sleep() + print("Execution deleted") diff --git a/tests/test_api_empty_sandbox.py b/tests/test_api_empty_sandbox.py new file mode 100644 index 0000000..2681ad5 --- /dev/null +++ b/tests/test_api_empty_sandbox.py @@ -0,0 +1,86 @@ +""" +Test the api methods that require an empty, PUBLIC blueprint +""" + +import pytest +from common import * + +from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession + + +@pytest.fixture(scope="module") +def blueprint_id(admin_session: SandboxRestApiSession, empty_blueprint): + res_id = get_blueprint_id_from_name(admin_session, empty_blueprint) + assert isinstance(res_id, str) + return res_id + + +@pytest.fixture(scope="module") +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"] + print(f"Sandbox started: {sandbox_id}") + yield sandbox_id + admin_session.stop_sandbox(sandbox_id) + print(f"\nSandbox ended: {sandbox_id}") + + +def test_start_stop(sandbox_id): + assert isinstance(sandbox_id, str) + print(f"Sandbox ID: {sandbox_id}") + + +def test_get_sandbox_details(admin_session, sandbox_id): + random_sleep() + details_res = admin_session.get_sandbox_details(sandbox_id) + assert isinstance(details_res, dict) + sb_name = details_res["name"] + print(f"Pulled details for sandbox '{sb_name}'") + + +def test_get_components(admin_session, sandbox_id): + random_sleep() + components_res = admin_session.get_sandbox_components(sandbox_id) + assert isinstance(components_res, list) + component_count = len(components_res) + print(f"component count found: {component_count}") + + +def test_get_sandbox_commands(admin_session, sandbox_id): + 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']}") + + +def test_get_sandbox_events(admin_session, sandbox_id): + 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"] + print(f"activity events count: {len(events)}") + + +def test_get_console_output(admin_session, sandbox_id): + 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"] + print(f"Sandbox output entries count: {len(entries)}") + + +def test_get_instructions(admin_session, sandbox_id): + random_sleep() + instructions_res = admin_session.get_sandbox_instructions(sandbox_id) + assert isinstance(instructions_res, str) + print(f"Pulled sandbox instructions: '{instructions_res}'") + + +def test_extend_sandbox(admin_session, sandbox_id): + 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']}") diff --git a/tests/test_api_no_sandbox.py b/tests/test_api_no_sandbox.py new file mode 100644 index 0000000..93f1f3e --- /dev/null +++ b/tests/test_api_no_sandbox.py @@ -0,0 +1,43 @@ +""" +Test the api methods that do NOT require a blueprint +Live Cloudshell server is still a dependency +""" +import pytest +from common import * +from constants import DEFAULT_EMPTY_BLUEPRINT + +from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession + + +@pytest.fixture(scope="module") +def api_token(admin_session: SandboxRestApiSession): + token = admin_session.get_token_for_target_user("admin") + return token + + +def test_delete_token(admin_session: SandboxRestApiSession, api_token: str): + assert isinstance(api_token, str) + print(f"Token response: '{api_token}'") + admin_session.delete_token(api_token) + + +def test_get_sandboxes(admin_session: SandboxRestApiSession): + res = admin_session.get_sandboxes() + random_sleep() + assert isinstance(res, list) + print(f"Sandbox count found in system: {len(res)}") + + +def test_get_blueprints(admin_session: SandboxRestApiSession): + bp_res = admin_session.get_blueprints() + random_sleep() + assert isinstance(bp_res, list) + print(f"Blueprint count found in system: '{len(bp_res)}'") + + +def test_get_default_blueprint(admin_session: SandboxRestApiSession): + bp_res = admin_session.get_blueprint_details(DEFAULT_EMPTY_BLUEPRINT) + random_sleep() + assert isinstance(bp_res, dict) + bp_name = bp_res["name"] + print(f"Pulled details for '{bp_name}'") diff --git a/version.txt b/version.txt new file mode 100644 index 0000000..359a5b9 --- /dev/null +++ b/version.txt @@ -0,0 +1 @@ +2.0.0 \ No newline at end of file