Add unit tests for cryptupload module

Author: csc-jm

tests/common/mockups.py

@@ -208,10 +208,11 @@ async def __aexit__(self, *_):
                     "project": "test-id-0",
                     "container": "test-container",
                     "object": "test-object",
+                    "object_name": "test-object1",
                     "receiver": "test-project-1",
                 },
                 "cookies": {},
-                "query": {},
+                "query": {"total": 1},
                 "headers": {},
                 "query_string": "",
                 "remote": "test-remote",
@@ -223,13 +224,25 @@ async def __aexit__(self, *_):
                     "Log": unittest.mock.MagicMock(logging.Logger),
                     "test-id": "placeholder",
                     "oidc_client": self.mock_oidc_client,
+                    "vault_client": types.SimpleNamespace(
+                        **{
+                            "put_header": unittest.mock.AsyncMock(),
+                        }
+                    ),
                 },
                 "url": types.SimpleNamespace(
                     **{
                         "host": "https://localhost",
                     }
                 ),
                 "path": "/",
+                "content": types.SimpleNamespace(
+                    **{
+                        "read": unittest.mock.AsyncMock(
+                            return_value=b"{'header': 'header'}"
+                        )
+                    }
+                ),
             }
         )
         super().setUp()

tests/upload/test_cryptupload.py

@@ -0,0 +1,39 @@
+"""Unit tests for swift_browser_ui.upload.cryptupload module."""
+
+import asyncio
+import aiohttp.web
+import unittest.mock
+
+from swift_browser_ui.upload.cryptupload import EncryptedUploadProxy
+import tests.common.mockups
+
+
+class CryptTestClass(tests.common.mockups.APITestBase):
+    """Test class for swift_browser_ui.upload.download functions."""
+
+    def setUp(self):
+        """Set up mocks."""
+        super().setUp()
+        self.mock_session = {
+            "endpoint": "https://test-endpoint-0/v1/AUTH_test-id",
+            "token": "test-token1",
+        }
+
+    async def test_add_headers(self):
+        """Test adding header for the upload file."""
+        proxy = EncryptedUploadProxy(self.mock_session, self.mock_client)
+        # When container creation fails
+        with self.assertRaises(aiohttp.web.HTTPForbidden):
+            await proxy.add_header(self.mock_request)
+
+        alt_client_resp = self.mock_client_response
+        alt_client_resp.status = 204
+        alt_client = self.mock_client
+        alt_client.head = unittest.mock.Mock(
+            return_value=self.MockHandler(
+                alt_client_resp,
+            )
+        )
+        resp = await proxy.add_header(self.mock_request)
+        self.assertTrue(proxy.header_uploaded)
+        self.assertEqual(resp.status, 201)

tests/upload/test_download.py

@@ -23,22 +23,25 @@ def setUp(self):
             "endpoint": "https://test-endpoint-0/v1/AUTH_test-id",
             "token": "test-token1",
         }
-    
+
     def mocked_requests_get(self, *args, **kwargs):
         class MockResponse:
             def __init__(self, headers, status_code):
                 self.status_code = status_code
                 self.headers = headers
+
             def __enter__(self):
                 return self
+
             def __exit__(self, exc_type, exc_value, traceback):
                 pass
+
             def iter_content(self, chunk_size):
                 cont = []
                 for i in range(chunk_size):
                     cont.append(1)
                 return cont
-        
+
         if kwargs["headers"]["X-Auth-Token"] == "test-token1":
             return MockResponse({}, 200)
         elif kwargs["headers"]["X-Auth-Token"] == "test-token2":
@@ -75,7 +78,9 @@ def test_file_wrapper(self):
 
     def test_parse_archive(self):
         """Test parsing a list of paths."""
-        cont_proxy = ContainerArchiveDownloadProxy(self.mock_session, "project", "container", "object")
+        cont_proxy = ContainerArchiveDownloadProxy(
+            self.mock_session, "project", "container", "object"
+        )
         with self.assertRaises(ValueError):
             cont_proxy._parse_archive_fs([""])
 
@@ -92,7 +97,9 @@ def test_parse_archive(self):
 
     def test_download_loop(self):
         """Test loop to initialize downloads."""
-        cont_proxy = ContainerArchiveDownloadProxy(self.mock_session, "project", "container", "object")
+        cont_proxy = ContainerArchiveDownloadProxy(
+            self.mock_session, "project", "container", "object"
+        )
         cont_proxy.archive = tarfile.open(
             name=cont_proxy.container + ".tar",
             mode="w|",
@@ -107,9 +114,9 @@ def test_download_loop(self):
                     "file": {
                         "name": "/file",
                         "type": "file",
-                        "tar_info": tarfile.TarInfo(name="/file")
+                        "tar_info": tarfile.TarInfo(name="/file"),
                     }
-                }
+                },
             }
         }
         cont_proxy.download_init()
@@ -123,13 +130,15 @@ def mocked_queue_get(*args, **kwargs):
     @mock.patch("queue.Queue.get", side_effect=mocked_queue_get)
     def test_tar_archiving_loop(self, mocked_queue):
         """Test end of loop to initialize tarballing."""
-        cont_proxy = ContainerArchiveDownloadProxy(self.mock_session, "project", "container", "object")
+        cont_proxy = ContainerArchiveDownloadProxy(
+            self.mock_session, "project", "container", "object"
+        )
         cont_proxy.archive = tarfile.open(
             name=cont_proxy.container + ".tar",
             mode="w|",
             fileobj=cont_proxy.output_queue,
         )
         cont_proxy.tar_archiving_loop()
         with self.assertRaises(OSError):
-            # Archive was closed 
-            cont_proxy.archive.getnames()
+            # Archive was closed
+            cont_proxy.archive.getnames()