data_algebra/test_util.py
def check_transform(
    ops,
    data,
    expect,
    *,
    float_tol: float = 1e-8,
    check_column_order: bool = False,
    cols_case_sensitive: bool = False,
    check_row_order: bool = False,
    check_parse: bool = True,
    models_to_skip: Optional[Iterable[str]] = None,
    empty_produces_empty: bool = True,
) -> None:
    """
    Test an operator dag produces the expected result, and parses correctly.
    Assert if there are issues.

    :param ops: data_algebra.data_ops.ViewRepresentation
    :param data: pd.DataFrame or map of strings to pd.DataFrame
    :param expect: pd.DataFrame
    :param float_tol: passed to equivalent_frames()
    :param check_column_order: passed to equivalent_frames()
    :param cols_case_sensitive: passed to equivalent_frames()
    :param check_row_order: passed to equivalent_frames()
    :param check_parse: if True check expression parses/formats to self
    :param models_to_skip: None or set of model names to skip testing
    :param empty_produces_empty: logical, if true assume emtpy inputs should produce empty output
    :return: nothing
    """

    # convert single table to dictionary
    if not isinstance(data, dict):
        cols_used = ops.columns_used()
        table_name = [k for k in cols_used.keys()][0]
        data = {table_name: data}

    db_handles = [
        # non-connected handles, lets us test some of the SQL generation path
        data_algebra.SQLite.SQLiteModel().db_handle(None),
        data_algebra.BigQuery.BigQueryModel().db_handle(None),
        data_algebra.PostgreSQL.PostgreSQLModel().db_handle(None),
        data_algebra.SparkSQL.SparkSQLModel().db_handle(None),
        data_algebra.MySQL.MySQLModel().db_handle(None),
    ]

    test_dbs = get_test_dbs()
    db_handles = db_handles + test_dbs

    if models_to_skip is not None:
        models_to_skip = set(models_to_skip)
        db_handles = [h for h in db_handles if str(h.db_model) not in models_to_skip]

    caught = None
    try:
        check_transform_on_handles(
            ops=ops,
            data=data,
            expect=expect,
            float_tol=float_tol,
            check_column_order=check_column_order,
            cols_case_sensitive=cols_case_sensitive,
            check_row_order=check_row_order,
            check_parse=check_parse,
            db_handles=db_handles,
            empty_produces_empty=empty_produces_empty,
        )
    except Exception as exc:
        caught = exc

    for handle in db_handles:
        # noinspection PyBroadException
        try:
            handle.close()
        except Exception:
            pass

    if caught is not None:
        raise caught

    Test an operator dag produces the expected result, and parses correctly.
    Assert if there are issues.

    :param ops: data_algebra.data_ops.ViewRepresentation
    :param data: pd.DataFrame or map of strings to pd.DataFrame
    :param expect: pd.DataFrame
    :param float_tol: passed to equivalent_frames()
    :param check_column_order: passed to equivalent_frames()
    :param cols_case_sensitive: passed to equivalent_frames()
    :param check_row_order: passed to equivalent_frames()
    :param check_parse: if True check expression parses/formats to self
    :param models_to_skip: None or set of model names to skip testing
    :param empty_produces_empty: logical, if true assume emtpy inputs should produce empty output
    :return: nothing
    
Test an operator dag produces the expected result, and parses correctly.
Assert if there are issues.
Test an operator dag produces the expected result and parses correctly . Assert if there are issues .
def check_transform(
    ops,
    data,
    expect,
    *,
    float_tol: float = 1e-8,
    check_column_order: bool = False,
    cols_case_sensitive: bool = False,
    check_row_order: bool = False,
    check_parse: bool = True,
    models_to_skip: Optional[Iterable[str]] = None,
    empty_produces_empty: bool = True,
) -> None:
    if not isinstance(data, dict):
        cols_used = ops.columns_used()
        table_name = [k for k in cols_used.keys()][0]
        data = {table_name: data}
    db_handles = [
        data_algebra.SQLite.SQLiteModel().db_handle(None),
        data_algebra.BigQuery.BigQueryModel().db_handle(None),
        data_algebra.PostgreSQL.PostgreSQLModel().db_handle(None),
        data_algebra.SparkSQL.SparkSQLModel().db_handle(None),
        data_algebra.MySQL.MySQLModel().db_handle(None),
    ]
    test_dbs = get_test_dbs()
    db_handles = db_handles + test_dbs
    if models_to_skip is not None:
        models_to_skip = set(models_to_skip)
        db_handles = [h for h in db_handles if str(h.db_model) not in models_to_skip]
    caught = None
    try:
        check_transform_on_handles(
            ops=ops,
            data=data,
            expect=expect,
            float_tol=float_tol,
            check_column_order=check_column_order,
            cols_case_sensitive=cols_case_sensitive,
            check_row_order=check_row_order,
            check_parse=check_parse,
            db_handles=db_handles,
            empty_produces_empty=empty_produces_empty,
        )
    except Exception as exc:
        caught = exc
    for handle in db_handles:
        try:
            handle.close()
        except Exception:
            pass
    if caught is not None:
        raise caught
def check_transform ( ops , data , expect , * , float_tol : float = 1e-8 , check_column_order : bool = False , cols_case_sensitive : bool = False , check_row_order : bool = False , check_parse : bool = True , models_to_skip : Optional [ Iterable [ str ] ] = None , empty_produces_empty : bool = True , ) -> None : if not isinstance ( data , dict ) : cols_used = ops . columns_used ( ) table_name = [ k for k in cols_used . keys ( ) ] [ 0 ] data = { table_name : data } db_handles = [ data_algebra . SQLite . SQLiteModel ( ) . db_handle ( None ) , data_algebra . BigQuery . BigQueryModel ( ) . db_handle ( None ) , data_algebra . PostgreSQL . PostgreSQLModel ( ) . db_handle ( None ) , data_algebra . SparkSQL . SparkSQLModel ( ) . db_handle ( None ) , data_algebra . MySQL . MySQLModel ( ) . db_handle ( None ) , ] test_dbs = get_test_dbs ( ) db_handles = db_handles + test_dbs if models_to_skip is not None : models_to_skip = set ( models_to_skip ) db_handles = [ h for h in db_handles if str ( h . db_model ) not in models_to_skip ] caught = None try : check_transform_on_handles ( ops = ops , data = data , expect = expect , float_tol = float_tol , check_column_order = check_column_order , cols_case_sensitive = cols_case_sensitive , check_row_order = check_row_order , check_parse = check_parse , db_handles = db_handles , empty_produces_empty = empty_produces_empty , ) except Exception as exc : caught = exc for handle in db_handles : try : handle . close ( ) except Exception : pass if caught is not None : raise caught
Test an operator dag produces the expected result, and parses correctly.
Test an operator dag produces the expected result and parses correctly .
"""
    Test an operator dag produces the expected result, and parses correctly.
    Assert if there are issues.

    :param ops: data_algebra.data_ops.ViewRepresentation
    :param data: pd.DataFrame or map of strings to pd.DataFrame
    :param expect: pd.DataFrame
    :param float_tol: passed to equivalent_frames()
    :param check_column_order: passed to equivalent_frames()
    :param cols_case_sensitive: passed to equivalent_frames()
    :param check_row_order: passed to equivalent_frames()
    :param check_parse: if True check expression parses/formats to self
    :param models_to_skip: None or set of model names to skip testing
    :param empty_produces_empty: logical, if true assume emtpy inputs should produce empty output
    :return: nothing
    """ # convert single table to dictionary # non-connected handles, lets us test some of the SQL generation path # noinspection PyBroadException
None
ops
None
data
pd.DataFrame or map of strings to pd.DataFrame
pd . DataFrame or map of strings to pd . DataFrame
expect
None
float_tol
float
passed to equivalent_frames()
passed to equivalent_frames ()
check_column_order
bool
passed to equivalent_frames()
passed to equivalent_frames ()
cols_case_sensitive
bool
passed to equivalent_frames()
passed to equivalent_frames ()
check_row_order
bool
passed to equivalent_frames()
passed to equivalent_frames ()
check_parse
bool
if True check expression parses/formats to self
if True check expression parses / formats to self
models_to_skip
Optional[Iterable[str]]
None or set of model names to skip testing
None or set of model names to skip testing
empty_produces_empty
bool
logical, if true assume emtpy inputs should produce empty output
logical if true assume emtpy inputs should produce empty output