diff --git a/sedbackend/apps/core/files/storage.py b/sedbackend/apps/core/files/storage.py index 3e44d957..483902b9 100644 --- a/sedbackend/apps/core/files/storage.py +++ b/sedbackend/apps/core/files/storage.py @@ -1,6 +1,5 @@ import uuid import shutil -import os from mysql.connector.pooling import PooledMySQLConnection from fastapi.logger import logger @@ -49,13 +48,15 @@ def db_delete_file(con: PooledMySQLConnection, file_id: int, current_user_id: in raise exc.PathMismatchException try: - os.remove(stored_file_path.path) delete_stmnt = MySQLStatementBuilder(con) delete_stmnt.delete(FILES_TABLE) \ .where('id=?', [file_id]) \ .execute(fetch_type=FetchType.FETCH_NONE) + os.remove(stored_file_path.path) except Exception: + logger.error(f'Unexpected error when deleting file from database or filesystem. ' + f'id = {file_id}, path = \"{stored_file_path.path}\"') raise exc.FileNotDeletedException return True diff --git a/sedbackend/main.py b/sedbackend/main.py index 86193b76..24216cc6 100644 --- a/sedbackend/main.py +++ b/sedbackend/main.py @@ -15,7 +15,7 @@ app = FastAPI( title="SED lab API", description="The SED lab API contains all HTTP operations available within the SED lab application.", - version="1.0.4", + version="1.0.5", ) app.include_router(api.router, prefix="/api") diff --git a/tests/apps/core/files/test_files.py b/tests/apps/core/files/test_files.py index dff0a9bc..bda9a0ac 100644 --- a/tests/apps/core/files/test_files.py +++ b/tests/apps/core/files/test_files.py @@ -1,7 +1,8 @@ import tempfile import tests.apps.core.projects.testutils as tu_proj import tests.apps.core.users.testutils as tu_users -import tests.apps.core.files.testutils as tu +import tests.apps.core.files.testutils as tu_files +import tests.testutils as tu import sedbackend.apps.core.files.implementation as impl import sedbackend.apps.core.files.models as models @@ -10,133 +11,148 @@ def test_get_file(client, std_headers, std_user): - #Setup - current_user = impl_users.impl_get_user_with_username(std_user.username) - project = tu_proj.seed_random_project(current_user.id) - subp = tu_proj.seed_random_subproject(current_user.id, project.id) - - - tmp_file = tempfile.SpooledTemporaryFile() - tmp_file.write(b"Hello World!") - - post_file = models.StoredFilePost( - filename="hello", - owner_id=current_user.id, - extension=".txt", - file_object=tmp_file, - subproject_id=subp.id - ) - saved_file = impl.impl_save_file(post_file) - - #Act - res = client.get(f"/api/core/files/{saved_file.id}/download", - headers=std_headers) - - #Assert - assert res.status_code == 200 - - #Cleanup - tu.delete_files([saved_file], [current_user]) - tu_proj.delete_subprojects([subp]) - tu_proj.delete_projects([project]) + # Setup + current_user = impl_users.impl_get_user_with_username(std_user.username) + project = tu_proj.seed_random_project(current_user.id) + subp = tu_proj.seed_random_subproject(current_user.id, project.id) + + # Seed file + rand_str = tu.random_str(100, 200) + tmp_file = tempfile.SpooledTemporaryFile() + tmp_file.write(bytes(rand_str, 'utf-8')) + + # Ensure that the file is written to disk + tmp_file.seek(0) + tmp_file.flush() + + post_file = models.StoredFilePost( + filename="hello.txt", + owner_id=current_user.id, + extension=".txt", + file_object=tmp_file, + subproject_id=subp.id + ) + saved_file = impl.impl_save_file(post_file) + + # Act + res = client.get(f"/api/core/files/{saved_file.id}/download", + headers=std_headers) + + # Assert + assert res.status_code == 200 + assert res.headers['Content-Disposition'] == f'attachment; filename="{saved_file.filename}"' + assert res.headers['Content-Type'] == 'text/plain; charset=utf-8' + assert res.content == bytes(rand_str, 'utf-8') + + # Cleanup + tu_files.delete_files([saved_file], [current_user]) + tu_proj.delete_subprojects([subp]) + tu_proj.delete_projects([project]) def test_delete_file_admin(client, admin_headers, admin_user): - #Setup - std_user = tu_users.seed_random_user(admin=False, disabled=False) - adm_user = impl_users.impl_get_user_with_username(admin_user.username) - project = tu_proj.seed_random_project(std_user.id, {adm_user.id: AccessLevel.ADMIN}) - subp = tu_proj.seed_random_subproject(std_user.id, project.id) - - - tmp_file = tempfile.SpooledTemporaryFile() - tmp_file.write(b"Hello World!") - - post_file = models.StoredFilePost( - filename="hello", - owner_id=std_user.id, - extension=".txt", - file_object=tmp_file, - subproject_id=subp.id - ) - saved_file = impl.impl_save_file(post_file) - - #Act - res = client.delete(f"/api/core/files/{saved_file.id}/delete", - headers=admin_headers) - - #Assert - assert res.status_code == 200 - - #Cleanup - tu_proj.delete_subprojects([subp]) - tu_proj.delete_projects([project]) - tu_users.delete_users([std_user]) - - + # Setup + std_user = tu_users.seed_random_user(admin=False, disabled=False) + adm_user = impl_users.impl_get_user_with_username(admin_user.username) + project = tu_proj.seed_random_project(std_user.id, {adm_user.id: AccessLevel.ADMIN}) + subp = tu_proj.seed_random_subproject(std_user.id, project.id) + + # Seed file + tmp_file = tempfile.SpooledTemporaryFile() + tmp_file.write(b"Hello World!") + + # Write to disk + tmp_file.seek(0) + tmp_file.flush() + + post_file = models.StoredFilePost( + filename="hello", + owner_id=std_user.id, + extension=".txt", + file_object=tmp_file, + subproject_id=subp.id + ) + saved_file = impl.impl_save_file(post_file) + + # Act + res = client.delete(f"/api/core/files/{saved_file.id}/delete", + headers=admin_headers) + + # Assert + assert res.status_code == 200 + + # Cleanup + tu_proj.delete_subprojects([subp]) + tu_proj.delete_projects([project]) + tu_users.delete_users([std_user]) def test_delete_file_standard(client, std_headers, std_user): - #Setup - file_owner = tu_users.seed_random_user(admin=False, disabled=False) - current_user = impl_users.impl_get_user_with_username(std_user.username) - project = tu_proj.seed_random_project(file_owner.id, {current_user.id: AccessLevel.READONLY}) - subp = tu_proj.seed_random_subproject(file_owner.id, project.id) - - - tmp_file = tempfile.SpooledTemporaryFile() - tmp_file.write(b"Hello World!") - - post_file = models.StoredFilePost( - filename="hello", - owner_id=file_owner.id, - extension=".txt", - file_object=tmp_file, - subproject_id=subp.id - ) - saved_file = impl.impl_save_file(post_file) - - #Act - res = client.delete(f"/api/core/files/{saved_file.id}/delete", - headers=std_headers) - - #Assert - assert res.status_code == 403 #403 forbidden, should not be able to access resource - - #Cleanup - tu_proj.delete_subprojects([subp]) - tu_proj.delete_projects([project]) - tu_users.delete_users([file_owner]) - + # Setup + file_owner = tu_users.seed_random_user(admin=False, disabled=False) + current_user = impl_users.impl_get_user_with_username(std_user.username) + project = tu_proj.seed_random_project(file_owner.id, {current_user.id: AccessLevel.READONLY}) + subp = tu_proj.seed_random_subproject(file_owner.id, project.id) + + # Seed file + tmp_file = tempfile.SpooledTemporaryFile() + tmp_file.write(b"Hello World!") + + # Write to disk + tmp_file.seek(0) + tmp_file.flush() + + post_file = models.StoredFilePost( + filename="hello", + owner_id=file_owner.id, + extension=".txt", + file_object=tmp_file, + subproject_id=subp.id + ) + saved_file = impl.impl_save_file(post_file) + + # Act + res = client.delete(f"/api/core/files/{saved_file.id}/delete", + headers=std_headers) + + # Assert + assert res.status_code == 403 # 403 forbidden, should not be able to access resource + + # Cleanup + tu_proj.delete_subprojects([subp]) + tu_proj.delete_projects([project]) + tu_users.delete_users([file_owner]) def test_delete_file_owner(client, std_headers, std_user): - #Setup - current_user = impl_users.impl_get_user_with_username(std_user.username) - project = tu_proj.seed_random_project(current_user.id) - subp = tu_proj.seed_random_subproject(current_user.id, project.id) - - - tmp_file = tempfile.SpooledTemporaryFile() - tmp_file.write(b"Hello World!") - - post_file = models.StoredFilePost( - filename="hello", - owner_id=current_user.id, - extension=".txt", - file_object=tmp_file, - subproject_id=subp.id - ) - saved_file = impl.impl_save_file(post_file) - - #Act - res = client.delete(f"/api/core/files/{saved_file.id}/delete", - headers=std_headers) - - #Assert - assert res.status_code == 200 - - #Cleanup - tu_proj.delete_subprojects([subp]) - tu_proj.delete_projects([project]) - \ No newline at end of file + # Setup + current_user = impl_users.impl_get_user_with_username(std_user.username) + project = tu_proj.seed_random_project(current_user.id) + subp = tu_proj.seed_random_subproject(current_user.id, project.id) + + # Seed file + tmp_file = tempfile.SpooledTemporaryFile() + tmp_file.write(b"Hello World!") + # Write to disk + tmp_file.seek(0) + tmp_file.flush() + + post_file = models.StoredFilePost( + filename="hello", + owner_id=current_user.id, + extension=".txt", + file_object=tmp_file, + subproject_id=subp.id + ) + saved_file = impl.impl_save_file(post_file) + + # Act + res = client.delete(f"/api/core/files/{saved_file.id}/delete", + headers=std_headers) + + # Assert + assert res.status_code == 200 + + # Cleanup + tu_proj.delete_subprojects([subp]) + tu_proj.delete_projects([project])