Tests for invalid and valid records from postgres-tap (#55)

* Add optional validate_records option * fixed pylint * Introduce InvalidTableStructureException * Tests for invalid and valid records from postgres-tap * trigger build * do not raise InvalidTableStructureException in flush_records * fixed test * trigger build * removed extra tab Co-authored-by: Peter Kosztolanyi <[email protected]>
transferwise · Jan 7, 2020 · 8d31e29 · 8d31e29
1 parent 5f6ba6a
commit 8d31e29
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 0 deletions.
diff --git a/tests/integration/resources/messages-pg-with-invalid-records.json b/tests/integration/resources/messages-pg-with-invalid-records.json
@@ -0,0 +1,4 @@
+{"type": "SCHEMA", "stream": "pg-invalid-records", "schema": {"definitions": {"sdc_recursive_boolean_array": {"items": {"$ref": "#/definitions/sdc_recursive_boolean_array"}, "type": ["null", "boolean", "array"]}, "sdc_recursive_integer_array": {"items": {"$ref": "#/definitions/sdc_recursive_integer_array"}, "type": ["null", "integer", "array"]}, "sdc_recursive_number_array": {"items": {"$ref": "#/definitions/sdc_recursive_number_array"}, "type": ["null", "number", "array"]}, "sdc_recursive_object_array": {"items": {"$ref": "#/definitions/sdc_recursive_object_array"}, "type": ["null", "object", "array"]}, "sdc_recursive_string_array": {"items": {"$ref": "#/definitions/sdc_recursive_string_array"}, "type": ["null", "string", "array"]}, "sdc_recursive_timestamp_array": {"format": "date-time", "items": {"$ref": "#/definitions/sdc_recursive_timestamp_array"}, "type": ["null", "string", "array"]}}, "properties": {"id": {"maximum": 9223372036854775807, "minimum": -9223372036854775808, "type": ["integer"]}, "results": {"type": ["null", "object"]}, "time_created": {"format": "date-time", "type": ["null", "string"]}}, "type": "object"}, "key_properties": ["id"], "bookmark_properties": ["id"]}
+{"type": "STATE", "value": {"bookmarks": {"pg-invalid-records": {"replication_key": "id", "replication_key_value": 52799009, "version": 1, "last_replication_method": "INCREMENTAL"}}, "currently_syncing": "pg-invalid-records"}}
+{"type": "ACTIVATE_VERSION", "stream": "pg-invalid-records", "version": 1}
+{"type": "RECORD", "stream": "pg-invalid-records", "record": {"id": 52799009, "results": "[{\"rule\": \"XYZ\", \"reason\": \"XYZ\", \"actions\": []}, {\"rule\": \"XYZ\", \"reason\": \"XYZ\", \"actions\": []}, {\"rule\": \"XYZ\", \"reason\": \"XYZ\", \"actions\": []}, {\"rule\": \"XYZ\", \"reason\": \"XYZ\", \"actions\": []}, {\"rule\": \"XYZ\", \"reason\": \"XYZ\", \"actions\": []}, {\"rule\": \"XYZ\", \"reason\": \"XYZ\", \"actions\": []}]", "time_created": "2019-12-17T11:40:15.296955+00:00"}, "version": 1, "time_extracted": "2019-12-17T12:04:46.015519Z"}
diff --git a/tests/integration/resources/messages-pg-with-valid-records.json b/tests/integration/resources/messages-pg-with-valid-records.json
@@ -0,0 +1,4 @@
+{"type": "SCHEMA", "stream": "pg-valid-records", "schema": {"definitions": {"sdc_recursive_boolean_array": {"items": {"$ref": "#/definitions/sdc_recursive_boolean_array"}, "type": ["null", "boolean", "array"]}, "sdc_recursive_integer_array": {"items": {"$ref": "#/definitions/sdc_recursive_integer_array"}, "type": ["null", "integer", "array"]}, "sdc_recursive_number_array": {"items": {"$ref": "#/definitions/sdc_recursive_number_array"}, "type": ["null", "number", "array"]}, "sdc_recursive_object_array": {"items": {"$ref": "#/definitions/sdc_recursive_object_array"}, "type": ["null", "object", "array"]}, "sdc_recursive_string_array": {"items": {"$ref": "#/definitions/sdc_recursive_string_array"}, "type": ["null", "string", "array"]}, "sdc_recursive_timestamp_array": {"format": "date-time", "items": {"$ref": "#/definitions/sdc_recursive_timestamp_array"}, "type": ["null", "string", "array"]}}, "properties": {"id": {"maximum": 9223372036854775807, "minimum": -9223372036854775808, "type": ["integer"]}, "results": {"type": ["null", "array", "object"]}, "time_created": {"format": "date-time", "type": ["null", "string"]}}, "type": "object"}, "key_properties": ["id"], "bookmark_properties": ["id"]}
+{"type": "STATE", "value": {"bookmarks": {"pg-valid-records": {"replication_key": "id", "replication_key_value": 52799009, "version": 1, "last_replication_method": "INCREMENTAL"}}, "currently_syncing": "pg-valid-records"}}
+{"type": "ACTIVATE_VERSION", "stream": "pg-valid-records", "version": 1}
+{"type": "RECORD", "stream": "pg-valid-records", "record": {"id": 52799009, "results": [{"rule": "XYZ", "reason": "XYZ", "actions": []}, {"rule": "XYZ", "reason": "XYZ", "actions": []}, {"rule": "XYZ", "reason": "XYZ", "actions": []}, {"rule": "XYZ", "reason": "XYZ", "actions": []}, {"rule": "XYZ", "reason": "XYZ", "actions": []}], "time_created": "2019-12-17T11:40:15.296955+00:00"}, "version": 1, "time_extracted": "2019-12-17T19:12:12.006049Z"}
diff --git a/tests/integration/test_target_snowflake.py b/tests/integration/test_target_snowflake.py
@@ -867,3 +867,22 @@ def test_record_validation(self):
         self.config['validate_records'] = False
         with assert_raises(ProgrammingError):
             self.persist_lines_with_cache(tap_lines)
+
+    def test_pg_records_validation(self):
+        """Test validating records from postgres tap"""
+        tap_lines_invalid_records = test_utils.get_test_tap_lines('messages-pg-with-invalid-records.json')
+
+        # Loading invalid records when record validation enabled should fail at ...
+        self.config['validate_records'] = True
+        with assert_raises(RecordValidationException):
+            self.persist_lines_with_cache(tap_lines_invalid_records)
+
+        # Loading invalid records when record validation disabled, should pass without any exceptions
+        self.config['validate_records'] = False
+        self.persist_lines_with_cache(tap_lines_invalid_records)
+
+        # Valid records should pass for both with and without validation
+        tap_lines_valid_records = test_utils.get_test_tap_lines('messages-pg-with-valid-records.json')
+
+        self.config['validate_records'] = True
+        self.persist_lines_with_cache(tap_lines_valid_records)