Skip to content

DBT Macro: Test Python Test

DBT macro name: trase_duckdb.test_python_test

DBT details

Description

No description

Details

Data Tests

Macros

{% test python_test(model, code_macro, macro_args=none, where_expression=none) %}
    {{ config(fail_calc="fail_count", tags=["elementary-tests"]) }}

    {% if not execute or not elementary.is_test_command() or not elementary.is_elementary_enabled() %}
        {% do return(none) %}
    {% endif %}

    {% if model is string %}
        {{
            exceptions.raise_compiler_error(
                "Unsupported model: "
                ~ model
                ~ " (this might happen if you provide a 'where' parameter to the test or override 'ref' or 'source')"
            )
        }}
    {% endif %}

    {% if not code_macro %}
        {% do exceptions.raise_compiler_error(
            "A `code_macro` must be provided to a Python test."
        ) %}
    {% endif %}
    {% if not macro_args %} {% set macro_args = {} %} {% endif %}

    {% set test_args = kwargs %}
    {% set test_node = dict(context.model) %}
    {% set model_relation = model.quote(false, false, false) %}
    {% set elementary_database_name, elementary_schema_name = (
        elementary.get_package_database_and_schema()
    ) %}
    {% set output_table = api.Relation.create(
        database=elementary_database_name,
        schema=elementary_schema_name,
        identifier="pytest_tmp__" ~ test_node.alias,
    ).quote(false, false, false) %}

    {% do test_node.update(
        {
            "database": elementary_database_name,
            "schema": elementary_schema_name,
        }
    ) %}

    {% set config_copy = dict(test_node.get("config", {})) %}
    {% do config_copy.update(test_args) %}
    {% do test_node.update({"config": config_copy}) %}

    {% if code_macro is string %} {% set user_py_code_macro = context[code_macro] %}
    {% else %} {% set user_py_code_macro = code_macro %}
    {% endif %}

    {% if not user_py_code_macro %}
        {% do exceptions.raise_compiler_error(
            "Unable to find the macro `" ~ code_macro ~ "`."
        ) %}
    {% endif %}
    {% set user_py_code = user_py_code_macro(macro_args) %}
    {% set compiled_py_code = adapter.dispatch("compile_py_code", "elementary")(
        model_relation,
        user_py_code,
        output_table,
        where_expression,
        code_type="test",
    ) %}

    {% do elementary.run_python(test_node, compiled_py_code) %}
    select fail_count
    from {{ output_table }}
{% endtest %}