Add TenantBackupRestore (#40)

* first commit * changes * TenantBackupRestore * change * prepared return statement * add log exception * add operation status ids * get_file * get_file * HTTPResponse * add docstrings * pre-commit changes * pre-commit changes * fix return statement * fix return statemnt * delete prints * pre-commit fix * task_status.wait_for_completed * add usage example to docstring * Add operations status to deafult value * wait_for_completed return self * rm print * rm task endpoint as function argument * add default timeout in wait_for_completed * headers * debug * add done_scheduled operation status * add operation statuses * no-post_file * format * rm return None statement from log_excepction * fix DS in wait_for_completed * import * space * logging * checks * TaskStatus * Set sleep_seconds def value = 5 * TaskStatus-sleep * TaskStatus * checks * docstring * delete_all * delete_all * import_file * oops * no-return-class * no-return-class * typo * list * backup_files * dq * change doc string * use dataclass for wait_for_completed * final version * pre-commit changes * change to double quotes * fix log exception add unitest for task status * wait_for_completed * status * add test for task_status * Update vmngclient/api/task_status_api.py Co-authored-by: Tomasz Warda <[email protected]> * add example to docstring * unittest * imported-but-unused Co-authored-by: danielui <[email protected]> Co-authored-by: danielui <[email protected]> Co-authored-by: Tomasz Warda <[email protected]>
CiscoDevNet · Jan 9, 2023 · 6d20108 · 6d20108
1 parent 7cd1f7d
commit 6d20108
Show file tree

Hide file tree

Showing 3 changed files with 230 additions and 1 deletion.
diff --git a/vmngclient/api/tenant_backup_restore_api.py b/vmngclient/api/tenant_backup_restore_api.py
@@ -0,0 +1,144 @@
+"""
+Module for apidocs/#/Tenant Backup Restore
+"""
+import logging
+import re
+from pathlib import Path
+from typing import List, Optional
+
+from vmngclient.api.task_status_api import TaskStatus, wait_for_completed
+from vmngclient.session import vManageSession
+
+logger = logging.getLogger(__name__)
+
+
+class TenantBackupRestoreAPI:
+    """
+    Class for tenant backup and restore
+
+    Scope:
+        Provider-as-tenant or tenant
+
+    Attributes:
+        session (vManageSession): logged in API client session
+
+    Example usage:
+        from vmngclient.api.tenant_backup_restore_api import TenantBackupRestoreAPI
+        from vmngclient.session import create_vManageSession
+
+
+        tenant_session = create_vManageSession(
+             tenant_domain, tenantadmin, password, port)
+        provider_tenant_session = create_vManageSession(
+            domain, admin, password, port, subdomain=tenant_domain)
+
+        tenant_backup_restore = Tenant_backup_restoreAPI(tenant_session)
+        provider_backup_restore = Tenant_backup_restoreAPI(provider_tenant_session)
+
+        file_name = tenant_backup_restore.export()
+        file_list = provider_backup_restore.list()
+        file = tenant_backup_restore.download(file_name)
+        status = provider_backup_restore.import_file(file)
+        deleted_list = tenant_backup_restore.delete(file_name)
+        deleted_list = provider_backup_restore.delete_all()
+    """
+
+    def __init__(self, session: vManageSession) -> None:
+        self.session = session
+
+    def list(self) -> list:
+        """Return a list of backup files stored on vManage
+
+        Returns:
+            list of filenames
+
+        Example usage:
+            fileList = ProviderBackupRestore.list()
+        """
+        return self.session.get_json("/dataservice/tenantbackup/list")["backup_files"]
+
+    def export(self, timeout: int = 300) -> str:
+        """Export tenant backup file from DB to vManage storage
+
+        Args:
+            timeout: Max polling time for task (default: 300 seconds)
+
+        Returns:
+            str: filename of exported tenant backup file stored on vManage
+
+        Example usage:
+            fileName = ProviderBackupRestore.export()
+        """
+        response = self.session.get_json("/dataservice/tenantbackup/export")
+        status = wait_for_completed(self.session, response["processId"], timeout)
+        string = re.search("""file location: (.*)""", status.activity[-1])
+        assert string, "File location not found."
+        return string.group(1)
+
+    def delete(self, file: str) -> List[str]:
+        """Delete tenant backup file
+
+        Args:
+            file (str): device ID (usually system-ip)
+
+        Returns:
+            http response for delete operation
+
+        Example usage:
+            message = ProviderBackupRestore.delete(fileName)
+        """
+        url = f"/dataservice/tenantbackup/delete?fileName={file}"
+        return self.session.delete(url).json()["Deleted"]
+
+    def delete_all(self) -> List[str]:
+        """Delete all tenant backup files
+
+        Returns:
+            http response for delete operation
+
+        Example usage:
+            message = ProviderBackupRestore.delete_all()
+        """
+        return self.delete("all")
+
+    def download(self, file: str, download_dir: Optional[Path] = None) -> Path:
+        """Download tenant backup file
+
+        Args:
+            file: full path or base name of tenant backup file
+            download_dir: download directory (defaul: current working directory)
+
+        Returns:
+            Path to downloaded tenant backup file
+
+        Example usage:
+            file = ProviderBackupRestore.download(fileName)
+        """
+
+        if "/dataservice/tenantbackup/download/" not in file:
+            tenant_id = file.split("_")[1]
+            file = f"/dataservice/tenantbackup/download/{tenant_id}/{file}"
+
+        download_dir = download_dir if download_dir else Path.cwd()
+        download = download_dir / Path(file).name
+        self.session.get_file(file, download)
+        return download
+
+    def import_file(self, file: Path, timeout: int = 300) -> TaskStatus:
+        """Upload the file for tenant import to the DB and poll for Success
+
+        Args:
+            file: The path of the file to be upladed
+            timeout: Max polling time for task (default: 300 seconds)
+
+        Returns:
+            TaskStatus object with Success and Activity messages
+
+        Example usage:
+             status = ProviderBackupRestore.import_file(file)
+        """
+
+        url = "/dataservice/tenantbackup/import"
+        files = {"file": (file.name, open(str(file), "rb"))}
+        response = self.session.post(url, data={}, files=files)
+        return wait_for_completed(self.session, response.json()["processId"], timeout)
diff --git a/vmngclient/tests/test_tenant_backup_restore_api.py b/vmngclient/tests/test_tenant_backup_restore_api.py
@@ -0,0 +1,85 @@
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from vmngclient.api.tenant_backup_restore_api import TenantBackupRestoreAPI
+
+
+class TestTenantBackupRestoreAPI(unittest.TestCase):
+    def setUp(self):
+        self.processId = "5a6250c5ef1b1202"
+        self.download_file_content = "Downloaded file content"
+        self.full_name = Path("/dataservice")
+        self.full_name = (
+            self.full_name
+            / "tenantbackup"
+            / "download"
+            / "3cf632d2-5a62-45cb-af50-50c5ef1b1202"
+            / "bkup_3cf632d2-5a62-45cb-af50-50c5ef1b1202_010823-180713_88aba4772eef495ab6857a5076625ba9.tar.gz"
+        )
+        self.downloaded_file = Path("/tmp") / self.full_name.name
+
+    @patch("vmngclient.session.vManageSession")
+    @patch("requests.Response")
+    def test_delete_full_name(self, mock_session, mock_response):
+        # Arrange
+        mock_session.delete.return_value = mock_response
+        mock_response.json.return_value = mock_response
+        expected_api = f"/dataservice/tenantbackup/delete?fileName={self.full_name}"
+        # Act
+        TenantBackupRestoreAPI(mock_session).delete(str(self.full_name))
+        # Assert
+        mock_session.delete.assert_called_once_with(expected_api)
+
+    @patch("vmngclient.session.vManageSession")
+    @patch("requests.Response")
+    def test_delete_name(self, mock_session, mock_response):
+        # Arrange
+        mock_session.delete.return_value = mock_response
+        mock_response.json.return_value = mock_response
+        expected_api = f"/dataservice/tenantbackup/delete?fileName={self.full_name.name}"
+        # Act
+        TenantBackupRestoreAPI(mock_session).delete(str(self.full_name.name))
+        # Assert
+        mock_session.delete.assert_called_once_with(expected_api)
+
+    @patch("vmngclient.session.vManageSession")
+    @patch("requests.Response")
+    def test_delete_all(self, mock_session, mock_response):
+        # Arrange
+        mock_session.delete.return_value = mock_response
+        mock_response.json.return_value = mock_response
+        expected_api = "/dataservice/tenantbackup/delete?fileName=all"
+        # Act
+        TenantBackupRestoreAPI(mock_session).delete_all()
+        # Assert
+        mock_session.delete.assert_called_once_with(expected_api)
+
+    @patch("vmngclient.session.vManageSession")
+    @patch("requests.Response")
+    def test_download_full(self, mock_session, mock_response):
+        # Arrange
+        mock_session.get_file.return_value = mock_response
+        mock_session.get.return_value = mock_response
+        mock_response.status_code = 200
+        mock_response.content = self.download_file_content
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Act
+            download_path = TenantBackupRestoreAPI(mock_session).download(str(self.full_name), Path(tmpdir))
+            # Assert
+            self.assertEqual(download_path, Path(tmpdir) / self.full_name.name)
+
+    @patch("vmngclient.session.vManageSession")
+    @patch("requests.Response")
+    def test_download_name(self, mock_session, mock_response):
+        # Arrange
+        mock_session.get_file.return_value = mock_response
+        mock_session.get.return_value = mock_response
+        mock_response.status_code = 200
+        mock_response.content = self.download_file_content
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Act
+            download_path = TenantBackupRestoreAPI(mock_session).download(self.full_name.name, Path(tmpdir))
+            # Assert
+            self.assertEqual(download_path, Path(tmpdir) / self.full_name.name)
diff --git a/vmngclient/utils/response.py b/vmngclient/utils/response.py
@@ -27,7 +27,7 @@ class VManageResponseDebugInfo:
 
 def response_debug(response: Response, headers: bool = False) -> str:
     request_body = response.request.body
-    if isinstance(request_body, bytes):
+    if isinstance(request_body, bytes) and request_body.isascii():
         request_body = str(request_body, encoding="utf-8")
     info = VManageResponseDebugInfo(
         request={