Merge pull request #128 from ttafsir/0.2.4-branch

0.2.4 branch

README.md

@@ -81,7 +81,8 @@ You can interact with the EVE-NG API through the `client.api` interface
 from evengsdk.client import EvengClient
 
 
-client = EvengClient("10.246.32.254", log_file="test.log")
+client = EvengClient("10.246.32.254", log_file="test.log", ssl_verify=False, protocol="https")
+client.disable_insecure_warnings()  # disable warnings for self-signed certificates
 client.login(username="admin", password="eve")
 client.set_log_level("DEBUG")
 

src/evengsdk/api.py

@@ -24,6 +24,13 @@ def __init__(self, client):
         self.log = client.log
         self.version = None
         self.supports_multi_tenants = False
+        self.is_community = True
+
+        status = self.get_server_status()
+        self.version = status["data"]["version"]
+
+        if self.version and "pro" in self.version.lower():
+            self.is_community = False
 
     def __repr__(self):
         return "{}({})".format(self.__class__.__name__, self.client.session)
@@ -145,10 +152,11 @@ def get_folder(self, folder: str) -> Dict:
         """
         return self.client.get(f"/folders/{folder}")
 
-    @staticmethod
-    def normalize_path(path: str) -> str:
+    def normalize_path(self, path: str) -> str:
         if not path.startswith("/"):
-            path = "/" + path
+            path = (
+                "/" + path if self.is_community else f"/{self.client.username}/{path}"
+            )
         path = Path(path).resolve()
 
         # Add extension if needed
@@ -315,28 +323,40 @@ def get_node_by_name(self, path: str, name: str) -> Dict:
             return next((v for _, v in node_data.items() if v["name"] == name), None)
         return
 
-    def get_node_configs(self, path: str) -> Dict:
+    def get_node_configs(self, path: str, configset: str = "default") -> Dict:
         """Return information about node configs
 
         :param path: path to lab file (include parent folder)
         :type path: str
+        :param configset: name of the configset to retrieve configs for (pro version)
+        :type configset: str, optional
         """
         url = "/labs" + f"{self.normalize_path(path)}/configs"
+        if not self.is_community:
+            return self.client.post(url, data=json.dumps({"cfsid": configset}))
         return self.client.get(url)
 
-    def get_node_config_by_id(self, path: str, node_id: int) -> Dict:
+    def get_node_config_by_id(
+        self, path: str, node_id: int, configset: str = "default"
+    ) -> Dict:
         """Return configuration information about a specific node given
         the configuration ID
 
         :param path: path to lab file (include parent folder)
         :type path: str
         :param node_id: ID for node to retrieve configuration for
         :type node_id: int
+        :param configset: name of the configset to retrieve configs for (pro version)
+        :type configset: str, optional
         """
         url = "/labs" + f"{self.normalize_path(path)}/configs/{node_id}"
+        if not self.is_community:
+            return self.client.post(url, data=json.dumps({"cfsid": configset}))
         return self.client.get(url)
 
-    def upload_node_config(self, path: str, node_id: str, config: str) -> Dict:
+    def upload_node_config(
+        self, path: str, node_id: str, config: str, configset: str = "default"
+    ) -> Dict:
         """Upload node's startup config.
 
         :param path: path to lab file (include parent folder)
@@ -350,6 +370,8 @@ def upload_node_config(self, path: str, node_id: str, config: str) -> Dict:
         """
         url = "/labs" + f"{self.normalize_path(path)}/configs/{node_id}"
         payload = {"id": node_id, "data": config}
+        if not self.is_community:
+            payload["cfsid"] = configset
         return self.client.put(url, data=json.dumps(payload))
 
     def enable_node_config(self, path: str, node_id: str) -> Dict:
@@ -542,14 +564,24 @@ def connect_node_to_node(
         r2 = self.connect_p2p_interface(path, d_node_id, dst_int, net_id)
         return r1["status"] == "success" and r2["status"] == "success"
 
+    def _recursively_start_nodes(self, path: str) -> Dict:
+        nodes = self.list_nodes(path)
+        results = []
+        for node_id, _ in nodes["data"].items():
+            r = self.start_node(path, node_id)
+            results.append(r)
+        return self._extract_recursive_statuses(results)
+
     def start_all_nodes(self, path: str) -> Dict:
         """Start one or all nodes configured in a lab
 
         :param path: path to lab file (including parent folder)
         :type path: str
         """
-        url = f"/labs{self.normalize_path(path)}/nodes/start"
-        return self.client.get(url)
+        if self.is_community:
+            url = f"/labs{self.normalize_path(path)}/nodes/start"
+            return self.client.get(url)
+        return self._recursively_start_nodes(path)
 
     def stop_all_nodes(self, path: str) -> Dict:
         """Stop one or all nodes configured in a lab
@@ -580,8 +612,27 @@ def stop_node(self, path: str, node_id: str) -> Dict:
         :type node_id: str
         """
         url = "/labs" + self.normalize_path(path) + f"/nodes/{node_id}/stop"
+        if not self.is_community:
+            url += "/stopmode=3"
         return self.client.get(url)
 
+    def _recursively_wipe_nodes(self, path: str) -> Dict:
+        nodes = self.list_nodes(path)
+        results = []
+        for node_id, _ in nodes["data"].items():
+            r = self.wipe_node(path, node_id)
+            results.append(r)
+        return self._extract_recursive_statuses(results)
+
+    def _extract_recursive_statuses(self, results):
+        success = all(r["status"] == "success" for r in results)
+        messages = [r["message"] for r in results]
+        return {
+            "status": "success" if success else "error",
+            "data": results,
+            "message": messages,
+        }
+
     def wipe_all_nodes(self, path: str) -> Dict:
         """Wipe one or all nodes configured in a lab. Wiping deletes
         all user config, included startup-config, VLANs, and so on. The
@@ -591,8 +642,10 @@ def wipe_all_nodes(self, path: str) -> Dict:
         :type path: [type]
         :return: str
         """
-        url = "/labs" + self.normalize_path(path) + "/nodes/wipe"
-        return self.client.get(url)
+        if self.is_community:
+            url = "/labs" + self.normalize_path(path) + "/nodes/wipe"
+            return self.client.get(url)
+        return self._recursively_wipe_nodes(path)
 
     def wipe_node(self, path: str, node_id: int) -> Dict:
         """Wipe single node configured in a lab. Wiping deletes
@@ -865,7 +918,7 @@ def add_node(
         template: str,
         delay: int = 0,
         name: str = "",
-        node_type: str = "",
+        node_type: str = "qemu",
         top: int = randint(30, 70),
         left: int = randint(30, 70),
         console: str = "telnet",

src/evengsdk/cli/lab/commands.py

@@ -51,6 +51,26 @@ def _get_client_session():
     return thread_local.client
 
 
+def _get_nested_labs(folders: List, session: EvengClient, labs: List = None) -> List:
+    """recursively get all nested labs from folders"""
+    if len(folders) == 1 and folders[0]["name"] in ("..", "/"):
+        return []
+
+    if labs is None:
+        labs = []
+
+    for folder in folders:
+        if folder["name"] == "..":
+            continue
+        resp = session.api.get_folder(f'{folder["path"]}')
+        nested_labs = resp.get("data", {}).get("labs")
+        folders = resp.get("data", {}).get("folders")
+        labs.append(nested_labs)
+        if len(folders) >= 1:
+            _get_nested_labs(folders, session, labs)
+    return labs
+
+
 def _get_lab_folder(name: str) -> List[Dict]:
     """Get labs from nested folder structure
 
@@ -59,17 +79,17 @@ def _get_lab_folder(name: str) -> List[Dict]:
     :return: list of labs
     """
     session = _get_client_session()
-    r = session.api.get_folder(name)
+    resp = session.api.get_folder(name)
+
     # get the labs from the folder
-    labs_from_folder = [r.get("data", {}).get("labs")]
+    labs_from_folder = [resp.get("data", {}).get("labs")]
+
     # let's get labs from nested folders too
-    nested_folders = r.get("data", {}).get("folders")
-    while len(nested_folders) > 1:
-        for folder in nested_folders:
-            if folder["name"] == "..":
-                continue
-            r = session.api.get_folder(f'{folder["path"]}')
-            labs_from_folder.append(r.get("data", {}).get("labs"))
+    sub_folders = resp.get("data", {}).get("folders")
+    labs_from_sub_folders = _get_nested_labs(sub_folders, session)
+
+    # flatten the results to single iterable
+    labs_from_folder.extend(labs_from_sub_folders)
     return labs_from_folder
 
 
@@ -89,18 +109,22 @@ def _get_all_labs(client: EvengClient) -> List:
     """
     Get all labs from EVE-NG host by parsing all nested folders
     """
-    resp = client.api.list_folders()
-    root_folders = resp.get("data", {}).get("folders")
-    labs_in_root_folder = resp.get("data", {}).get("labs")
+    with console.status("[bold green]Retrieving labs...") as status:
+        status.update("Retrieving folders...")
+        resp = client.api.list_folders()
+        root_folders = resp.get("data", {}).get("folders")
+        labs_in_root_folder = resp.get("data", {}).get("labs")
 
-    # Get the lab information from all other folders (non-root)
-    labs_in_nested_folders = chain(
-        *thread_executor(_get_lab_folder, (x["name"] for x in root_folders))
-    )
-    # flatten the results to single iterable
-    # (labs from root folder + labs from nested)
-    all_lab_info = chain(labs_in_root_folder, *labs_in_nested_folders)
-    return thread_executor(_get_lab_details, (x["path"] for x in all_lab_info))
+        # Get the lab information from all other folders (non-root)
+        status.update("Retrieving nested folders...")
+        labs_in_nested_folders = chain(
+            *thread_executor(_get_lab_folder, (x["name"] for x in root_folders))
+        )
+        # flatten the results to single iterable
+        # (labs from root folder + labs from nested)
+        all_lab_info = chain(labs_in_root_folder, *labs_in_nested_folders)
+        status.update("Retrieving lab details from all folders...")
+        return thread_executor(_get_lab_details, (x["path"] for x in all_lab_info))
 
 
 def create_network_links(

src/evengsdk/cli/version.py

@@ -1,2 +1,2 @@
 # -*- coding: utf-8 -*-
-__version__ = "0.2.3"
+__version__ = "0.2.4"

src/evengsdk/client.py

@@ -28,6 +28,7 @@ def __init__(
         self.api = None
         self.port = port
         self.ssl_verify = ssl_verify
+        self.user = None
 
         # Create Logger and set Set log level
         self.log = logging.getLogger("eveng-client")
@@ -37,9 +38,13 @@ def __init__(
         else:
             self.log.addHandler(logging.NullHandler())
 
-        # disable insecure warnings
+        # Disable insecure warnings
         if disable_insecure_warnings:
-            requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
+            self.disable_insecure_warnings()
+
+    def disable_insecure_warnings(self):
+        # disable insecure warnings
+        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
 
     @property
     def url_prefix(self):
@@ -91,6 +96,7 @@ def login(self, username: str, password: str, *args, **kwargs) -> None:
         if r.ok:
             try:
                 "logged in" in r.json()
+                self.username = username
                 self.api = EvengApi(self)  # create API wrapper object
             except json.decoder.JSONDecodeError:
                 self.log.error("Error logging in: {}".format(r.text))
@@ -138,8 +144,14 @@ def _make_request(
 
         # EVE-NG API returns HTTP error code and message in JSON response
         if hasattr(r, "json"):
-            err_code = r.json().get("code")
-            err_msg = r.json().get("message")
+            try:
+                err_code = r.json().get("code")
+            except json.JSONDecodeError:
+                err_code = r.text
+            try:
+                err_msg = r.json().get("message")
+            except json.JSONDecodeError:
+                err_msg = r.text
             raise EvengHTTPError("Error: {} {}".format(err_code, err_msg))
 
         # Other HTTP errors for which we don't have a JSON response

src/tests/api/conftest.py

@@ -0,0 +1,22 @@
+from datetime import datetime
+
+import pytest
+
+
+@pytest.fixture(scope="module")
+def setup_lab(lab, authenticated_client):
+    lab_obj = authenticated_client.api.create_lab(**lab)
+    yield lab_obj
+    authenticated_client.api.delete_lab(lab["path"] + lab["name"])
+
+
+@pytest.fixture(scope="module")
+def lab():
+    now = datetime.now()
+    ts = str(datetime.timestamp(now)).split(".")[0]
+    return {"name": f"test-lab-{ts}", "description": "Test Lab", "path": "/"}
+
+
+@pytest.fixture()
+def lab_path(lab):
+    return lab["path"] + lab["name"]

src/tests/api/test_api.py

@@ -0,0 +1,34 @@
+class TestEvengApi:
+    """Test cases"""
+
+    def test_api_get_server_status(self, authenticated_client):
+        """
+        Verify server status using the API
+        """
+        r = authenticated_client.api.get_server_status()
+        assert r["data"].get("cpu") is not None
+
+    def test_list_node_templates(self, authenticated_client):
+        """
+        Verify we can list node templates from API
+        """
+        r = authenticated_client.api.list_node_templates()
+        assert r["data"] is not None
+
+    def test_node_template_detail(self, authenticated_client):
+        """
+        Verify that we get retrieve the details of a node template
+        """
+        node_types = ["a10"]
+        for n_type in node_types:
+            detail = authenticated_client.api.node_template_detail(n_type)
+            assert isinstance(detail, dict)
+
+    def test_list_network_types(self, authenticated_client):
+        """
+        Verify that we can retrieve EVE-NG networks. The
+        data returned is a dictionary that includes
+        network types and instances.
+        """
+        r = authenticated_client.api.list_networks()
+        assert r["data"]["bridge"] is not None