Materialize/misc/dbt-materialize/tests/adapter/test_run_hooks.py

# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License in the LICENSE file at the
# root of this repository, or online at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os

import pytest
from dbt.tests.adapter.hooks.test_run_hooks import (
    BaseAfterRunHooks,
    BasePrePostRunHooks,
)
from dbt.tests.util import run_dbt
from dbt_common.exceptions import DbtDatabaseError
from fixtures import run_hook, test_run_operation


class TestPrePostRunHooksMaterialize(BasePrePostRunHooks):
    @pytest.fixture(scope="function")
    def setUp(self, project):
        project.run_sql(f"drop table if exists { project.test_schema }.on_run_hook")
        project.run_sql(run_hook.format(schema=project.test_schema))
        project.run_sql(f"drop table if exists { project.test_schema }.schemas")
        project.run_sql(f"drop table if exists { project.test_schema }.db_schemas")
        os.environ["TERM_TEST"] = "TESTING"

    @pytest.fixture(scope="class")
    def project_config_update(self):
        return test_run_operation

    def check_hooks(self, state, project, host):
        ctx = self.get_ctx_vars(state, project)

        assert ctx["test_state"] == state
        assert ctx["target_dbname"] == "materialize"
        assert ctx["target_host"] == "materialized"
        assert ctx["target_name"] == "default"
        assert ctx["target_schema"] == project.test_schema
        assert ctx["target_threads"] == 1
        assert ctx["target_type"] == project.adapter_type

        assert (
            ctx["run_started_at"] is not None and len(ctx["run_started_at"]) > 0
        ), "run_started_at was not set"
        assert (
            ctx["invocation_id"] is not None and len(ctx["invocation_id"]) > 0
        ), "invocation_id was not set"


class TestAfterRunHooksMaterialize(BaseAfterRunHooks):
    def test_missing_column_pre_hook(self, project):
        with pytest.raises(DbtDatabaseError):
            run_dbt(["run"], expect_pass=False)

    pass