all tests passing

Author: katzy687

cloudshell/sandbox_rest/sandbox_api.py

@@ -392,18 +392,21 @@ def get_execution_details(self, execution_id: str) -> dict:
             raise SandboxRestException(f"Failed to get execution details for '{execution_id}'", response)
         return response.json()
 
-    def delete_execution(self, execution_id: str) -> dict:
+    def delete_execution(self, execution_id: str) -> None:
+        """
+        API returns dict with single key on successful deletion of execution
+        {"result": "success"}
+        """
         self._validate_auth_headers()
         response = requests.delete(f"{self._versioned_url}/executions/{execution_id}", headers=self._auth_headers)
         if not response.ok:
             raise SandboxRestException(f"Failed to delete execution for '{execution_id}'", response)
-        return response.json()
+        response_dict = response.json()
+        if not response_dict["result"] == "success":
+            raise SandboxRestException(f"Failed execution deletion of id {execution_id}\n"
+                                       f"Response: {response_dict}")
 
 
 if __name__ == "__main__":
     admin_api = SandboxRestApiSession(host="localhost", username="admin", password="admin", domain="end_users")
     user_token = admin_api.get_token_for_target_user("end_user")
-    user_api = SandboxRestApiSession(host="localhost", token=user_token)
-    response = user_api.start_sandbox(blueprint_id="end user bp")
-    commands = user_api.get_sandbox_command_details()
-    pass

tests/common.py

@@ -1,6 +1,12 @@
 import json
+from 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 get_blueprint_id_from_name(api: SandboxRestApiSession, bp_name: str):
+    res = api.get_blueprint_details(bp_name)
+    return res["id"]

tests/conftest.py

@@ -2,13 +2,29 @@
 
 import pytest
 from env_settings import *
+from constants import *
 
 from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession
 
 
 @pytest.fixture(scope="session")
 def admin_session() -> SandboxRestApiSession:
-    with SandboxRestApiSession(CLOUDSHELL_SERVER, CLOUDSHELL_ADMIN_USER, CLOUDSHELL_ADMIN_PASSWORD) as api:
+    with SandboxRestApiSession(host=CLOUDSHELL_SERVER,
+                               username=CLOUDSHELL_ADMIN_USER,
+                               password=CLOUDSHELL_ADMIN_PASSWORD,
+                               domain=CLOUDSHELL_DOMAIN) as api:
         yield api
         time.sleep(2)
         print("admin session token revoked")
+
+
+@pytest.fixture(scope="session")
+def empty_blueprint():
+    return DEFAULT_EMPTY_BLUEPRINT
+
+
+@pytest.fixture(scope="session")
+def dut_blueprint():
+    return DUT_BLUEPRINT
+
+

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"

tests/env_settings.py

@@ -3,6 +3,7 @@
 CLOUDSHELL_ADMIN_USER=admin
 CLOUDSHELL_ADMIN_PASSWORD=admin
 CLOUDSHELL_SERVER=localhost
+CLOUDSHELL_DOMAIN=Global
 """
 
 import os
@@ -15,8 +16,4 @@
 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")
-
-
-DUT_RESOURCE = "DUT_1"
-DEFAULT_BLUEPRINT_TEMPLATE = "CloudShell Sandbox Template"
-HEALTH_CHECK_COMMAND = "health_check"
+CLOUDSHELL_DOMAIN = os.environ.get("CLOUDSHELL_DOMAIN")

tests/test_api _no_sandbox.py

@@ -1,34 +1,33 @@
 """
-Functionally test the api methods that do not require a live sandbox
-- get all blueprints
-- get all sandboxes
-- get blueprint by id
-- get token + delete token
+Test the api methods that do NOT require a blueprint
+Live Cloudshell server is still a dependency
 """
-from env_settings import DEFAULT_BLUEPRINT_TEMPLATE
+from constants import DEFAULT_EMPTY_BLUEPRINT
 
 from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession
 
 
 def test_get_sandboxes(admin_session: SandboxRestApiSession):
     res = admin_session.get_sandboxes()
+    assert(type(res) is list)
     print(f"Sandbox count found in system: {len(res)}")
 
 
 def test_get_blueprints(admin_session: SandboxRestApiSession):
     bp_res = admin_session.get_blueprints()
+    assert(type(bp_res) is 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_BLUEPRINT_TEMPLATE)
+    bp_res = admin_session.get_blueprint_details(DEFAULT_EMPTY_BLUEPRINT)
+    assert(type(bp_res) is dict)
     bp_name = bp_res["name"]
     print(f"Pulled details for '{bp_name}'")
 
 
 def test_get_and_delete_token(admin_session: SandboxRestApiSession):
     """ get token for admin user """
     token_res = admin_session.get_token_for_target_user("admin")
+    assert(type(token_res) is str)
     print(f"Token response: '{token_res}'")
-    admin_session.delete_token(token_res)
-    print("deleted token")

tests/test_api_empty_sandbox.py

@@ -1,18 +1,26 @@
 """
-Test the api methods that require a live sandbox against default
-- start sandbox
+Test the api methods that require an empty, PUBLIC blueprint
 """
-import pytest
-from env_settings import DEFAULT_BLUEPRINT_TEMPLATE
+import time
 
+import pytest
 from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession
+from constants import *
+from common import *
+
+
+@pytest.fixture(scope="module")
+def blueprint_id(admin_session: SandboxRestApiSession, empty_blueprint):
+    res_id = get_blueprint_id_from_name(admin_session, empty_blueprint)
+    assert(type(res_id) is str)
+    return res_id
 
 
 @pytest.fixture(scope="module")
-def sandbox_id(admin_session: SandboxRestApiSession):
+def sandbox_id(admin_session: SandboxRestApiSession, blueprint_id):
     # start sandbox
     start_res = admin_session.start_sandbox(
-        blueprint_id=DEFAULT_BLUEPRINT_TEMPLATE, sandbox_name="Pytest empty blueprint test"
+        blueprint_id=blueprint_id, sandbox_name="Pytest empty blueprint test"
     )
     sandbox_id = start_res["id"]
     print(f"Sandbox started: {sandbox_id}")
@@ -21,46 +29,77 @@ def sandbox_id(admin_session: SandboxRestApiSession):
     print(f"\nSandbox ended: {sandbox_id}")
 
 
+@pytest.fixture(scope="module")
+def setup_execution_id(admin_session: SandboxRestApiSession, sandbox_id: str):
+    polling_minutes = 2
+    counter = 0
+    while True:
+        if counter > polling_minutes:
+            raise Exception("Timeout waiting for setup to end")
+        state = admin_session.get_sandbox_details(sandbox_id)["state"]
+        if state == "Ready":
+            break
+        time.sleep(60)
+        counter += 1
+
+    print("Rerunning Setup...")
+    res = admin_session.run_sandbox_command(sandbox_id=sandbox_id,
+                                            command_name=BLUEPRINT_SETUP_COMMAND)
+    assert (type(res) is dict)
+    print("Setup re-run execution response")
+    pretty_print_response(res)
+    execution_id = res["executionId"]
+    return execution_id
+
+
 def test_start_stop(admin_session, sandbox_id):
-    pass
+    assert (type(sandbox_id) is str)
+    print(f"Sandbox ID: {sandbox_id}")
 
 
 def test_get_sandbox_details(admin_session, sandbox_id):
     details_res = admin_session.get_sandbox_details(sandbox_id)
+    assert (type(details_res) is dict)
     sb_name = details_res["name"]
     print(f"Pulled details for sandbox '{sb_name}'")
 
 
 def test_get_components(admin_session, sandbox_id):
-    sb_components = admin_session.get_sandbox_components(sandbox_id)
-    component_count = len(sb_components)
+    components_res = admin_session.get_sandbox_components(sandbox_id)
+    assert (type(components_res) is list)
+    component_count = len(components_res)
     print(f"component count found: {component_count}")
 
 
 def test_get_sandbox_commands(admin_session, sandbox_id):
-    sb_commands = admin_session.get_sandbox_commands(sandbox_id)
-    print(f"Sandbox commands: {[x['name'] for x in sb_commands]}")
-    first_sb_command = admin_session.get_sandbox_command_details(sandbox_id, sb_commands[0]["name"])
+    commands_res = admin_session.get_sandbox_commands(sandbox_id)
+    assert (type(commands_res) is 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):
-    activity = admin_session.get_sandbox_activity(sandbox_id)
-    events = activity["events"]
+    activity_res = admin_session.get_sandbox_activity(sandbox_id)
+    assert (type(activity_res) is 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):
-    sb_output = admin_session.get_sandbox_output(sandbox_id)
-    entries = sb_output["entries"]
+    output_res = admin_session.get_sandbox_output(sandbox_id)
+    assert (type(output_res) is 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):
-    instructions = admin_session.get_sandbox_instructions(sandbox_id)
-    print(f"Pulled sandbox instructions: '{instructions}'")
+    instructions_res = admin_session.get_sandbox_instructions(sandbox_id)
+    assert (type(instructions_res) is str)
+    print(f"Pulled sandbox instructions: '{instructions_res}'")
 
 
 def test_extend_sandbox(admin_session, sandbox_id):
     extend_response = admin_session.extend_sandbox(sandbox_id, "PT0H10M")
+    assert (type(extend_response) is dict and "remaining_time" in extend_response)
     print(f"extended sandbox. Remaining time: {extend_response['remaining_time']}")

tests/test_api_heath_check_sandbox.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 time
+
+import pytest
+from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession
+from constants import *
+from common import *
+
+
+@pytest.fixture(scope="module")
+def blueprint_id(admin_session: SandboxRestApiSession, dut_blueprint):
+    res_id = get_blueprint_id_from_name(admin_session, dut_blueprint)
+    assert(type(res_id) is 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}")
+    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, dut_blueprint: str):
+    components = admin_session.get_sandbox_components(sandbox_id)
+    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)
+    assert(type(res) is 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)
+    assert(type(res) is dict)
+    print("Execution Details")
+    pretty_print_response(res)
+
+
+def test_delete_execution(admin_session, execution_id, test_get_execution_details):
+    print("Stopping execution...")
+    time.sleep(1)
+    admin_session.delete_execution(execution_id)
+    print("Execution deleted")
+
+

tests/test_rerun_setup.py

@@ -0,0 +1,54 @@
+"""
+Test re-running setup to validate blueprint level commands, getting details, and then ending setup execution, finally end sandbox
+"""
+import time
+
+import pytest
+from cloudshell.sandbox_rest.sandbox_api import SandboxRestApiSession
+from constants import *
+from common import *
+
+
+@pytest.fixture(scope="module")
+def sandbox_id(admin_session: SandboxRestApiSession, empty_blueprint):
+    # start sandbox
+    start_res = admin_session.start_sandbox(
+        blueprint_id=empty_blueprint, 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}")
+
+
+@pytest.fixture(scope="module")
+def setup_execution_id(admin_session: SandboxRestApiSession, sandbox_id: str):
+    polling_minutes = 2
+    counter = 0
+    while True:
+        if counter > polling_minutes:
+            raise Exception("Timeout waiting for setup to end")
+        state = admin_session.get_sandbox_details(sandbox_id)["state"]
+        if state == "Ready":
+            break
+        time.sleep(60)
+        counter += 1
+
+    print("Rerunning Setup...")
+    res = admin_session.run_sandbox_command(sandbox_id=sandbox_id,
+                                            command_name=BLUEPRINT_SETUP_COMMAND)
+    assert (type(res) is dict)
+    print("Setup re-run execution response")
+    pretty_print_response(res)
+    execution_id = res["executionId"]
+    return execution_id
+
+
+def test_cancel_setup(admin_session, setup_execution_id):
+    print("Setup Execution Details")
+    res = admin_session.get_execution_details(setup_execution_id)
+    pretty_print_response(res)
+    print("Ending setup execution")
+    admin_session.delete_execution(setup_execution_id)
+    print("Setup execution cancelled")