Remove explicit pass from several extract_on_complete methods. (#…

…1771) Adjust expected events in integration tests. Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com>
OpenLineage · Apr 15, 2023 · 0eef1fd · 0eef1fd
1 parent 5e34208
commit 0eef1fd
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 13 deletions.
diff --git a/integration/airflow/openlineage/airflow/extractors/gcs_extractor.py b/integration/airflow/openlineage/airflow/extractors/gcs_extractor.py
@@ -40,6 +40,3 @@ def extract(self) -> Optional[TaskMetadata]:
             inputs=input_objects,
             outputs=[output_object],
         )
-
-    def extract_on_complete(self, task_instance) -> Optional[TaskMetadata]:
-        pass
diff --git a/integration/airflow/openlineage/airflow/extractors/great_expectations_extractor.py b/integration/airflow/openlineage/airflow/extractors/great_expectations_extractor.py
@@ -31,9 +31,6 @@ def get_operator_classnames(cls) -> List[str]:
     def extract(self) -> Optional[TaskMetadata]:
         return None
 
-    def extract_on_complete(self, task_instance) -> Optional[TaskMetadata]:
-        return None
-
 
 if _has_great_expectations:
     GreatExpectationsExtractor = GreatExpectationsExtractorImpl

diff --git a/integration/airflow/openlineage/airflow/extractors/s3_extractor.py b/integration/airflow/openlineage/airflow/extractors/s3_extractor.py
@@ -41,9 +41,6 @@ def extract(self) -> Optional[TaskMetadata]:
             outputs=[output_object],
         )
 
-    def extract_on_complete(self, task_instance) -> Optional[TaskMetadata]:
-        pass
-
 
 class S3FileTransformExtractor(BaseExtractor):
     @classmethod

diff --git a/integration/airflow/tests/integration/requests/gcs.json b/integration/airflow/tests/integration/requests/gcs.json
@@ -21,12 +21,22 @@
     }
 },{
     "eventType" : "COMPLETE",
-    "inputs" : [],
+    "inputs" : [
+        {
+            "name" : "{{ env_var('GOOGLE_CLOUD_STORAGE_SOURCE_URI') }}",
+            "namespace" : "{{ env_var('GOOGLE_CLOUD_STORAGE_SOURCE_URI') | url_scheme_authority }}"
+        }
+    ],
     "job" : {
         "facets": {},
         "name" : "gcs_dag.gcs_task"
     },
-    "outputs" : [],
+    "outputs" : [
+        {
+            "name" : "{{ env_var('GOOGLE_CLOUD_STORAGE_DESTINATION_URI') }}",
+            "namespace" : "{{ env_var('GOOGLE_CLOUD_STORAGE_DESTINATION_URI') | url_scheme_authority }}"
+        }
+    ],
     "run" : {
         "facets" : {}
     }

diff --git a/integration/airflow/tests/integration/requests/s3copy.json b/integration/airflow/tests/integration/requests/s3copy.json
@@ -21,12 +21,22 @@
     }
 },{
     "eventType": "COMPLETE",
-    "inputs": [],
+    "inputs": [
+        {
+            "name": "testfile",
+            "namespace": "s3://testbucket"
+        }
+    ],
     "job": {
         "facets": {},
         "name": "s3copy_dag.s3copy_task"
     },
-    "outputs": [],
+    "outputs": [
+        {
+            "name": "testfile2",
+            "namespace": "s3://testbucket"
+        }
+    ],
     "run": {
         "facets": {}
     }