feat(extensions): add new blueprint and tests for extension imports

Author: MaxNumerique

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -502,3 +502,80 @@ def import_project() -> flask.Response:
         except KeyError:
             snapshot = {}
     return flask.make_response({"snapshot": snapshot}, 200)
+
+
+@routes.route(
+    schemas_dict["import_extension"]["route"],
+    methods=schemas_dict["import_extension"]["methods"],
+)
+def import_extension() -> flask.Response:
+    """Import a .vext extension file and extract its contents."""
+    utils_functions.validate_request(flask.request, schemas_dict["import_extension"])
+
+    if "file" not in flask.request.files:
+        flask.abort(400, "No .vext file provided under 'file'")
+
+    vext_file = flask.request.files["file"]
+    assert vext_file.filename is not None
+    filename = werkzeug.utils.secure_filename(os.path.basename(vext_file.filename))
+
+    if not filename.lower().endswith(".vext"):
+        flask.abort(400, "Uploaded file must be a .vext")
+
+    # Create extensions directory in the data folder
+    data_folder_path: str = flask.current_app.config.get("DATA_FOLDER_PATH", "")
+    extensions_folder = os.path.join(data_folder_path, "extensions")
+    os.makedirs(extensions_folder, exist_ok=True)
+
+    # Extract extension name from filename (e.g., "vease-modeling-0.0.0.vext" -> "vease-modeling")
+    extension_name = (
+        filename.rsplit("-", 1)[0] if "-" in filename else filename.replace(".vext", "")
+    )
+    extension_path = os.path.join(extensions_folder, extension_name)
+
+    # Remove existing extension if present
+    if os.path.exists(extension_path):
+        shutil.rmtree(extension_path)
+
+    os.makedirs(extension_path, exist_ok=True)
+
+    # Extract the .vext file
+    vext_file.stream.seek(0)
+    with zipfile.ZipFile(vext_file.stream) as zip_archive:
+        zip_archive.extractall(extension_path)
+
+    # Find the extracted files
+    dist_path = os.path.join(extension_path, "dist")
+    if not os.path.exists(dist_path):
+        flask.abort(400, "Invalid .vext file: missing dist folder")
+
+    # Look for the backend executable and frontend JS
+    backend_executable = None
+    frontend_file = None
+
+    for file in os.listdir(dist_path):
+        file_path = os.path.join(dist_path, file)
+        if os.path.isfile(file_path):
+            if file.endswith(".es.js"):
+                frontend_file = file_path
+            elif not file.endswith(".js") and not file.endswith(".css"):
+                # Assume it's the backend executable
+                backend_executable = file_path
+                # Make it executable
+                os.chmod(backend_executable, 0o755)
+
+    if not frontend_file:
+        flask.abort(400, "Invalid .vext file: missing frontend JavaScript")
+
+    if not backend_executable:
+        flask.abort(400, "Invalid .vext file: missing backend executable")
+
+    return flask.make_response(
+        {
+            "extension_name": extension_name,
+            "frontend_path": frontend_file,
+            "backend_path": backend_executable,
+            "extension_folder": extension_path,
+        },
+        200,
+    )

src/opengeodeweb_back/routes/schemas/__init__.py

@@ -15,3 +15,4 @@
 from .cell_attribute_names import *
 from .allowed_objects import *
 from .allowed_files import *
+from .import_extension import *

src/opengeodeweb_back/routes/schemas/import_extension.json

@@ -0,0 +1,10 @@
+{
+  "route": "/import_extension",
+  "methods": [
+    "POST"
+  ],
+  "type": "object",
+  "properties": {},
+  "required": [],
+  "additionalProperties": false
+}

src/opengeodeweb_back/routes/schemas/import_extension.py

@@ -0,0 +1,10 @@
+from dataclasses_json import DataClassJsonMixin
+from dataclasses import dataclass
+
+
+@dataclass
+class ImportExtension(DataClassJsonMixin):
+    def __post_init__(self) -> None:
+        print(self, flush=True)
+
+    pass

tests/test_models_routes.py

@@ -185,3 +185,82 @@ def test_save_viewable_workflow_from_object(client: FlaskClient) -> None:
     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")
+
+
+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"]
+    client.application.config["DATA_FOLDER_PATH"] = os.path.join(
+        str(tmp_path), "extension_test_data"
+    )
+    vext_path = tmp_path / "test-extension-1.0.0.vext"
+    with zipfile.ZipFile(vext_path, "w", compression=zipfile.ZIP_DEFLATED) as zipf:
+        zipf.writestr(
+            "dist/test-extension-extension.es.js",
+            "export const metadata = { id: 'test-extension', name: 'Test Extension' };",
+        )
+        zipf.writestr("dist/test-extension-back", "#!/bin/bash\necho 'mock backend'")
+        zipf.writestr("dist/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_path" in json_data
+    assert "backend_path" in json_data
+    assert "extension_folder" 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)
+    dist_path = os.path.join(extension_path, "dist")
+    assert os.path.exists(dist_path)
+    frontend_js = json_data["frontend_path"]
+    assert os.path.exists(frontend_js)
+    assert frontend_js.endswith("-extension.es.js")
+    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"]
+    client.application.config["DATA_FOLDER_PATH"] = os.path.join(
+        str(tmp_path), "extension_invalid_test"
+    )
+    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