fix(s3): Update operator tests to work with full key paths

Author: tomasfarias

airflow_dbt_python/hooks/backends/s3.py

@@ -121,7 +121,7 @@ def push_many(
         all_files = Path(source).glob("**/*")
 
         if delete_before:
-            keys = self.hook.list_keys(bucket_name)
+            keys = self.hook.list_keys(bucket_name, prefix=key)
             self.hook.delete_objects(bucket_name, keys)
 
         if key.endswith(".zip"):
@@ -230,6 +230,10 @@ def load_file_handle_replace_error(
         success = True
 
         if bucket_name is None:
+            # We can't call S3Hook.load_file with bucket_name=None as it checks for the
+            # presence of the parameter to decide whether setting a bucket_name is
+            # required. By passing bucket_name=None, the parameter is set, and
+            # 'None' will be used as the bucket name.
             bucket_name, key = self.hook.parse_s3_url(key)
 
         self.log.info("Loading file %s to S3: %s", file_path, key)

tests/operators/test_dbt_clean.py

@@ -100,12 +100,12 @@ def test_dbt_clean_after_compile_in_s3(
     # Run compile first to ensure a target/ directory exists to be cleaned
     comp.execute({})
 
-    keys = s3_hook.list_keys(s3_bucket, f"s3://{s3_bucket}/project/target")
+    keys = s3_hook.list_keys(s3_bucket, prefix="project/target")
     assert len(keys) > 0
 
     clean_result = op.execute({})
 
     assert clean_result is None
 
-    keys = s3_hook.list_keys(s3_bucket, f"s3://{s3_bucket}/project/target")
+    keys = s3_hook.list_keys(s3_bucket, prefix="project/target")
     assert len(keys) == 0

tests/operators/test_dbt_docs_generate.py

@@ -93,19 +93,13 @@ def test_dbt_docs_generate_push_to_s3(
             )
 
     # Ensure we are working with an empty target in S3.
-    keys = s3_hook.list_keys(
-        s3_bucket,
-        f"s3://{s3_bucket}/project/target/",
-    )
+    keys = s3_hook.list_keys(s3_bucket, "project/target")
     if keys is not None and len(keys) > 0:
         s3_hook.delete_objects(
             s3_bucket,
             keys,
         )
-        keys = s3_hook.list_keys(
-            s3_bucket,
-            f"s3://{s3_bucket}/project/target/",
-        )
+        keys = s3_hook.list_keys(s3_bucket, "project/target")
     assert keys is None or len(keys) == 0
 
     op = DbtDocsGenerateOperator(
@@ -117,10 +111,7 @@ def test_dbt_docs_generate_push_to_s3(
     results = op.execute({})
     assert results is not None
 
-    keys = s3_hook.list_keys(
-        s3_bucket,
-        f"s3://{s3_bucket}/project/target/",
-    )
-    assert f"s3://{s3_bucket}/project/target/manifest.json" in keys
-    assert f"s3://{s3_bucket}/project/target/catalog.json" in keys
-    assert f"s3://{s3_bucket}/project/target/index.html" in keys
+    keys = s3_hook.list_keys(s3_bucket)
+    assert f"project/target/manifest.json" in keys
+    assert f"project/target/catalog.json" in keys
+    assert f"project/target/index.html" in keys