test

Author: MaxNumerique

src/opengeodeweb_back/geode_functions.py

@@ -291,3 +291,14 @@ def create_coordinate_system(
     create_crs(
         geode_object, data, name, input_coordiante_system, output_coordiante_system
     )
+
+
+def file_exists_in_upload(filename: str) -> bool:
+    """Vérifie si un fichier existe dans le dossier d'upload."""
+    file_path = upload_file_path(filename)
+    return os.path.exists(file_path)
+
+def file_exists_in_data(data_id: str, filename: str) -> bool:
+    """Vérifie si un fichier existe dans le dossier de données."""
+    file_path = data_file_path(data_id, filename)
+    return os.path.exists(file_path)

src/opengeodeweb_back/routes/blueprint_routes.py

@@ -76,10 +76,31 @@ def upload_file():
     UPLOAD_FOLDER = flask.current_app.config["UPLOAD_FOLDER"]
     if not os.path.exists(UPLOAD_FOLDER):
         os.mkdir(UPLOAD_FOLDER)
+    
     file = flask.request.files["file"]
     filename = werkzeug.utils.secure_filename(os.path.basename(file.filename))
-    file.save(os.path.join(UPLOAD_FOLDER, filename))
-    return flask.make_response({"message": "File uploaded"}, 201)
+    file_path = os.path.join(UPLOAD_FOLDER, filename)
+    
+    # Vérifier si le fichier existe déjà
+    file_existed = os.path.exists(file_path)
+    if file_existed:
+        replace_if_exists = flask.request.form.get('replace_if_exists', 'false').lower() == 'true'
+        if not replace_if_exists:
+            return flask.make_response({
+                "error": "File already exists",
+                "message": f"Le fichier '{filename}' existe déjà. Utilisez 'replace_if_exists=true' pour le remplacer.",
+                "existing_file": filename
+            }, 409)  # 409 Conflict
+    
+    file.save(file_path)
+    
+    response_data = {
+        "message": "File uploaded",
+        "filename": filename,
+        "replaced": file_existed
+    }
+    
+    return flask.make_response(response_data, 201)
 
 
 with open(

src/opengeodeweb_back/routes/schemas/upload_file.json

@@ -9,7 +9,12 @@
     "filename": {
       "type": "string",
       "minLength": 1
+    },
+    "replace_if_exists": {
+      "type": "boolean",
+      "default": false
     }
   },
+  "required": ["filename"],
   "additionalProperties": false
 }

tests/conftest.py

@@ -13,8 +13,15 @@
 
 @pytest.fixture(scope="session", autouse=True)
 def copy_data():
+    # Nettoyer les dossiers de test
     shutil.rmtree("./data", ignore_errors=True)
+    shutil.rmtree("./uploads", ignore_errors=True)
+    
+    # Copier les données de test
     shutil.copytree("./tests/data/", f"./data/{TEST_ID}/", dirs_exist_ok=True)
+    
+    # Créer le dossier uploads pour les tests
+    os.makedirs("./uploads", exist_ok=True)
 
 
 @pytest.fixture

tests/test_geode_functions.py

@@ -327,3 +327,41 @@ def test_geode_objects_output_extensions():
                         output_extension_value,
                     ) in output_geode_object_value.items():
                         assert type(output_extension_value["is_saveable"]) is bool
+
+
+def test_file_exists_in_upload():
+    """Test de la fonction file_exists_in_upload."""
+    # Test avec un fichier qui n'existe pas
+    assert geode_functions.file_exists_in_upload("nonexistent.txt") == False
+    
+    # Test avec un fichier qui existe (après l'avoir créé)
+    test_filename = "test_exists.txt"
+    test_path = geode_functions.upload_file_path(test_filename)
+    os.makedirs(os.path.dirname(test_path), exist_ok=True)
+    with open(test_path, 'w') as f:
+        f.write("test")
+    
+    assert geode_functions.file_exists_in_upload(test_filename) == True
+    
+    # Nettoyer
+    os.remove(test_path)
+
+def test_file_exists_in_data():
+    """Test de la fonction file_exists_in_data."""
+    test_data_id = "test_data_id"
+    test_filename = "test_data.txt"
+    
+    # Test avec un fichier qui n'existe pas
+    assert geode_functions.file_exists_in_data(test_data_id, test_filename) == False
+    
+    # Test avec un fichier qui existe (après l'avoir créé)
+    test_path = geode_functions.data_file_path(test_data_id, test_filename)
+    os.makedirs(os.path.dirname(test_path), exist_ok=True)
+    with open(test_path, 'w') as f:
+        f.write("test")
+    
+    assert geode_functions.file_exists_in_data(test_data_id, test_filename) == True
+    
+    # Nettoyer
+    os.remove(test_path)
+    os.rmdir(os.path.dirname(test_path))

tests/test_routes.py

@@ -46,11 +46,70 @@ def get_full_data():
 
 
 def test_upload_file(client, filename="corbi.og_brep"):
+    # Test upload normal
     response = client.put(
         f"/upload_file",
-        data={"file": FileStorage(open(f"./tests/{filename}", "rb"))},
+        data={"file": FileStorage(open(f"./tests/data/{filename}", "rb"))},
     )
+    assert response.status_code == 201
+    assert response.json["message"] == "File uploaded"
+    assert response.json["filename"] == filename
+    assert response.json["replaced"] == False
+
+    # Test upload du même fichier sans remplacement (doit échouer)
+    response = client.put(
+        f"/upload_file",
+        data={
+            "file": FileStorage(open(f"./tests/data/{filename}", "rb")),
+            "replace_if_exists": "false"
+        },
+    )
+    assert response.status_code == 409
+    assert "File already exists" in response.json["error"]
+
+    # Test upload du même fichier avec remplacement (doit réussir)
+    response = client.put(
+        f"/upload_file",
+        data={
+            "file": FileStorage(open(f"./tests/data/{filename}", "rb")),
+            "replace_if_exists": "true"
+        },
+    )
+    assert response.status_code == 201
+    assert response.json["message"] == "File uploaded"
+    assert response.json["filename"] == filename
+    assert response.json["replaced"] == True
 
+
+def test_upload_file_new_parameter_validation(client):
+    """Test que le paramètre replace_if_exists est optionnel."""
+    route = "/upload_file"
+    
+    # Test avec replace_if_exists = true
+    response = client.put(
+        route,
+        data={
+            "file": FileStorage(open("./tests/data/corbi.og_brep", "rb")),
+            "replace_if_exists": "true"
+        },
+    )
+    assert response.status_code == 201
+    
+    # Test avec replace_if_exists = false
+    response = client.put(
+        route,
+        data={
+            "file": FileStorage(open("./tests/data/test.og_brep", "rb")),
+            "replace_if_exists": "false"
+        },
+    )
+    assert response.status_code == 201
+    
+    # Test sans replace_if_exists (doit utiliser la valeur par défaut false)
+    response = client.put(
+        route,
+        data={"file": FileStorage(open("./tests/data/cube.og_brep", "rb"))},
+    )
     assert response.status_code == 201