diff --git a/opengeodeweb_back_schemas.json b/opengeodeweb_back_schemas.json index a8a9bd9d..52a0151c 100644 --- a/opengeodeweb_back_schemas.json +++ b/opengeodeweb_back_schemas.json @@ -32,26 +32,6 @@ "additionalProperties": false } }, - "models": { - "model_components": { - "$id": "opengeodeweb_back/models/model_components", - "route": "/model_components", - "methods": [ - "POST" - ], - "type": "object", - "properties": { - "id": { - "type": "string", - "minLength": 1 - } - }, - "required": [ - "id" - ], - "additionalProperties": false - } - }, "vertex_attribute_names": { "$id": "opengeodeweb_back/vertex_attribute_names", "route": "/vertex_attribute_names", diff --git a/requirements.txt b/requirements.txt index 8fdabbf9..4504a827 100644 --- a/requirements.txt +++ b/requirements.txt @@ -60,4 +60,3 @@ werkzeug==3.1.2 # flask # flask-cors -opengeodeweb-microservice==1.*,>=1.0.16rc1 diff --git a/src/opengeodeweb_back/app.py b/src/opengeodeweb_back/app.py index 72eec40e..12b65f09 100644 --- a/src/opengeodeweb_back/app.py +++ b/src/opengeodeweb_back/app.py @@ -10,7 +10,6 @@ from werkzeug.exceptions import HTTPException from opengeodeweb_back import utils_functions, app_config from opengeodeweb_back.routes import blueprint_routes -from opengeodeweb_back.routes.models import blueprint_models from opengeodeweb_back.routes.create import blueprint_create from opengeodeweb_microservice.database import connection @@ -85,11 +84,6 @@ def register_ogw_back_blueprints(app: flask.Flask) -> None: url_prefix="/opengeodeweb_back", name="opengeodeweb_back", ) - app.register_blueprint( - blueprint_models.routes, - url_prefix="/opengeodeweb_back/models", - name="opengeodeweb_models", - ) app.register_blueprint( blueprint_create.routes, url_prefix="/opengeodeweb_back/create", diff --git a/src/opengeodeweb_back/routes/blueprint_routes.py b/src/opengeodeweb_back/routes/blueprint_routes.py index 2d4cc9d8..faf2f1e3 100644 --- a/src/opengeodeweb_back/routes/blueprint_routes.py +++ b/src/opengeodeweb_back/routes/blueprint_routes.py @@ -22,7 +22,6 @@ # Local application imports from opengeodeweb_back import geode_functions, utils_functions from opengeodeweb_back.routes import schemas -from opengeodeweb_back.routes.models import blueprint_models from opengeodeweb_back.geode_objects import geode_objects from opengeodeweb_back.geode_objects.geode_mesh import GeodeMesh from opengeodeweb_back.geode_objects.geode_graph import GeodeGraph @@ -34,13 +33,6 @@ routes = flask.Blueprint("routes", __name__, url_prefix="/opengeodeweb_back") - -routes.register_blueprint( - blueprint_models.routes, - url_prefix=blueprint_models.routes.url_prefix, - name=blueprint_models.routes.name, -) - schemas_dict = get_schemas_dict(os.path.join(os.path.dirname(__file__), "schemas")) diff --git a/src/opengeodeweb_back/routes/models/blueprint_models.py b/src/opengeodeweb_back/routes/models/blueprint_models.py deleted file mode 100644 index 01668f07..00000000 --- a/src/opengeodeweb_back/routes/models/blueprint_models.py +++ /dev/null @@ -1,83 +0,0 @@ -import os -import xml.etree.ElementTree as ET -import flask -from opengeodeweb_microservice.schemas import get_schemas_dict - -from opengeodeweb_back import geode_functions, utils_functions -from opengeodeweb_back.geode_objects.geode_model import GeodeModel -from . import schemas - -routes = flask.Blueprint("models", __name__, url_prefix="/models") -schemas_dict = get_schemas_dict(os.path.join(os.path.dirname(__file__), "schemas")) - - -@routes.route( - schemas_dict["model_components"]["route"], - methods=schemas_dict["model_components"]["methods"], -) -def model_components() -> flask.Response: - json_data = utils_functions.validate_request( - flask.request, schemas_dict["model_components"] - ) - params = schemas.ModelComponents.from_dict(json_data) - model = geode_functions.load_geode_object(params.id) - if not isinstance(model, GeodeModel): - flask.abort(400, f"{params.id} is not a GeodeModel") - - vtm_file_path = geode_functions.data_file_path(params.id, "viewable.vtm") - tree = ET.parse(vtm_file_path) - root = tree.find("vtkMultiBlockDataSet") - if root is None: - flask.abort(500, "Failed to read viewable file") - uuid_to_flat_index = {} - current_index = 0 - assert root is not None - for elem in root.iter(): - if "uuid" in elem.attrib and elem.tag == "DataSet": - uuid_to_flat_index[elem.attrib["uuid"]] = current_index - current_index += 1 - model_mesh_components = model.mesh_components() - mesh_components = [] - for mesh_component, ids in model_mesh_components.items(): - component_type = mesh_component.get() - for id in ids: - geode_id = id.string() - component_name = geode_id - viewer_id = uuid_to_flat_index[geode_id] - boundaries = model.boundaries(id) - boundaries_uuid = [boundary.id().string() for boundary in boundaries] - internals = model.internals(id) - internals_uuid = [internal.id().string() for internal in internals] - mesh_component_object = { - "viewer_id": viewer_id, - "geode_id": geode_id, - "name": component_name, - "type": component_type, - "boundaries": boundaries_uuid, - "internals": internals_uuid, - } - mesh_components.append(mesh_component_object) - - model_collection_components = model.collection_components() - collection_components = [] - for collection_component, ids in model_collection_components.items(): - component_type = collection_component.get() - for id in ids: - geode_id = id.string() - items = model.items(id) - items_uuid = [item.id().string() for item in items] - collection_component_object = { - "geode_id": geode_id, - "name": geode_id, - "type": component_type, - "items": items_uuid, - } - collection_components.append(collection_component_object) - - return flask.make_response( - { - "mesh_components": mesh_components, - "collection_components": collection_components, - }, - 200, - ) diff --git a/src/opengeodeweb_back/routes/models/schemas/__init__.py b/src/opengeodeweb_back/routes/models/schemas/__init__.py deleted file mode 100644 index 4ef41586..00000000 --- a/src/opengeodeweb_back/routes/models/schemas/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .model_components import * diff --git a/src/opengeodeweb_back/routes/models/schemas/model_components.json b/src/opengeodeweb_back/routes/models/schemas/model_components.json deleted file mode 100644 index b054564d..00000000 --- a/src/opengeodeweb_back/routes/models/schemas/model_components.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "route": "/model_components", - "methods": [ - "POST" - ], - "type": "object", - "properties": { - "id": { - "type": "string", - "minLength": 1 - } - }, - "required": [ - "id" - ], - "additionalProperties": false -} \ No newline at end of file diff --git a/src/opengeodeweb_back/routes/models/schemas/model_components.py b/src/opengeodeweb_back/routes/models/schemas/model_components.py deleted file mode 100644 index a471788a..00000000 --- a/src/opengeodeweb_back/routes/models/schemas/model_components.py +++ /dev/null @@ -1,10 +0,0 @@ -from dataclasses_json import DataClassJsonMixin -from dataclasses import dataclass - - -@dataclass -class ModelComponents(DataClassJsonMixin): - def __post_init__(self) -> None: - print(self, flush=True) - - id: str diff --git a/src/opengeodeweb_back/utils_functions.py b/src/opengeodeweb_back/utils_functions.py index cf71173a..c5da4ab6 100644 --- a/src/opengeodeweb_back/utils_functions.py +++ b/src/opengeodeweb_back/utils_functions.py @@ -2,6 +2,7 @@ import os import threading import time +import xml.etree.ElementTree as ET import zipfile from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor @@ -22,6 +23,7 @@ # Local application imports from . import geode_functions from .geode_objects import geode_objects +from .geode_objects.geode_model import GeodeModel from .geode_objects.geode_object import GeodeObject @@ -191,11 +193,69 @@ def create_data_folder_from_id(data_id: str) -> str: return data_path +def model_components( + data_id: str, model: GeodeModel, viewable_file: str +) -> dict[str, Any]: + vtm_file_path = geode_functions.data_file_path(data_id, viewable_file) + tree = ET.parse(vtm_file_path) + root = tree.find("vtkMultiBlockDataSet") + if root is None: + flask.abort(500, "Failed to read viewable file") + uuid_to_flat_index = {} + current_index = 0 + assert root is not None + for elem in root.iter(): + if "uuid" in elem.attrib and elem.tag == "DataSet": + uuid_to_flat_index[elem.attrib["uuid"]] = current_index + current_index += 1 + model_mesh_components = model.mesh_components() + mesh_components = [] + for mesh_component, ids in model_mesh_components.items(): + component_type = mesh_component.get() + for id in ids: + geode_id = id.string() + component_name = geode_id + viewer_id = uuid_to_flat_index[geode_id] + boundaries = model.boundaries(id) + boundaries_uuid = [boundary.id().string() for boundary in boundaries] + internals = model.internals(id) + internals_uuid = [internal.id().string() for internal in internals] + mesh_component_object = { + "viewer_id": viewer_id, + "geode_id": geode_id, + "name": component_name, + "type": component_type, + "boundaries": boundaries_uuid, + "internals": internals_uuid, + } + mesh_components.append(mesh_component_object) + + model_collection_components = model.collection_components() + collection_components = [] + for collection_component, ids in model_collection_components.items(): + component_type = collection_component.get() + for id in ids: + geode_id = id.string() + items = model.items(id) + items_uuid = [item.id().string() for item in items] + collection_component_object = { + "geode_id": geode_id, + "name": geode_id, + "type": component_type, + "items": items_uuid, + } + collection_components.append(collection_component_object) + return { + "mesh_components": mesh_components, + "collection_components": collection_components, + } + + def save_all_viewables_and_return_info( geode_object: GeodeObject, data: Data, data_path: str, -) -> dict[str, str | list[str]]: +) -> dict[str, Any]: with ThreadPoolExecutor() as executor: native_files, viewable_path, light_path = executor.map( lambda args: args[0](args[1]), @@ -225,7 +285,8 @@ def save_all_viewables_and_return_info( name = geode_object.identifier.name() if not name: flask.abort(400, "Geode object has no name defined.") - return { + + response: dict[str, Any] = { "native_file": data.native_file, "viewable_file": data.viewable_file, "id": data.id, @@ -234,11 +295,14 @@ def save_all_viewables_and_return_info( "binary_light_viewable": binary_light_viewable.decode("utf-8"), "geode_object_type": data.geode_object, } + if isinstance(geode_object, GeodeModel): + response |= model_components(data.id, geode_object, data.viewable_file) + return response def generate_native_viewable_and_light_viewable_from_object( geode_object: GeodeObject, -) -> dict[str, str | list[str]]: +) -> dict[str, Any]: data = Data.create( geode_object=geode_object.geode_object_type(), viewer_object=geode_object.viewer_type(), @@ -250,7 +314,7 @@ def generate_native_viewable_and_light_viewable_from_object( def generate_native_viewable_and_light_viewable_from_file( geode_object_type: GeodeObjectType, input_file: str -) -> dict[str, str | list[str]]: +) -> dict[str, Any]: generic_geode_object = geode_objects[geode_object_type] data = Data.create( geode_object=geode_object_type, diff --git a/tests/test_models_routes.py b/tests/test_models_routes.py deleted file mode 100644 index bcd05b63..00000000 --- a/tests/test_models_routes.py +++ /dev/null @@ -1,276 +0,0 @@ -import os -import shutil -import zipfile -import json -from flask.testing import FlaskClient -from werkzeug.datastructures import FileStorage -from pathlib import Path - -from opengeodeweb_microservice.database.data import Data -from opengeodeweb_microservice.database.connection import get_session -from opengeodeweb_back import geode_functions -from opengeodeweb_back.geode_objects.geode_brep import GeodeBRep - - -from .test_routes import test_save_viewable_file - -base_dir = os.path.abspath(os.path.dirname(__file__)) -data_dir = os.path.join(base_dir, "data") - - -def test_model_components(client: FlaskClient) -> None: - geode_object_type = "BRep" - filename = "cube.og_brep" - response = test_save_viewable_file(client, geode_object_type, filename) - - route = "/opengeodeweb_back/models/model_components" - brep_filename = os.path.join(data_dir, "cube.og_brep") - - response = client.post(route, json={"id": response.get_json()["id"]}) - assert response.status_code == 200 - assert "mesh_components" in response.get_json() - mesh_components = response.get_json()["mesh_components"] - assert isinstance(mesh_components, list) - assert len(mesh_components) > 0 - for mesh_component in mesh_components: - assert isinstance(mesh_component, object) - assert isinstance(mesh_component["geode_id"], str) - assert isinstance(mesh_component["viewer_id"], int) - assert isinstance(mesh_component["name"], str) - assert isinstance(mesh_component["type"], str) - assert isinstance(mesh_component["boundaries"], list) - for boundary_uuid in mesh_component["boundaries"]: - assert isinstance(boundary_uuid, str) - assert isinstance(mesh_component["internals"], list) - for internal_uuid in mesh_component["internals"]: - assert isinstance(internal_uuid, str) - assert "collection_components" in response.get_json() - collection_components = response.get_json()["collection_components"] - assert isinstance(collection_components, list) - for collection_component in collection_components: - assert isinstance(collection_component, object) - assert isinstance(collection_component["geode_id"], str) - assert isinstance(collection_component["name"], str) - assert isinstance(collection_component["items"], list) - for item_uuid in collection_component["items"]: - assert isinstance(item_uuid, str) - - -def test_export_project_route(client: FlaskClient, tmp_path: Path) -> None: - route = "/opengeodeweb_back/export_project" - snapshot = { - "styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}} - } - filename = "export_project_test.vease" - project_folder = client.application.config["DATA_FOLDER_PATH"] - os.makedirs(project_folder, exist_ok=True) - 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", - viewer_elements_type="default", - native_file="native.txt", - ) - data2 = Data( - id="test_data_2", - geode_object="Section", - viewer_object="Section", - viewer_elements_type="default", - native_file="native.txt", - ) - 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, "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, "native.txt"), "w") as f: - f.write("native file content") - - response = client.post(route, json={"snapshot": snapshot, "filename": filename}) - assert response.status_code == 200 - assert response.headers.get("new-file-name") == filename - assert response.mimetype == "application/octet-binary" - response.direct_passthrough = False - 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/native.txt" in names - assert "test_data_2/native.txt" in names - - 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 = { - "styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}} - } - - original_data_folder = client.application.config["DATA_FOLDER_PATH"] - client.application.config["DATA_FOLDER_PATH"] = os.path.join( - str(tmp_path), "project_data" - ) - db_path = os.path.join(client.application.config["DATA_FOLDER_PATH"], "project.db") - - import sqlite3, zipfile, json - - temp_db = tmp_path / "temp_project.db" - conn = sqlite3.connect(str(temp_db)) - conn.execute( - "CREATE TABLE datas (id TEXT PRIMARY KEY, geode_object TEXT, viewer_object TEXT, viewer_elements_type TEXT, native_file TEXT, " - "viewable_file TEXT, light_viewable_file TEXT)" - ) - conn.commit() - conn.close() - - z = tmp_path / "import_project_test.vease" - with zipfile.ZipFile(z, "w", compression=zipfile.ZIP_DEFLATED) as zipf: - zipf.writestr("snapshot.json", json.dumps(snapshot)) - zipf.write(str(temp_db), "project.db") - - with open(z, "rb") as f: - resp = client.post( - route, - data={"file": (f, "import_project_test.vease")}, - content_type="multipart/form-data", - ) - - assert resp.status_code == 200 - assert resp.get_json().get("snapshot") == snapshot - assert os.path.exists(db_path) - - from opengeodeweb_microservice.database import connection - - client.application.config["DATA_FOLDER_PATH"] = original_data_folder - test_db_path = os.environ.get("TEST_DB_PATH") - if test_db_path: - connection.init_database(test_db_path, create_tables=True) - - client.application.config["DATA_FOLDER_PATH"] = original_data_folder - - -def test_save_viewable_workflow_from_object(client: FlaskClient) -> None: - route = "/opengeodeweb_back/create/point" - point_data = { - "name": "workflow_point_3d", - "x": 0.0, - "y": 0.0, - "z": 0.0, - } - - response = client.post(route, json=point_data) - assert response.status_code == 200 - - data_id = response.get_json()["id"] - assert isinstance(data_id, str) and len(data_id) > 0 - assert response.get_json()["geode_object_type"] == "PointSet3D" - assert response.get_json()["viewable_file"].endswith(".vtp") - - -def test_import_extension_route(client: FlaskClient, tmp_path: Path) -> None: - """Test importing a .vext extension file.""" - route = "/opengeodeweb_back/import_extension" - original_data_folder = client.application.config["DATA_FOLDER_PATH"] - new_data_folder = os.path.join(str(tmp_path), "extension_test_data") - client.application.config["DATA_FOLDER_PATH"] = new_data_folder - client.application.config["EXTENSIONS_FOLDER_PATH"] = os.path.join( - new_data_folder, "extensions" - ) - vext_path = tmp_path / "test-extension-1.0.0.vext" - with zipfile.ZipFile(vext_path, "w", compression=zipfile.ZIP_DEFLATED) as zipf: - zipf.writestr( - "test-extension-extension.es.js", - "export const metadata = { id: 'test-extension', name: 'Test Extension' };", - ) - zipf.writestr("test-extension-back", "#!/bin/bash\necho 'mock backend'") - zipf.writestr("test-extension.css", ".test { color: red; }") - with open(vext_path, "rb") as f: - response = client.post( - route, - data={"file": (f, "test-extension-1.0.0.vext")}, - content_type="multipart/form-data", - ) - assert response.status_code == 200 - json_data = response.get_json() - assert "extension_name" in json_data - assert "frontend_content" in json_data - assert "backend_path" in json_data - assert json_data["extension_name"] == "test-extension" - extensions_folder = os.path.join( - client.application.config["DATA_FOLDER_PATH"], "extensions" - ) - extension_path = os.path.join(extensions_folder, "test-extension") - assert os.path.exists(extension_path) - - # Verify frontend content is returned - frontend_content = json_data["frontend_content"] - assert isinstance(frontend_content, str) - assert len(frontend_content) > 0 - assert "export const metadata" in frontend_content - - backend_exec = json_data["backend_path"] - assert os.path.exists(backend_exec) - assert os.access(backend_exec, os.X_OK) - client.application.config["DATA_FOLDER_PATH"] = original_data_folder - - -def test_import_extension_invalid_file(client: FlaskClient, tmp_path: Path) -> None: - """Test importing an invalid .vext file (missing dist folder).""" - route = "/opengeodeweb_back/import_extension" - original_data_folder = client.application.config["DATA_FOLDER_PATH"] - new_data_folder = os.path.join(str(tmp_path), "extension_invalid_test") - client.application.config["DATA_FOLDER_PATH"] = new_data_folder - client.application.config["EXTENSIONS_FOLDER_PATH"] = os.path.join( - new_data_folder, "extensions" - ) - vext_path = tmp_path / "invalid-extension.vext" - with zipfile.ZipFile(vext_path, "w") as zipf: - zipf.writestr("README.md", "This is invalid") - with open(vext_path, "rb") as f: - response = client.post( - route, - data={"file": (f, "invalid-extension.vext")}, - content_type="multipart/form-data", - ) - assert response.status_code == 400 - client.application.config["DATA_FOLDER_PATH"] = original_data_folder - - -def test_import_extension_wrong_extension(client: FlaskClient, tmp_path: Path) -> None: - """Test uploading a file with wrong extension.""" - route = "/opengeodeweb_back/import_extension" - wrong_file = tmp_path / "not-an-extension.zip" - with open(wrong_file, "wb") as f: - f.write(b"test content") - with open(wrong_file, "rb") as f: - response = client.post( - route, - data={"file": (f, "not-an-extension.zip")}, - content_type="multipart/form-data", - ) - assert response.status_code == 400 diff --git a/tests/test_routes.py b/tests/test_routes.py index fad9b009..c20b8907 100644 --- a/tests/test_routes.py +++ b/tests/test_routes.py @@ -5,6 +5,9 @@ from werkzeug.datastructures import FileStorage from flask.testing import FlaskClient from werkzeug.test import TestResponse +from pathlib import Path +import json +import zipfile # Local application imports from opengeodeweb_microservice.database.data import Data @@ -424,3 +427,172 @@ def get_full_data() -> test_utils.JsonData: return {"geode_object_type": "BRep"} test_utils.test_route_wrong_params(client, route, get_full_data) + + +def test_model_components(client: FlaskClient) -> None: + geode_object_type = "BRep" + filename = "cube.og_brep" + response = test_save_viewable_file(client, geode_object_type, filename) + assert response.status_code == 200 + assert "mesh_components" in response.get_json() + mesh_components = response.get_json()["mesh_components"] + assert isinstance(mesh_components, list) + assert len(mesh_components) > 0 + for mesh_component in mesh_components: + assert isinstance(mesh_component, object) + assert isinstance(mesh_component["geode_id"], str) + assert isinstance(mesh_component["viewer_id"], int) + assert isinstance(mesh_component["name"], str) + assert isinstance(mesh_component["type"], str) + assert isinstance(mesh_component["boundaries"], list) + for boundary_uuid in mesh_component["boundaries"]: + assert isinstance(boundary_uuid, str) + assert isinstance(mesh_component["internals"], list) + for internal_uuid in mesh_component["internals"]: + assert isinstance(internal_uuid, str) + assert "collection_components" in response.get_json() + collection_components = response.get_json()["collection_components"] + assert isinstance(collection_components, list) + for collection_component in collection_components: + assert isinstance(collection_component, object) + assert isinstance(collection_component["geode_id"], str) + assert isinstance(collection_component["name"], str) + assert isinstance(collection_component["items"], list) + for item_uuid in collection_component["items"]: + assert isinstance(item_uuid, str) + + +def test_export_project_route(client: FlaskClient, tmp_path: Path) -> None: + route = "/opengeodeweb_back/export_project" + snapshot = { + "styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}} + } + filename = "export_project_test.vease" + project_folder = client.application.config["DATA_FOLDER_PATH"] + os.makedirs(project_folder, exist_ok=True) + 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", + viewer_elements_type="default", + native_file="native.txt", + ) + data2 = Data( + id="test_data_2", + geode_object="Section", + viewer_object="Section", + viewer_elements_type="default", + native_file="native.txt", + ) + 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, "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, "native.txt"), "w") as f: + f.write("native file content") + + response = client.post(route, json={"snapshot": snapshot, "filename": filename}) + assert response.status_code == 200 + assert response.headers.get("new-file-name") == filename + assert response.mimetype == "application/octet-binary" + response.direct_passthrough = False + 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/native.txt" in names + assert "test_data_2/native.txt" in names + + 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 = { + "styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}} + } + + original_data_folder = client.application.config["DATA_FOLDER_PATH"] + client.application.config["DATA_FOLDER_PATH"] = os.path.join( + str(tmp_path), "project_data" + ) + db_path = os.path.join(client.application.config["DATA_FOLDER_PATH"], "project.db") + + import sqlite3, zipfile, json + + temp_db = tmp_path / "temp_project.db" + conn = sqlite3.connect(str(temp_db)) + conn.execute( + "CREATE TABLE datas (id TEXT PRIMARY KEY, geode_object TEXT, viewer_object TEXT, viewer_elements_type TEXT, native_file TEXT, " + "viewable_file TEXT, light_viewable_file TEXT)" + ) + conn.commit() + conn.close() + + z = tmp_path / "import_project_test.vease" + with zipfile.ZipFile(z, "w", compression=zipfile.ZIP_DEFLATED) as zipf: + zipf.writestr("snapshot.json", json.dumps(snapshot)) + zipf.write(str(temp_db), "project.db") + + with open(z, "rb") as f: + resp = client.post( + route, + data={"file": (f, "import_project_test.vease")}, + content_type="multipart/form-data", + ) + + assert resp.status_code == 200 + assert resp.get_json().get("snapshot") == snapshot + assert os.path.exists(db_path) + + from opengeodeweb_microservice.database import connection + + client.application.config["DATA_FOLDER_PATH"] = original_data_folder + test_db_path = os.environ.get("TEST_DB_PATH") + if test_db_path: + connection.init_database(test_db_path, create_tables=True) + + client.application.config["DATA_FOLDER_PATH"] = original_data_folder + + +def test_save_viewable_workflow_from_object(client: FlaskClient) -> None: + route = "/opengeodeweb_back/create/point" + point_data = { + "name": "workflow_point_3d", + "x": 0.0, + "y": 0.0, + "z": 0.0, + } + + response = client.post(route, json=point_data) + assert response.status_code == 200 + + data_id = response.get_json()["id"] + assert isinstance(data_id, str) and len(data_id) > 0 + assert response.get_json()["geode_object_type"] == "PointSet3D" + assert response.get_json()["viewable_file"].endswith(".vtp")