Materialize/misc/dbt-materialize/tests/adapter/test_source_freshness.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 pytest
from dbt.cli.main import dbtRunner

freshness_via_metadata_schema_yml = """version: 2
sources:
  - name: test_source
    loader: custom
    freshness:
      warn_after: {count: 10, period: hour}
      error_after: {count: 1, period: day}
    schema: my_schema
    quoting:
      identifier: True
    tables:
      - name: test_table
        identifier: source
"""


class TestMetadataFreshnessFails:
    @pytest.fixture(scope="class")
    def models(self):
        return {"schema.yml": freshness_via_metadata_schema_yml}

    # We mark TableLastModifiedMetadata as not implemented, so trying to use metadata-based source freshness checks should result in a parse-time warning.
    def test_metadata_freshness_fails(self, project):
        got_warning = False

        def warning_probe(e):
            nonlocal got_warning
            if e.info.name == "FreshnessConfigProblem" and e.info.level == "warn":
                got_warning = True

        runner = dbtRunner(callbacks=[warning_probe])
        runner.invoke(["parse"])

        assert got_warning