easy_sql.sql_tester

Module Contents

Classes

SqlReader

SqlTester

TableColumnTypes

TableData

TestCase

WorkPath

Attributes

work_path

easy_sql.sql_tester.work_path
class easy_sql.sql_tester.SqlReader
find_file_path(self, file_name)
Parameters

file_name (str) –

Return type

str

read_as_content(self, test_data_file)
Parameters

test_data_file (str) –

read_as_file_path(self, test_data_file)
Parameters

test_data_file (str) –

abstract read_sql(self, test_data_file)
Parameters

test_data_file (str) –

Return type

str

class easy_sql.sql_tester.SqlTester(backend_creator=None, table_column_types=None, sql_reader_creator=None, sql_processor_creator=None, unit_test_case=None, dry_run=True, env='test', work_dir=None, backend='spark', scala_udf_initializer=None)
Parameters
convert_cases_to_json(self, test_data_file)
Parameters

test_data_file (str) –

generate_python_unittest_file(self, test_data_file, backend='spark')
Parameters
  • test_data_file (str) –

  • backend (str) –

parse_test_cases(self, test_data_file, table_column_types)
Parameters

table_column_types (TableColumnTypes) –

Return type

List[TestCase]

run_case(self, case)
Parameters

case (TestCase) –

Return type

bool

run_test(self, test_data_file, case_idx=- 1)
Parameters
  • test_data_file (str) –

  • case_idx (int) –

Return type

TestResult

run_test_from_cases(self, etl_file, test_cases, case_idx=- 1)
Parameters
  • etl_file (str) –

  • test_cases (List[TestCase]) –

  • case_idx (int) –

run_tests(self, test_data_files)
Parameters

test_data_files (List[str]) –

class easy_sql.sql_tester.TableColumnTypes(predefined_table_col_types, partition_col_types, backend)
Parameters
  • predefined_table_col_types (Dict[str, Dict[str, str]]) –

  • partition_col_types (Dict[str, str]) –

  • backend (str) –

cast_as_type(self, table_name, col_name, col_value, date_converter=None, col_type=None)
Parameters
  • table_name (str) –

  • col_name (str) –

  • col_value (Any) –

  • date_converter (Optional[Callable]) –

  • col_type (Optional[str]) –

Return type

Tuple[str, Any]

column_types_to_schema(self, backend, columns, types)
Parameters
Return type

Union[pyspark.sql.types.StructType, List[easy_sql.sql_processor.backend.rdb.Col]]

column_types_to_schema_rdb(self, backend, columns, types)
Parameters
Return type

List[easy_sql.sql_processor.backend.rdb.Col]

column_types_to_schema_spark(self, spark, columns, types)
Parameters
  • spark (pyspark.sql.SparkSession) –

  • columns (List[str]) –

  • types (List[str]) –

Return type

pyspark.sql.types.StructType

get_col_type(self, table_name, col_name)
Parameters
  • table_name (str) –

  • col_name (str) –

Return type

str

class easy_sql.sql_tester.TableData(name, columns, column_types, values, value_descriptions)
Parameters
  • name (str) –

  • columns (List[str]) –

  • column_types (List[str]) –

  • values (List[List[Any]]) –

  • value_descriptions (List[str]) –

as_dict(self)
Return type

Dict[str, Any]

static from_dict(data)
Parameters

data (Dict) –

Return type

TableData

pt_col(self, possible_pt_cols)
Parameters

possible_pt_cols (List[str]) –

Return type

Optional[str]

class easy_sql.sql_tester.TestCase(sql_file_path=None, sql_file_content=None, default_col_type='string', *, name=None, vars=None, includes=None)
Parameters
  • sql_file_path (Optional[str]) –

  • sql_file_content (Optional[str]) –

  • name (Optional[str]) –

  • vars (Optional[Dict[str, str]]) –

  • includes (Optional[Dict[str, str]]) –

as_dict(self)
property completed(self)
static from_dict(data)
Parameters

data (Dict) –

Return type

TestCase

is_type_in_column_names(self, cells)
Return type

bool

property missed_fields(self)
parse_cell_value_as_date(self, wb, value)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • value (Any) –

Return type

Optional[datetime.datetime]

parse_funcs(self, row_start_idx, rows)
Parameters
  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

parse_includes(self, wb, row_start_idx, rows)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

parse_input(self, wb, row_start_idx, rows, table_column_types)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

  • table_column_types (TableColumnTypes) –

parse_output(self, wb, row_start_idx, rows, table_column_types)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

  • table_column_types (TableColumnTypes) –

parse_table(self, wb, label, row_start_idx, rows, table_column_types)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • label (str) –

  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

  • table_column_types (TableColumnTypes) –

Return type

TableData

parse_test_case_of_label(self, wb, label, row_start_idx, rows, table_column_types)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • label (str) –

  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

  • table_column_types (TableColumnTypes) –

parse_udfs(self, row_start_idx, rows)
Parameters
  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

parse_var_from_cell(self, wb, var_name, var_value)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • var_name (openpyxl.cell.Cell) –

  • var_value (openpyxl.cell.Cell) –

Return type

Tuple[str, Any]

parse_vars(self, wb, row_start_idx, rows)
Parameters
  • wb (openpyxl.workbook.Workbook) –

  • row_start_idx (int) –

  • rows (List[List[openpyxl.cell.Cell]]) –

read_sql_content(self)
Return type

str

property simple_sql_name(self)
class easy_sql.sql_tester.WorkPath(work_path=None)
Parameters

work_path (Optional[str]) –

path(self, relative_path)
Parameters

relative_path (str) –

relative_path(self, abs_path)
Parameters

abs_path (str) –

set_work_path(self, work_path)
Parameters

work_path (str) –

work_path(self)