Skip to content

Commit

Permalink
Fix prefix path parsing in OBJ plugin (#686)
Browse files Browse the repository at this point in the history
Co-authored-by: Zhiwei Liang <[email protected]>
Co-authored-by: Zhiwei Liang <[email protected]>
  • Loading branch information
3 people authored Dec 9, 2024
1 parent 61e9bc6 commit d4224ff
Show file tree
Hide file tree
Showing 2 changed files with 47 additions and 1 deletion.
2 changes: 1 addition & 1 deletion linodecli/plugins/obj/objects.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ def upload_object(
bucket = parsed.bucket
if "/" in parsed.bucket:
bucket = parsed.bucket.split("/")[0]
prefix = parsed.bucket.lstrip(f"{bucket}/")
prefix = parsed.bucket.removeprefix(f"{bucket}/")

upload_options = {
"Bucket": bucket,
Expand Down
46 changes: 46 additions & 0 deletions tests/integration/obj/test_obj_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,52 @@ def test_obj_single_file_single_bucket_with_prefix(
assert f1.read() == f2.read()


def test_obj_single_file_single_bucket_with_prefix_ltrim(
create_bucket: Callable[[Optional[str]], str],
generate_test_files: GetTestFilesType,
keys: Keys,
monkeypatch: MonkeyPatch,
):
patch_keys(keys, monkeypatch)
file_path = generate_test_files()[0]
bucket_name = create_bucket()
# using 'bk' in prefix to test out ltrim behaviour (bucket contains 'bk')
exec_test_command(
BASE_CMD + ["put", str(file_path), f"{bucket_name}/bkprefix"]
)
process = exec_test_command(BASE_CMD + ["la"])
output = process.stdout.decode()

assert f"{bucket_name}/bkprefix/{file_path.name}" in output

file_size = file_path.stat().st_size
assert str(file_size) in output

process = exec_test_command(BASE_CMD + ["ls"])
output = process.stdout.decode()
assert bucket_name in output
assert file_path.name not in output

process = exec_test_command(BASE_CMD + ["ls", bucket_name])
output = process.stdout.decode()
assert bucket_name not in output
assert "bkprefix" in output

downloaded_file_path = file_path.parent / f"downloaded_{file_path.name}"
process = exec_test_command(
BASE_CMD
+ [
"get",
bucket_name,
"bkprefix/" + file_path.name,
str(downloaded_file_path),
]
)
output = process.stdout.decode()
with open(downloaded_file_path) as f2, open(file_path) as f1:
assert f1.read() == f2.read()


def test_multi_files_multi_bucket(
create_bucket: Callable[[Optional[str]], str],
generate_test_files: GetTestFilesType,
Expand Down

0 comments on commit d4224ff

Please sign in to comment.