test(files): update

Author: grololo06

QA/py/tests/api_wrappers.py

@@ -87,4 +87,19 @@ def api_check_job_failed(fastapi, job_id, expected_message):
 
 
 REMOVE_FILE_URL = "/user_files/rm/"
-UPLOAD_FILE_URL = "/user_files/"
+MY_FILES_URL = "/user_files/"
+
+
+def upload_file(fastapi, local_file, dest_path, auth):
+    with open(local_file, "rb") as fin:
+        upload_rsp = fastapi.post(
+            MY_FILES_URL,
+            headers=auth,
+            data={
+                "path": dest_path
+            },  # /!\ If no pathparam error-> random use-once directory!
+            files={"file": fin},
+        )
+        assert upload_rsp.status_code == 200
+        srv_file_path = upload_rsp.json()
+        return srv_file_path

QA/py/tests/holes.py

@@ -10,7 +10,7 @@
 from tests.credentials import USER_AUTH, CREATOR_AUTH
 from tests.test_classification import OBJECT_SET_SUMMARY_URL
 from tests.test_import import PLAIN_FILE_PATH, do_test_import
-from tests.api_wrappers import UPLOAD_FILE_URL
+from tests.api_wrappers import MY_FILES_URL
 
 MYFILES_URL = "/my_files/{sub_path}"
 COMMON_FILES_URL = "/common_files/?path={sub_path}"
@@ -58,7 +58,7 @@ def put_path(fastapi, path1="file", path2=None, tag=None, should_fail=False):
         files_params = {"file": (path1, fin)}
         params = {"path": path2, "tag": tag}
         upload_rsp = fastapi.post(
-            UPLOAD_FILE_URL, headers=CREATOR_AUTH, data=params, files=files_params
+            MY_FILES_URL, headers=CREATOR_AUTH, data=params, files=files_params
         )
         if upload_rsp.status_code == 200:
             assert not should_fail

QA/py/tests/test_import.py

@@ -25,8 +25,9 @@
     api_file_import,
     api_wait_for_stable_job,
     api_check_job_questions,
-    UPLOAD_FILE_URL,
+    MY_FILES_URL,
     REMOVE_FILE_URL,
+    upload_file,
 )
 from tests.credentials import ADMIN_AUTH, ADMIN_USER_ID, CREATOR_USER_ID, CREATOR_AUTH
 from tests.jobs import (
@@ -655,16 +656,10 @@ def test_import_breaking_unicity(fastapi):
 def test_uvp6_via_myfile(fastapi, caplog):
     prj_id = create_project(CREATOR_USER_ID, "UVP6 via MyFiles")
     local_file = SHARED_DIR / V6_FILE
-    with open(local_file, "rb") as fin:
-        upload_rsp = fastapi.post(
-            UPLOAD_FILE_URL,
-            headers=CREATOR_AUTH,
-            files={"file": fin},
-        )
-        assert upload_rsp.status_code == 200
+    upload_file(fastapi, local_file, local_file.name, CREATOR_AUTH)
     no_zip = local_file.name.replace(".zip", "")
 
-    list_rsp = fastapi.get(f"{UPLOAD_FILE_URL}", headers=CREATOR_AUTH)
+    list_rsp = fastapi.get(f"{MY_FILES_URL}", headers=CREATOR_AUTH)
     assert list_rsp.status_code == 200
     user_files = list_rsp.json()
     assert no_zip in [a_file["name"] for a_file in user_files["entries"]]

QA/py/tests/test_import_simple.py

@@ -25,7 +25,7 @@
 from tests.credentials import ADMIN_USER_ID, CREATOR_AUTH, CREATOR_USER_ID
 from tests.test_import import PLAIN_DIR, create_project, PLAIN_FILE_PATH
 from tests.jobs import check_job_ok
-from tests.api_wrappers import api_wait_for_stable_job, UPLOAD_FILE_URL
+from tests.api_wrappers import api_wait_for_stable_job, MY_FILES_URL
 
 IMPORT_IMAGES_URL = "/simple_import/{project_id}?dry_run={dry_run}"
 
@@ -125,7 +125,7 @@ def do_simple_import_images(fastapi, title):
     prj_id = create_project(CREATOR_USER_ID, title)
     with open(PLAIN_FILE_PATH, "rb") as fin:
         upload_rsp = fastapi.post(
-            UPLOAD_FILE_URL,
+            MY_FILES_URL,
             headers=CREATOR_AUTH,
             files={"file": fin},
         )

QA/py/tests/test_my_files.py

@@ -5,22 +5,20 @@
 # Exhibit some not-so-intuitive behavior of /my_files and associated upload
 #
 
-import logging
 import shutil
 from typing import Dict
 
 import pytest
 from starlette import status
 
 from tests.credentials import CREATOR_AUTH, CREATOR_USER_ID
-from tests.jobs import (
-    api_check_job_ok,
-)
 from tests.api_wrappers import (
     api_file_import,
     api_wait_for_stable_job,
     api_check_job_errors,
-    UPLOAD_FILE_URL,
+    MY_FILES_URL,
+    upload_file,
+    api_check_job_ok,
 )
 from tests.test_import import (
     SHARED_DIR,
@@ -41,20 +39,33 @@ def test_my_files(fastapi, tstlogs, title):
 
     DEST_FILE_NAME = "LOKI_46-24hours_01.zip"
     DEST_DIR_NAME = "LOKI_46-24hours_01"
+
     # Copy an existing test file into current dir, simulating client side
     shutil.copyfile(SHARED_DIR / V6_FILE, tstlogs / DEST_FILE_NAME)
     # Upload this file
-    upload_file(fastapi, DEST_FILE_NAME, DIRPATH + "/" + DEST_FILE_NAME, tstlogs)
+    remote_path = upload_file(
+        fastapi,
+        tstlogs / DEST_FILE_NAME,
+        DIRPATH + SEPARATOR + DEST_FILE_NAME,
+        CREATOR_AUTH,
+    )
+    assert DIRPATH in remote_path  # The subdirectory was created
 
     # And another
     DEST_FILE_NAME2 = "readme.txt"
     # Copy an existing test file into current dir, simulating client side
     shutil.copyfile(SHARED_DIR / "HOWTO.txt", tstlogs / DEST_FILE_NAME2)
     # Upload this file
-    upload_file(fastapi, DEST_FILE_NAME2, DIRPATH + "/" + DEST_FILE_NAME2, tstlogs)
+    upload_file(
+        fastapi,
+        tstlogs / DEST_FILE_NAME2,
+        DIRPATH + SEPARATOR + DEST_FILE_NAME2,
+        CREATOR_AUTH,
+    )
+    assert DIRPATH in remote_path  # The subdirectory was created
 
     # The pathparam becomes a top-level directory
-    list_rsp = fastapi.get(UPLOAD_FILE_URL + SEPARATOR, headers=CREATOR_AUTH)
+    list_rsp = fastapi.get(MY_FILES_URL + SEPARATOR, headers=CREATOR_AUTH)
     assert list_rsp.status_code == 200
     my_files_root: Dict = list_rsp.json()
     assert my_files_root["path"] == ""
@@ -73,17 +84,13 @@ def test_my_files(fastapi, tstlogs, title):
 
     # The files are stored in the subdirectory
     DIRDEST = DIRPATH + SEPARATOR + DEST_DIR_NAME
-    list_rsp = fastapi.get(UPLOAD_FILE_URL + SEPARATOR + DIRDEST, headers=CREATOR_AUTH)
+    list_rsp = fastapi.get(MY_FILES_URL + SEPARATOR + DIRDEST, headers=CREATOR_AUTH)
     assert list_rsp.status_code == 200
     my_files_subdir: Dict = list_rsp.json()
     assert my_files_subdir["path"] == DIRDEST
     assert (
         len(my_files_subdir["entries"]) == 1
     )  # The second file being .txt will have size 0 on re-read
-    # test no longer valid as the zip is deleted after successful unzip
-    # the_file = [fil for fil in my_files_subdir["entries"] if fil["size"] == 22654][0]
-    # del the_file["mtime"]  # Unpredictable
-    # assert the_file == {"name": DEST_FILE_NAME, "size": 22654, "type": "F"}
 
     # Import the file without the right path -> Error
     req = {"source_path": DEST_FILE_NAME}
@@ -94,33 +101,9 @@ def test_my_files(fastapi, tstlogs, title):
     errors = api_check_job_errors(fastapi, job_id)
     assert "FileNotFoundError" in "".join(errors)
 
-    # Import the file with the right path
-    # The below (unfortunately) hard-coded path is valid on the current configuration of EcoTaxa
-    file_path = "/tmp/ecotaxa_user.{}/{}/{}".format(
-        CREATOR_USER_ID,  # Should come from /api/users/me
-        DIRPATH,  # existing tag, created the on the first file creation with it
-        DEST_FILE_NAME,  # can come from an entry in GET /my_files/DIRPATH
-    )
-    req = {"source_path": file_path}
+    req = {"source_path": DIRDEST}
     rsp = api_file_import(fastapi, prj_id, req, auth=CREATOR_AUTH)
     assert rsp.status_code == status.HTTP_200_OK
-    job_id = rsp.json()["job_id"]
-    api_wait_for_stable_job(fastapi, job_id)
-    # api_check_job_ok(fastapi, job_id) # TODO: It's _not_ anymore the right path
-
-
-def upload_file(fastapi, dest_file_name, pathparam, tstlogs):
-    print("uploadfile----------" + str(dest_file_name), pathparam)
-    with open(tstlogs / dest_file_name, "rb") as fin:
-        upload_rsp = fastapi.post(
-            UPLOAD_FILE_URL,
-            headers=CREATOR_AUTH,
-            data={
-                "path": pathparam
-            },  # /!\ If no pathparam error-> random use-once directory!
-            files={"file": fin},
-        )
-        assert upload_rsp.status_code == 200
-        srv_file_path = upload_rsp.json()
-        print("srv_file", srv_file_path)
-        assert DIRPATH in srv_file_path
+    job = api_wait_for_stable_job(fastapi, rsp.json()["job_id"])
+    api_wait_for_stable_job(fastapi, job.id)
+    api_check_job_ok(fastapi, job.id)

py/BO/Bundle.py

@@ -7,6 +7,7 @@
 import shutil
 import zipfile
 from pathlib import Path
+
 # noinspection PyPackageRequirements
 from typing import Callable, List, Dict, Tuple, Generator, Set