import route and schema

Author: MaxNumerique

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -1,6 +1,7 @@
 # Standard library imports
 import os
 import time
+import shutil
 
 # Third party imports
 import flask
@@ -286,12 +287,61 @@ def export_project() -> flask.Response:
     filename: str = params.filename or f"project_{int(time.time())}.zip"
     export_zip_path = os.path.join(upload_folder, filename)
 
-    with zipfile.ZipFile(export_zip_path, "w", compression=8) as zipf:
+    with zipfile.ZipFile(export_zip_path, "w", compression=8) as zip_file:
         pattern = os.path.join(data_folder_path, "**", "*")
         for full_path in glob.glob(pattern, recursive=True):
             if os.path.isfile(full_path):
                 archive_name = os.path.relpath(full_path, start=data_folder_path)
-                zipf.write(full_path, archive_name)
-        zipf.writestr("snapshot.json", flask.json.dumps(params.snapshot))
+                zip_file.write(full_path, archive_name)
+        zip_file.writestr("snapshot.json", flask.json.dumps(params.snapshot))
 
     return utils_functions.send_file(upload_folder, [export_zip_path], filename)
+
+
+@routes.route(
+    schemas_dict["import_project"]["route"],
+    methods=schemas_dict["import_project"]["methods"],
+)
+def import_project() -> flask.Response:
+    if flask.request.method == "OPTIONS":
+        return flask.make_response({}, 200)
+    if "file" not in flask.request.files:
+        flask.abort(400, "No zip file provided under 'file'")
+
+    zip_file = flask.request.files["file"]
+    filename = werkzeug.utils.secure_filename(os.path.basename(zip_file.filename))
+    if not filename.lower().endswith(".zip"):
+        flask.abort(400, "Uploaded file must be a .zip")
+
+    data_folder_path: str = flask.current_app.config["DATA_FOLDER_PATH"]
+    os.makedirs(data_folder_path, exist_ok=True)
+    for entry in os.listdir(data_folder_path):
+        entry_path = os.path.join(data_folder_path, entry)
+        try:
+            if os.path.isdir(entry_path):
+                shutil.rmtree(entry_path, ignore_errors=True)
+            else:
+                os.remove(entry_path)
+        except FileNotFoundError:
+            pass
+        except PermissionError:
+            flask.abort(423, "Project files are locked; cannot overwrite")
+
+    zip_file.stream.seek(0)
+    with zipfile.ZipFile(zip_file.stream) as zf:
+        base = os.path.abspath(data_folder_path)
+        for member in zf.namelist():
+            target = os.path.abspath(
+                os.path.normpath(os.path.join(base, member))
+            )
+            if not (target == base or target.startswith(base + os.sep)):
+                flask.abort(400, "Zip contains unsafe paths")
+        zf.extractall(data_folder_path)
+        snapshot = {}
+        try:
+            raw = zf.read("snapshot.json").decode("utf-8")
+            snapshot = flask.json.loads(raw)
+        except KeyError:
+            snapshot = {}
+
+    return flask.make_response({"snapshot": snapshot}, 200)

src/opengeodeweb_back/routes/schemas/__init__.py

@@ -13,3 +13,4 @@
 from .allowed_objects import *
 from .allowed_files import *
 from .export_project import *
+from .import_project import *

src/opengeodeweb_back/routes/schemas/import_project.json

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

src/opengeodeweb_back/routes/schemas/import_project.py

@@ -0,0 +1,7 @@
+from dataclasses_json import DataClassJsonMixin
+from dataclasses import dataclass
+
+
+@dataclass
+class ImportProject(DataClassJsonMixin):
+    pass

tests/test_models_routes.py

@@ -60,7 +60,7 @@ def test_extract_brep_uuids(client, test_id):
         assert isinstance(uuid_dict, dict)
 
 
-def test_export_project_route(client):
+def test_export_project_route(client, tmp_path):
     route = "/opengeodeweb_back/export_project"
     snapshot = {"styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}}}
     filename = "export_project_test.zip"
@@ -69,15 +69,48 @@ def test_export_project_route(client):
     assert response.headers.get("new-file-name") == filename
     assert response.mimetype == "application/octet-binary"
     response.direct_passthrough = False
-    data = response.get_data()
-    with zipfile.ZipFile(io.BytesIO(data), "r") as zf:
-        names = zf.namelist()
+    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(zf.read("snapshot.json").decode("utf-8"))
+        parsed = json.loads(zip_file.read("snapshot.json").decode("utf-8"))
         assert parsed == snapshot
         assert "1/project.db" in names
     response.close()
     upload_folder = client.application.config["UPLOAD_FOLDER"]
     export_path = os.path.join(upload_folder, filename)
     if os.path.exists(export_path):
         os.remove(export_path)
+
+
+def test_import_project_route(client, tmp_path):
+    route = "/opengeodeweb_back/import_project"
+    snapshot = {"styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}}}
+
+    data_folder = client.application.config["DATA_FOLDER_PATH"]
+    pre_existing_db_path = os.path.join(data_folder, "1", "project.db")
+    os.makedirs(os.path.dirname(pre_existing_db_path), exist_ok=True)
+    with open(pre_existing_db_path, "wb") as file:
+        file.write(b"old_db_content")
+
+    tmp_zip = tmp_path / "import_project_test.zip"
+    new_database_bytes = b"new_db_content"
+    with zipfile.ZipFile(tmp_zip, "w", compression=zipfile.ZIP_DEFLATED) as zip_file:
+        zip_file.writestr("snapshot.json", json.dumps(snapshot))
+        zip_file.writestr("1/project.db", new_database_bytes)
+
+    with open(tmp_zip, "rb") as file:
+        response = client.post(
+            route,
+            data={"file": (file, "import_project_test.zip")},
+            content_type="multipart/form-data",
+        )
+
+    assert response.status_code == 200
+    assert response.json.get("snapshot") == snapshot
+
+    assert os.path.exists(pre_existing_db_path)
+    with open(pre_existing_db_path, "rb") as file:
+        assert file.read() == new_database_bytes