test stable

Author: MaxNumerique

.gitignore

@@ -13,3 +13,4 @@ build
 .coverage*
 .mypy_cache
 *.db
+tests/data

src/opengeodeweb_viewer/config.py

@@ -49,8 +49,9 @@ def _copy_test_assets(
 
 def test_config(path):
     default_config()
-    tmp_data_root = tempfile.mkdtemp(prefix="ogw_test_data_")
-    os.environ["DATA_FOLDER_PATH"] = tmp_data_root
+    os.environ["DATA_FOLDER_PATH"] = os.path.join(
+        os.path.dirname(__file__), "..", "..", "tests", "data"
+    )
 
     os.environ["DATABASE_PATH"] = str(path)
     db_file = os.path.join(path, "project.db")

src/tests/conftest.py

@@ -70,61 +70,52 @@ def _reader_for_file(path: str):
             return vtk.vtkJPEGReader()
         return vtk.vtkJPEGReader()
 
-    def images_diff(self, first_image_path, second_image_path, tolerance: float = 0.0):
-        first_reader = self._reader_for_file(first_image_path)
+    def images_diff(self, first_image_path, second_image_path):
+        if ".png" in first_image_path:
+            first_reader = vtk.vtkPNGReader()
+        elif (".jpg" in first_image_path) or (".jpeg" in first_image_path):
+            first_reader = vtk.vtkJPEGReader()
         first_reader.SetFileName(first_image_path)
 
-        second_reader = self._reader_for_file(second_image_path)
+        if ".png" in second_image_path:
+            second_reader = vtk.vtkPNGReader()
+        elif (".jpg" in second_image_path) or (".jpeg" in second_image_path):
+            second_reader = vtk.vtkJPEGReader()
         second_reader.SetFileName(second_image_path)
 
         images_diff = vtk.vtkImageDifference()
         images_diff.SetInputConnection(first_reader.GetOutputPort())
         images_diff.SetImageConnection(second_reader.GetOutputPort())
         images_diff.Update()
 
-        error = images_diff.GetThresholdedError()
-        print(f"Image comparison error for {second_image_path}: {error} (tolerance: {tolerance})", flush=True)
-        return error <= tolerance
+        print(f"{images_diff.GetThresholdedError()=}")
+        return images_diff.GetThresholdedError()
 
-    def compare_image(self, nb_messages, filename, tolerance=0.0):
-
-        self.ws.settimeout(4.0)
-        image = None
-        deadline = time.time() + 12.0
-        while time.time() < deadline:
-            try:
-                msg = self.ws.recv()
-            except WebSocketTimeoutException:
-                continue
-            if isinstance(msg, bytes):
-                image = msg
-                break
-        if not isinstance(msg, bytes):
-            return False
-        test_file_path = os.path.abspath(
-            os.path.join(self.test_output_dir, "test.jpeg")
-        )
-        with open(test_file_path, "wb") as f:
-            f.write(image)
-        format = "jpeg"
-        try:
-            meta = self.ws.recv()
-            try:
-                format = json.loads(meta)["result"]["format"]
-            except Exception:
-                pass
-        except WebSocketTimeoutException:
-            pass
-
-        if format != "jpeg":
-            new_path = os.path.abspath(
+    def compare_image(self, nb_messages, filename):
+        for message in range(nb_messages):
+            print(f"{message=}", flush=True)
+            image = self.ws.recv()
+            if isinstance(image, bytes):
+                test_file_path = os.path.abspath(
+                    os.path.join(self.test_output_dir, "test.jpeg")
+                )
+                with open(test_file_path, "wb") as f:
+                    f.write(image)
+                    f.close()
+        if isinstance(image, bytes):
+            response = self.ws.recv()
+            print(f"{response=}", flush=True)
+            format = json.loads(response)["result"]["format"]
+            test_file_path = os.path.abspath(
                 os.path.join(self.test_output_dir, f"test.{format}")
             )
-            os.replace(test_file_path, new_path)
-            test_file_path = new_path
+            with open(test_file_path, "wb") as f:
+                f.write(image)
+                f.close()
+
+            path_image = os.path.join(self.images_dir_path, filename)
 
-        path_image = os.path.join(self.images_dir_path, filename)
-        return self.images_diff(test_file_path, path_image, tolerance)
+            return self.images_diff(test_file_path, path_image) == 0.0
 
     def _init_ws(self):
         self.ws.send(
@@ -204,10 +195,10 @@ def configure_test_environment():
     os.environ["TEST_DB_PATH"] = str(db_path)
 
     yield
-    tmp_data_path = os.environ.get("DATA_FOLDER_PATH")
-    if tmp_data_path and "ogw_test_data_" in tmp_data_path:
-        shutil.rmtree(tmp_data_path, ignore_errors=True)
-        print(f"Cleaned up test data folder: {tmp_data_path}", flush=True)
+    # tmp_data_path = os.environ.get("DATA_FOLDER_PATH")
+    # if tmp_data_path and "ogw_test_data_" in tmp_data_path:
+    #     shutil.rmtree(tmp_data_path, ignore_errors=True)
+    #     print(f"Cleaned up test data folder: {tmp_data_path}", flush=True)
 
 
 @pytest.fixture

src/tests/data/123456789/CrossSection.vtm

@@ -32,6 +32,7 @@ def test_points_size(server, dataset_factory):
         [{"id": mesh_id, "size": 15}],
     )
     assert server.compare_image(3, "mesh/points/size.jpeg") == True
+    print("####")
 
 
 def test_points_color(server, dataset_factory):

src/tests/data/123456789/hat.vtp

@@ -5,7 +5,7 @@
 
 def test_register_mesh(server, dataset_factory):
     dataset_factory(id="123456789", viewable_file_name="hat.vtp")
-    
+
     server.call(
         VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["register"]["rpc"],
         [{"id": "123456789"}],
@@ -15,12 +15,12 @@ def test_register_mesh(server, dataset_factory):
 
 def test_deregister_mesh(server, dataset_factory):
     dataset_factory(id="123456789", viewable_file_name="hat.vtp")
-    
+
     server.call(
         VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["register"]["rpc"],
         [{"id": "123456789"}],
     )
-    
+
     server.call(
         VtkMeshView.mesh_prefix + VtkMeshView.mesh_schemas_dict["deregister"]["rpc"],
         [{"id": "123456789"}],

src/tests/data/edged_curve.vtp

@@ -93,7 +93,7 @@ def test_take_screenshot(server, dataset_factory):
         server.images_dir_path, "viewer/take_screenshot_with_background.jpg"
     )
 
-    assert server.images_diff(first_image_path, second_image_path) == 0.0
+    assert server.images_diff(first_image_path, second_image_path) == True
 
     # Take a screenshot without background png
     server.call(
@@ -122,7 +122,7 @@ def test_take_screenshot(server, dataset_factory):
         server.images_dir_path, "viewer/take_screenshot_without_background.png"
     )
 
-    assert server.images_diff(first_image_path, second_image_path) == 0.0
+    assert server.images_diff(first_image_path, second_image_path) == True
 
     # Take a screenshot with background png
     server.call(
@@ -151,7 +151,7 @@ def test_take_screenshot(server, dataset_factory):
         server.images_dir_path, "viewer/take_screenshot_with_background.png"
     )
 
-    assert server.images_diff(first_image_path, second_image_path) == 0.0
+    assert server.images_diff(first_image_path, second_image_path) == True
 
 
 def test_picked_ids(server, dataset_factory):
@@ -169,7 +169,9 @@ def test_picked_ids(server, dataset_factory):
         print("Warning: picked_ids returned None response", flush=True)
         return
     if "result" not in response:
-        print(f"Warning: No 'result' key in picked_ids response: {response}", flush=True)
+        print(
+            f"Warning: No 'result' key in picked_ids response: {response}", flush=True
+        )
         return
     result = response["result"]
     if result is None:
@@ -339,4 +341,4 @@ def test_combined_scaling_and_grid(server, dataset_factory):
         + VtkViewerView.viewer_schemas_dict["set_z_scaling"]["rpc"],
         [{"z_scale": 2.5}],
     )
-    assert server.compare_image(3, "viewer/combined_scaling_and_grid.jpeg") == True
+    assert server.compare_image(3, "viewer/combined_scaling_and_grid.jpeg") == True