feat(utils): add 7z support for matrices import

antarest/matrixstore/service.py

@@ -12,6 +12,7 @@
 import numpy as np
 from fastapi import UploadFile
 from numpy import typing as npt
+from py7zr import SevenZipFile
 
 from antarest.core.config import Config
 from antarest.core.filetransfer.model import FileDownloadTaskDTO
@@ -187,12 +188,22 @@ def create_by_importation(self, file: UploadFile, is_json: bool = False) -> List
                 with contextlib.closing(f):
                     buffer = io.BytesIO(f.read())
                 matrix_info: List[MatrixInfoDTO] = []
-                with zipfile.ZipFile(buffer) as zf:
-                    for info in zf.infolist():
-                        if info.is_dir() or info.filename in EXCLUDED_FILES:
-                            continue
-                        matrix_id = self._file_importation(zf.read(info.filename), is_json=is_json)
-                        matrix_info.append(MatrixInfoDTO(id=matrix_id, name=info.filename))
+                if file.filename.endswith("zip"):
+                    with zipfile.ZipFile(buffer) as zf:
+                        for info in zf.infolist():
+                            if info.is_dir() or info.filename in EXCLUDED_FILES:
+                                continue
+                            matrix_id = self._file_importation(zf.read(info.filename), is_json=is_json)
+                            matrix_info.append(MatrixInfoDTO(id=matrix_id, name=info.filename))
+                else:
+                    with SevenZipFile(buffer, "r") as szf:
+                        for info in szf.list():
+                            if info.is_directory or info.filename in EXCLUDED_FILES:  # type:ignore
+                                continue
+                            file_content = next(iter(szf.read(info.filename).values()))
+                            matrix_id = self._file_importation(file_content.read(), is_json=is_json)
+                            matrix_info.append(MatrixInfoDTO(id=matrix_id, name=info.filename))
+                            szf.reset()
                 return matrix_info
             else:
                 matrix_id = self._file_importation(f.read(), is_json=is_json)

tests/integration/test_integration.py

@@ -224,31 +224,6 @@ def test_main(client: TestClient, admin_access_token: str, study_id: str) -> Non
     )
     assert len(res.json()) == 4
 
-    # tests outputs import for .zip
-    output_path_zip = ASSETS_DIR / "output_adq.zip"
-    client.post(
-        f"/v1/studies/{study_id}/output",
-        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
-        files={"output": io.BytesIO(output_path_zip.read_bytes())},
-    )
-    res = client.get(
-        f"/v1/studies/{study_id}/outputs",
-        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
-    )
-    assert len(res.json()) == 5
-    # tests outputs import for .7z
-    output_path_seven_zip = ASSETS_DIR / "output_adq.7z"
-    client.post(
-        f"/v1/studies/{study_id}/output",
-        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
-        files={"output": io.BytesIO(output_path_seven_zip.read_bytes())},
-    )
-    res = client.get(
-        f"/v1/studies/{study_id}/outputs",
-        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
-    )
-    assert len(res.json()) == 6
-
     # study creation
     created = client.post(
         "/v1/studies?name=foo",
@@ -2466,6 +2441,60 @@ def test_import(client: TestClient, admin_access_token: str, study_id: str) -> N
     )
     assert res.status_code == 201
 
+    # tests outputs import for .zip
+    output_path_zip = ASSETS_DIR / "output_adq.zip"
+    client.post(
+        f"/v1/studies/{study_id}/output",
+        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
+        files={"output": io.BytesIO(output_path_zip.read_bytes())},
+    )
+    res = client.get(
+        f"/v1/studies/{study_id}/outputs",
+        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
+    )
+    assert len(res.json()) == 6
+
+    # tests outputs import for .7z
+    output_path_seven_zip = ASSETS_DIR / "output_adq.7z"
+    client.post(
+        f"/v1/studies/{study_id}/output",
+        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
+        files={"output": io.BytesIO(output_path_seven_zip.read_bytes())},
+    )
+    res = client.get(
+        f"/v1/studies/{study_id}/outputs",
+        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
+    )
+    assert len(res.json()) == 7
+
+    # test matrices import for .zip file
+    matrices_zip_path = ASSETS_DIR / "matrices.zip"
+    res = client.post(
+        "/v1/matrix/_import",
+        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
+        files={"file": (matrices_zip_path.name, io.BytesIO(matrices_zip_path.read_bytes()), "application/zip")},
+    )
+    assert res.status_code == 200
+    result = res.json()
+    assert len(result) == 2
+    assert result[0]["name"] == "fr.txt"
+    assert result[1]["name"] == "it.txt"
+
+    # test matrices import for .7z file
+    matrices_seven_zip_path = ASSETS_DIR / "matrices.7z"
+    res = client.post(
+        "/v1/matrix/_import",
+        headers={"Authorization": f'Bearer {george_credentials["access_token"]}'},
+        files={
+            "file": (matrices_seven_zip_path.name, io.BytesIO(matrices_seven_zip_path.read_bytes()), "application/zip")
+        },
+    )
+    assert res.status_code == 200
+    result = res.json()
+    assert len(result) == 2
+    assert result[0]["name"] == "fr.txt"
+    assert result[1]["name"] == "it.txt"
+
 
 def test_copy(client: TestClient, admin_access_token: str, study_id: str) -> None:
     admin_headers = {"Authorization": f"Bearer {admin_access_token}"}