end_to_end_test.py 文件源码

def expected_complex_data(self, actual_complex_data, complex_table_schema):
        expected_complex_data_dict = {'id': 1}
        for indx, complex_column_schema in enumerate(complex_table_schema):
            column_name = self._build_sql_column_name(indx)
            if isinstance(complex_column_schema.sqla_obj, mysql.SET):
                expected_complex_data_dict[column_name] = \
                    sorted(actual_complex_data[column_name])
            elif isinstance(complex_column_schema.sqla_obj, mysql.DATETIME):
                date_time_obj = \
                    complex_column_schema.data.isoformat()
                expected_complex_data_dict[column_name] = date_time_obj
            elif isinstance(complex_column_schema.sqla_obj, mysql.TIMESTAMP):
                date_time_obj = \
                    complex_column_schema.data.replace(tzinfo=pytz.utc)
                expected_complex_data_dict[column_name] = date_time_obj
            elif isinstance(complex_column_schema.sqla_obj, mysql.TIME):
                number_of_micros = transform_timedelta_to_number_of_microseconds(
                    complex_column_schema.data
                )
                expected_complex_data_dict[column_name] = number_of_micros
            else:
                expected_complex_data_dict[column_name] = \
                    complex_column_schema.data
        return expected_complex_data_dict