fix(export_file): fix some tests and use native_file if there is not input_file

Author: SpliiT

src/opengeodeweb_back/geode_functions.py

@@ -55,7 +55,8 @@ def geode_object_output_extensions(
     geode_object: GeodeObject,
 ) -> dict[GeodeObjectType, dict[str, bool]]:
     results: dict[GeodeObjectType, dict[str, bool]] = {}
-    for mixin_geode_object in geode_objects[geode_object.geode_object_type()].__mro__:
+    current_type = geode_object.geode_object_type()
+    for mixin_geode_object in geode_objects[current_type].__mro__:
         output_extensions_method = getattr(
             mixin_geode_object, "output_extensions", None
         )
@@ -66,11 +67,16 @@ def geode_object_output_extensions(
             continue
         object_output_extensions: dict[str, bool] = {}
         is_saveable_method = getattr(mixin_geode_object, "is_saveable")
+        target_type = None
+        if hasattr(mixin_geode_object, "geode_object_type"):
+            target_type = mixin_geode_object.geode_object_type()
+        if target_type and target_type != current_type:
+            continue
         for output_extension in output_extensions:
             bool_is_saveable = is_saveable_method(
                 geode_object, f"test.{output_extension}"
             )
             object_output_extensions[output_extension] = bool_is_saveable
-        if hasattr(mixin_geode_object, "geode_object_type"):
-            results[mixin_geode_object.geode_object_type()] = object_output_extensions
+        if target_type:
+            results[target_type] = object_output_extensions
     return results

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -29,7 +29,6 @@
 
 routes = flask.Blueprint("routes", __name__, url_prefix="/opengeodeweb_back")
 
-
 routes.register_blueprint(
     blueprint_models.routes,
     url_prefix=blueprint_models.routes.url_prefix,
@@ -38,7 +37,6 @@
 
 schemas_dict = get_schemas_dict(os.path.join(os.path.dirname(__file__), "schemas"))
 
-
 @routes.route(
     schemas_dict["allowed_files"]["route"],
     methods=schemas_dict["allowed_files"]["methods"],
@@ -51,7 +49,6 @@ def allowed_files() -> flask.Response:
             extensions.add(extension)
     return flask.make_response({"extensions": list(extensions)}, 200)
 
-
 @routes.route(
     schemas_dict["upload_file"]["route"],
     methods=schemas_dict["upload_file"]["methods"],
@@ -66,7 +63,6 @@ def upload_file() -> flask.Response:
     file.save(os.path.join(UPLOAD_FOLDER, filename))
     return flask.make_response({"message": "File uploaded"}, 201)
 
-
 @routes.route(
     schemas_dict["allowed_objects"]["route"],
     methods=schemas_dict["allowed_objects"]["methods"],
@@ -92,7 +88,6 @@ def allowed_objects() -> flask.Response:
         }
     return flask.make_response({"allowed_objects": allowed_objects}, 200)
 
-
 @routes.route(
     schemas_dict["missing_files"]["route"],
     methods=schemas_dict["missing_files"]["methods"],
@@ -131,7 +126,6 @@ def missing_files() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["geographic_coordinate_systems"]["route"],
     methods=schemas_dict["geographic_coordinate_systems"]["methods"],
@@ -156,7 +150,6 @@ def crs_converter_geographic_coordinate_systems() -> flask.Response:
         crs_list.append(crs)
     return flask.make_response({"crs_list": crs_list}, 200)
 
-
 @routes.route(
     schemas_dict["inspect_file"]["route"],
     methods=schemas_dict["inspect_file"]["methods"],
@@ -174,7 +167,6 @@ def inspect_file() -> flask.Response:
     inspection_result = extract_inspector_result(inspection_data)
     return flask.make_response({"inspection_result": inspection_result}, 200)
 
-
 def extract_inspector_result(inspection_data: Any) -> object:
     new_object = {}
 
@@ -202,7 +194,6 @@ def extract_inspector_result(inspection_data: Any) -> object:
             new_object["issues"] = issues
     return new_object
 
-
 @routes.route(
     schemas_dict["geode_objects_and_output_extensions"]["route"],
     methods=schemas_dict["geode_objects_and_output_extensions"]["methods"],
@@ -224,7 +215,6 @@ def geode_objects_and_output_extensions() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["save_viewable_file"]["route"],
     methods=schemas_dict["save_viewable_file"]["methods"],
@@ -242,7 +232,6 @@ def save_viewable_file() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["texture_coordinates"]["route"],
     methods=schemas_dict["texture_coordinates"]["methods"],
@@ -258,7 +247,6 @@ def texture_coordinates() -> flask.Response:
     texture_coordinates = geode_object.texture_manager().texture_names()
     return flask.make_response({"texture_coordinates": texture_coordinates}, 200)
 
-
 @routes.route(
     schemas_dict["vertex_attribute_names"]["route"],
     methods=schemas_dict["vertex_attribute_names"]["methods"],
@@ -279,7 +267,6 @@ def vertex_attribute_names() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["cell_attribute_names"]["route"],
     methods=schemas_dict["cell_attribute_names"]["methods"],
@@ -300,7 +287,6 @@ def cell_attribute_names() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["polygon_attribute_names"]["route"],
     methods=schemas_dict["polygon_attribute_names"]["methods"],
@@ -321,7 +307,6 @@ def polygon_attribute_names() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["polyhedron_attribute_names"]["route"],
     methods=schemas_dict["polyhedron_attribute_names"]["methods"],
@@ -344,7 +329,6 @@ def polyhedron_attribute_names() -> flask.Response:
         200,
     )
 
-
 @routes.route(
     schemas_dict["ping"]["route"],
     methods=schemas_dict["ping"]["methods"],
@@ -354,14 +338,12 @@ def ping() -> flask.Response:
     flask.current_app.config.update(LAST_PING_TIME=time.time())
     return flask.make_response({"message": "Flask server is running"}, 200)
 
-
 @routes.route(schemas_dict["kill"]["route"], methods=schemas_dict["kill"]["methods"])
 def kill() -> flask.Response:
     print("Manual server kill, shutting down...", flush=True)
     os._exit(0)
     return flask.make_response({"message": "Flask server is dead"}, 200)
 
-
 @routes.route(
     schemas_dict["export_project"]["route"],
     methods=schemas_dict["export_project"]["methods"],
@@ -381,6 +363,14 @@ def export_project() -> flask.Response:
     export_vease_path = os.path.join(project_folder, filename)
 
     with get_session() as session:
+        session.query(Data).filter(
+            (Data.input_file == None) | (Data.input_file == "")
+        ).update(
+            {Data.input_file: Data.native_file},
+            synchronize_session=False
+        )
+        session.commit()
+        
         rows = session.query(Data.id, Data.input_file, Data.additional_files).all()
 
     with zipfile.ZipFile(
@@ -410,7 +400,6 @@ def export_project() -> flask.Response:
 
     return utils_functions.send_file(project_folder, [export_vease_path], filename)
 
-
 @routes.route(
     schemas_dict["import_project"]["route"],
     methods=schemas_dict["import_project"]["methods"],
@@ -427,7 +416,6 @@ def import_project() -> flask.Response:
 
     data_folder_path: str = flask.current_app.config["DATA_FOLDER_PATH"]
 
-    # 423 Locked bypass : remove stopped requests
     if connection.scoped_session_registry:
         connection.scoped_session_registry.remove()
     if connection.engine:

tests/test_geode_functions.py

@@ -59,6 +59,6 @@ def test_input_output() -> None:
                             geode_object,
                             os.path.join(os.path.abspath(f"./output"), filename),
                         )
-                        assert type(saved_files) is list
+                        assert type(saved_files) is list 
                         for saved_file in saved_files:
-                            os.remove(saved_file)
+                            os.remove(saved_file)

tests/test_models_routes.py

@@ -14,7 +14,6 @@
 base_dir = os.path.abspath(os.path.dirname(__file__))
 data_dir = os.path.join(base_dir, "data")
 
-
 def test_model_mesh_components(client: FlaskClient, test_id: str) -> None:
     route = "/opengeodeweb_back/models/vtm_component_indices"
 
@@ -35,7 +34,6 @@ def test_model_mesh_components(client: FlaskClient, test_id: str) -> None:
     for uuid in uuid_dict.keys():
         assert isinstance(uuid, str)
 
-
 def test_extract_brep_uuids(client: FlaskClient, test_id: str) -> None:
     route = "/opengeodeweb_back/models/mesh_components"
     brep_filename = os.path.join(data_dir, "cube.og_brep")
@@ -57,7 +55,6 @@ def test_extract_brep_uuids(client: FlaskClient, test_id: str) -> None:
         uuid_dict = response.get_json()["uuid_dict"]
         assert isinstance(uuid_dict, dict)
 
-
 def test_export_project_route(client: FlaskClient, tmp_path: Path) -> None:
     route = "/opengeodeweb_back/export_project"
     snapshot = {
@@ -69,6 +66,41 @@ def test_export_project_route(client: FlaskClient, tmp_path: Path) -> None:
     database_root_path = os.path.join(project_folder, "project.db")
     with open(database_root_path, "wb") as f:
         f.write(b"test_project_db")
+
+    with get_session() as session:
+        session.query(Data).delete()
+        session.commit()
+
+        data1 = Data(
+            id="test_data_1",
+            geode_object="BRep",
+            viewer_object="BRep",
+            input_file=None,
+            native_file="test_native.txt",
+            additional_files=[]
+        )
+        data2 = Data(
+            id="test_data_2",
+            geode_object="Section",
+            viewer_object="Section",
+            input_file="test_input.txt",
+            native_file="test_native2.txt",
+            additional_files=[]
+        )
+        session.add(data1)
+        session.add(data2)
+        session.commit()
+
+        data1_dir = os.path.join(project_folder, "test_data_1")
+        os.makedirs(data1_dir, exist_ok=True)
+        with open(os.path.join(data1_dir, "test_native.txt"), "w") as f:
+            f.write("native file content")
+
+        data2_dir = os.path.join(project_folder, "test_data_2")
+        os.makedirs(data2_dir, exist_ok=True)
+        with open(os.path.join(data2_dir, "test_input.txt"), "w") as f:
+            f.write("input file content")
+
     response = client.post(route, json={"snapshot": snapshot, "filename": filename})
     assert response.status_code == 200
     assert response.headers.get("new-file-name") == filename
@@ -77,18 +109,28 @@ def test_export_project_route(client: FlaskClient, tmp_path: Path) -> None:
     zip_bytes = response.get_data()
     tmp_zip_path = tmp_path / filename
     tmp_zip_path.write_bytes(zip_bytes)
+
     with zipfile.ZipFile(tmp_zip_path, "r") as zip_file:
         names = zip_file.namelist()
         assert "snapshot.json" in names
         parsed = json.loads(zip_file.read("snapshot.json").decode("utf-8"))
         assert parsed == snapshot
         assert "project.db" in names
+        assert "test_data_1/test_native.txt" in names
+        assert "test_data_2/test_input.txt" in names
+    with get_session() as session:
+        data1_updated = session.query(Data).filter_by(id="test_data_1").first()
+        assert data1_updated.input_file == "test_native.txt"
+
+        data2_updated = session.query(Data).filter_by(id="test_data_2").first()
+        assert data2_updated.input_file == "test_input.txt"
+
     response.close()
+
     export_path = os.path.join(project_folder, filename)
     if os.path.exists(export_path):
         os.remove(export_path)
 
-
 def test_import_project_route(client: FlaskClient, tmp_path: Path) -> None:
     route = "/opengeodeweb_back/import_project"
     snapshot = {
@@ -137,7 +179,6 @@ def test_import_project_route(client: FlaskClient, tmp_path: Path) -> None:
 
     client.application.config["DATA_FOLDER_PATH"] = original_data_folder
 
-
 def test_save_viewable_workflow_from_file(client: FlaskClient) -> None:
     file = os.path.join(data_dir, "cube.og_brep")
     upload_resp = client.put(
@@ -164,7 +205,6 @@ def test_save_viewable_workflow_from_file(client: FlaskClient) -> None:
     refreshed = Data.get(data_id)
     assert refreshed is not None
 
-
 def test_save_viewable_workflow_from_object(client: FlaskClient) -> None:
     route = "/opengeodeweb_back/create/create_aoi"
     aoi_data = {
@@ -184,4 +224,4 @@ def test_save_viewable_workflow_from_object(client: FlaskClient) -> None:
     data_id = response.get_json()["id"]
     assert isinstance(data_id, str) and len(data_id) > 0
     assert response.get_json()["geode_object_type"] == "EdgedCurve3D"
-    assert response.get_json()["viewable_file"].endswith(".vtp")
+    assert response.get_json()["viewable_file"].endswith(".vtp")