From 7ab383c3d380e4271a6fb0e45de1d02ab394d59c Mon Sep 17 00:00:00 2001 From: Haleemur Ali Date: Tue, 19 Jul 2022 22:25:39 -0400 Subject: [PATCH] add extraction test for blob data --- tests/integration/test_tap_mysql.py | 32 +++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/integration/test_tap_mysql.py b/tests/integration/test_tap_mysql.py index ce032df..d7a3cb1 100644 --- a/tests/integration/test_tap_mysql.py +++ b/tests/integration/test_tap_mysql.py @@ -1,3 +1,4 @@ +import codecs import os import unittest from unittest.mock import patch @@ -1483,3 +1484,34 @@ def tearDown(self) -> None: with connect_with_backoff(self.conn) as open_conn: with open_conn.cursor() as cursor: cursor.execute('DROP TABLE bit_booleans_table;') + +class TestBinaryMapping(unittest.TestCase): + + def setUp(self): + self.conn = test_utils.get_test_connection() + self.blob_data = b'hello world' + with connect_with_backoff(self.conn) as conn: + with conn.cursor() as cursor: + cursor.execute("CREATE TABLE binary_table(id int, c_blob blob)") + cursor.execute("INSERT INTO binary_table VALUES (%s, %s)", (1, self.blob_data)) + self.catalog = test_utils.discover_catalog(self.conn, {}) + + def tearDown(self): + with connect_with_backoff(self.conn) as open_conn: + with open_conn.cursor() as cursor: + cursor.execute('DROP TABLE binary_table') + + def test_sync_messages_are_correct(self): + self.catalog.streams[0] = test_utils.set_replication_method_and_key(self.catalog.streams[0], 'FULL_TABLE', None) + self.catalog.streams[0] = test_utils.set_selected(self.catalog.streams[0], True) + + global SINGER_MESSAGES + SINGER_MESSAGES.clear() + + tap_mysql.do_sync(self.conn, {}, self.catalog, {}) + + record_messages = list(filter(lambda m: isinstance(m, singer.RecordMessage), SINGER_MESSAGES)) + + self.assertEqual(len(record_messages), 1) + self.assertEqual([rec.record for rec in record_messages], + [{'id': 1, 'c_blob': codecs.encode(self.blob_data, 'hex').decode('utf-8').upper()}])