Skip to content

Add integration tests for Files API #552

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hectorcast-db merged 4 commits into from
Feb 22, 2024
Merged

Add integration tests for Files API #552

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
471b1e2
Add integration tests for Files API
bonnetn Feb 20, 2024
3281b7e
Bugfix
bonnetn Feb 20, 2024
aa83bb0
Remove change in generated file
bonnetn Feb 21, 2024
8638d99
Merge branch 'main' into add-integration-tests-for-files-api
bonnetn Feb 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 74 additions & 4 deletions tests/integration/test_files.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,13 +221,9 @@ def test_files_api_upload_download(ucws, random):
f = io.BytesIO(b"some text data")
target_file = f'/Volumes/main/{schema}/{volume}/filesit-{random()}.txt'
w.files.upload(target_file, f)
m = w.files.get_metadata(target_file)
assert m.content_type == 'application/octet-stream'
with w.files.download(target_file).contents as f:
assert f.read() == b"some text data"

w.files.delete(target_file)


def test_files_api_read_twice_from_one_download(ucws, random):
w = ucws
Expand All @@ -249,6 +245,80 @@ def test_files_api_read_twice_from_one_download(ucws, random):
res.read()


def test_files_api_delete_file(ucws, random):
w = ucws
schema = 'filesit-' + random()
volume = 'filesit-' + random()
with ResourceWithCleanup.create_schema(w, 'main', schema):
with ResourceWithCleanup.create_volume(w, 'main', schema, volume):
f = io.BytesIO(b"some text data")
target_file = f'/Volumes/main/{schema}/{volume}/filesit-{random()}.txt'
w.files.upload(target_file, f)
w.files.delete(target_file)


def test_files_api_get_metadata(ucws, random):
w = ucws
schema = 'filesit-' + random()
volume = 'filesit-' + random()
with ResourceWithCleanup.create_schema(w, 'main', schema):
with ResourceWithCleanup.create_volume(w, 'main', schema, volume):
f = io.BytesIO(b"some text data")
target_file = f'/Volumes/main/{schema}/{volume}/filesit-{random()}.txt'
w.files.upload(target_file, f)
m = w.files.get_metadata(target_file)
assert m.content_type == 'application/octet-stream'
assert m.content_length == 14
assert m.last_modified is not None


def test_files_api_create_directory(ucws, random):
w = ucws
schema = 'filesit-' + random()
volume = 'filesit-' + random()
with ResourceWithCleanup.create_schema(w, 'main', schema):
with ResourceWithCleanup.create_volume(w, 'main', schema, volume):
target_directory = f'/Volumes/main/{schema}/{volume}/filesit-{random()}/'
w.files.create_directory(target_directory)


def test_files_api_list_directory_contents(ucws, random):
w = ucws
schema = 'filesit-' + random()
volume = 'filesit-' + random()
with ResourceWithCleanup.create_schema(w, 'main', schema):
with ResourceWithCleanup.create_volume(w, 'main', schema, volume):
target_directory = f'/Volumes/main/{schema}/{volume}/filesit-{random()}'
w.files.upload(target_directory + "/file1.txt", io.BytesIO(b"some text data"))
w.files.upload(target_directory + "/file2.txt", io.BytesIO(b"some text data"))
w.files.upload(target_directory + "/file3.txt", io.BytesIO(b"some text data"))

result = list(w.files.list_directory_contents(target_directory))
assert len(result) == 3


def test_files_api_delete_directory(ucws, random):
w = ucws
schema = 'filesit-' + random()
volume = 'filesit-' + random()
with ResourceWithCleanup.create_schema(w, 'main', schema):
with ResourceWithCleanup.create_volume(w, 'main', schema, volume):
target_directory = f'/Volumes/main/{schema}/{volume}/filesit-{random()}/'
w.files.create_directory(target_directory)
w.files.delete_directory(target_directory)


def test_files_api_get_directory_metadata(ucws, random):
w = ucws
schema = 'filesit-' + random()
volume = 'filesit-' + random()
with ResourceWithCleanup.create_schema(w, 'main', schema):
with ResourceWithCleanup.create_volume(w, 'main', schema, volume):
target_directory = f'/Volumes/main/{schema}/{volume}/filesit-{random()}/'
w.files.create_directory(target_directory)
w.files.get_directory_metadata(target_directory)


@pytest.mark.benchmark
def test_files_api_download_benchmark(ucws, random):
w = ucws
Expand Down