Fix prefix path parsing in OBJ plugin (#686)

Co-authored-by: Zhiwei Liang <[email protected]>
Co-authored-by: Zhiwei Liang <[email protected]>

Author: jorenkoyen

linodecli/plugins/obj/objects.py

@@ -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,

tests/integration/obj/test_obj_plugin.py

@@ -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,