Skip to content

Commit

Permalink
Version 1.0.5 (#97)
Browse files Browse the repository at this point in the history 
* Improved robustness of delete file mechanism

* Improved file tests

* Bumped version to 1.0.5

---------

Co-authored-by: Oscar Bennet <oscar.bennet@hotmail.se>
Co-authored-by: EppChops <erik.00.berg@gmail.com>
Co-authored-by: Erik Berg <57296415+EppChops@users.noreply.github.com>
  • Loading branch information
@johnmartins @obennet @EppChops
4 people authored Jun 12, 2023
1 parent cfb5a14 commit 86467d9
Show file tree
Hide file tree
Showing 3 changed files with 142 additions and 125 deletions.
5 changes: 3 additions & 2 deletions sedbackend/apps/core/files/storage.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import uuid
import shutil
import os

from mysql.connector.pooling import PooledMySQLConnection
from fastapi.logger import logger
Expand Down Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion sedbackend/main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
260 changes: 138 additions & 122 deletions tests/apps/core/files/test_files.py
View file
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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])

# 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])

0 comments on commit 86467d9

Please sign in to comment.