Class TestBinlog2sqlUtil

tests/test_binlog2sql_util.py:12–94

class TestBinlog2sqlUtil(unittest.TestCase):

    def test_is_valid_datetime(self):
        self.assertTrue(is_valid_datetime('2015-12-12 12:12:12'))
        self.assertFalse(is_valid_datetime('2015-12-12 12:12'))
        self.assertFalse(is_valid_datetime('2015-12-12'))
        self.assertFalse(is_valid_datetime(None))

    @mock.patch('binlog2sql.binlog2sql_util.os.path')
    def test_create_unique_file(self, mock_path):
        filename = "test.sql"
        mock_path.exists.return_value = False
        self.assertEqual(create_unique_file(filename), filename)
        mock_path.exists.return_value = True
        try:
            create_unique_file(filename)
        except Exception as e:
            self.assertEqual(str(e), "cannot create unique file %s.[0-1000]" % filename)

    def test_command_line_args(self):
        try:
            command_line_args(['--flashback', '--no-primary-key'])
        except Exception as e:
            self.assertEqual(str(e), "Lack of parameter: start_file")
        try:
            command_line_args(['--start-file', 'mysql-bin.000058', '--flashback', '--no-primary-key'])
        except Exception as e:
            self.assertEqual(str(e), "Only one of flashback or no_pk can be True")
        try:
            command_line_args(['--start-file', 'mysql-bin.000058', '--flashback', '--stop-never'])
        except Exception as e:
            self.assertEqual(str(e), "Only one of flashback or stop-never can be True")
        try:
            command_line_args(['--start-file', 'mysql-bin.000058', '--start-datetime', '2016-12-12'])
        except Exception as e:
            self.assertEqual(str(e), "Incorrect datetime argument")

    def test_compare_items(self):
        self.assertEqual(compare_items(('data', '12345')), '`data`=%s')
        self.assertEqual(compare_items(('data', None)), '`data` IS %s')

    def test_fix_object(self):
        self.assertEqual(fix_object('ascii'), 'ascii')
        self.assertEqual(fix_object(u'unicode'), u'unicode'.encode('utf-8'))

    def test_generate_sql_pattern(self):
        row = {'values': {'data': 'hello', 'id': 1}}
        mock_write_event = mock.create_autospec(WriteRowsEvent)
        mock_write_event.schema = 'test'
        mock_write_event.table = 'tbl'
        mock_write_event.primary_key = 'id'
        pattern = generate_sql_pattern(binlog_event=mock_write_event, row=row, flashback=False, no_pk=False)
        self.assertEqual(pattern, {'values': ['hello', 1],
                                   'template': 'INSERT INTO `test`.`tbl`(`data`, `id`) VALUES (%s, %s);'})
        pattern = generate_sql_pattern(binlog_event=mock_write_event, row=row, flashback=True, no_pk=False)
        self.assertEqual(pattern, {'values': ['hello', 1],
                                   'template': 'DELETE FROM `test`.`tbl` WHERE `data`=%s AND `id`=%s LIMIT 1;'})
        pattern = generate_sql_pattern(binlog_event=mock_write_event, row=row, flashback=False, no_pk=True)