SIMPLE-7783 Use test_data_dir fixture, allow Path for image upload (#148)

Author: virlos

conftest.py

@@ -44,8 +44,3 @@ def pytest_addoption(
     # pytest-asycnio
     if pluginmanager.get_plugin("asyncio") is None:
         parser.addini("asyncio_mode", "suppress the warning")
-
-
-@pytest.fixture
-def test_dir(request: pytest.FixtureRequest) -> Path:
-    return request.path.parent

tests/conftest.py

@@ -18,6 +18,7 @@
 # limitations under the License.
 #
 
+from functools import partial
 from pathlib import Path
 from unittest.mock import patch
 
@@ -60,7 +61,12 @@ def mocked_session():
         yield session
 
 
-def resp_body_from_file(request: httpx.Request) -> httpx.Response:
+@pytest.fixture(scope="session")
+def test_data_dir() -> Path:
+    return Path(__file__).parent / "test_data"
+
+
+def resp_body_from_file(test_data_dir: Path, request: httpx.Request) -> httpx.Response:
     """
     A callback that returns the contents of a file based on the request.
 
@@ -75,13 +81,12 @@ def resp_body_from_file(request: httpx.Request) -> httpx.Response:
     elif endpoint_parts[0] == "labs":
         lab_id = endpoint_parts[1]
         filename = "_".join(endpoint_parts[2:]) + "-" + lab_id + ".json"
-    test_dir = Path(__file__).parent.resolve()
-    file_path = test_dir / "test_data" / filename
+    file_path = test_data_dir / filename
     return httpx.Response(200, text=file_path.read_text())
 
 
 @pytest.fixture
-def respx_mock_with_labs(respx_mock):
+def respx_mock_with_labs(respx_mock, test_data_dir: Path):
     """
     A test fixture that provides basic lab data with respx_mock so that unit tests can
     call ``client.all_labs`` or ``client.join_existing_lab``.  The sample data includes
@@ -143,8 +148,9 @@ def respx_mock_with_labs(respx_mock):
         "labs/863799a0-3d09-4af4-be26-cad997b6ab27/simulation_stats",
         "labs/863799a0-3d09-4af4-be26-cad997b6ab27/layer3_addresses",
     )
+    side_effect = partial(resp_body_from_file, test_data_dir)
     for api in resp_from_files:
-        respx_mock.get(FAKE_HOST_API + api).mock(side_effect=resp_body_from_file)
+        respx_mock.get(FAKE_HOST_API + api).mock(side_effect=side_effect)
 
 
 @pytest.fixture

tests/test_client_library.py

@@ -690,12 +690,12 @@ def test_different_version_strings():
 
 
 def test_import_lab_offline_deprecated(
-    client_library_server_current, mocked_session, tmp_path: Path, test_dir
+    client_library_server_current, mocked_session, tmp_path: Path, test_data_dir: Path
 ):
     client_library = ClientLibrary(
         url="https://0.0.0.0/fake_url/", username="test", password="pa$$"
     )
-    topology_file_path = test_dir / "test_data/sample_topology.json"
+    topology_file_path = test_data_dir / "sample_topology.json"
     with open(topology_file_path) as fh:
         topology_file = fh.read()
         with pytest.deprecated_call():

tests/test_image_upload.py

@@ -73,9 +73,9 @@
 # then locally run test_image_upload_file, and this will generate all the files
 # in the expected_pass_list into test_data.
 @pytest.fixture
-def create_test_files(change_test_dir):
+def create_test_files(test_data_dir):
     for file_path in EXPECTED_PASS_LIST:
-        path = pathlib.Path("test_data") / file_path
+        path = test_data_dir / file_path
         path.write_text("test")
 
 

virl2_client/models/node_image_definition.py

@@ -210,9 +210,7 @@ def download_image_definition(self, definition_id: str) -> str:
         return self._session.get(url).text
 
     def upload_image_file(
-        self,
-        filename: str,
-        rename: str | None = None,
+        self, filename: Path | str, rename: str | None = None,
     ) -> None:
         """
         Upload an image file.
@@ -224,8 +222,8 @@ def upload_image_file(
 
         path = pathlib.Path(filename)
 
-        extension = TARGZ if filename.endswith(TARGZ) else path.suffix
         name = rename or path.name
+        extension = TARGZ if path.name.endswith(TARGZ) else path.suffix
 
         if not name.endswith(extension):
             message = (
@@ -248,6 +246,7 @@ def upload_image_file(
             )
             raise InvalidImageFile(message)
 
+        # path may be a PureWindowsPath, cannot use path.is_file
         if not os.path.isfile(filename):
             raise FileNotFoundError(filename)
         # TODO: a library should not be printing to stdout unless interactive
@@ -270,10 +269,13 @@ def callback_read(__n):
             return callback_read
 
         _file = open(filename, "rb")
-        _file.read = callback_read_factory(_file, print_progress_bar)
-        files = {"field0": (name, _file)}
+        try:
+            _file.read = callback_read_factory(_file, print_progress_bar)
+            files = {"field0": (name, _file)}
 
-        self._session.post(url, files=files, headers=headers)
+            self._session.post(url, files=files, headers=headers)
+        finally:
+            _file.close()
         print("Upload completed")
 
     def download_image_file_list(self) -> list[str]: