got a test working

Co-authored-by: Jeremy Cohen <jeremy@dbtlabs.com>

core/dbt/contracts/graph/model_config.py

@@ -335,6 +335,39 @@ class BaseConfig(AdditionalPropertiesAllowed, Replaceable):
 @dataclass
 class SourceConfig(BaseConfig):
     enabled: bool = True
+    quoting: Dict[str, Any] = field(
+        default_factory=dict,
+        metadata=MergeBehavior.Update.meta(),
+    )
+    freshness: Optional[Dict[str, Any]] = field(
+        default=None,
+        metadata=CompareBehavior.Exclude.meta(),
+    )
+    loader: Optional[str] = field(
+        default=None,
+        metadata=CompareBehavior.Exclude.meta(),
+    )
+    # TODO what type is this? docs say: "<column_name_or_expression>"
+    loaded_at_field: Optional[str] = field(
+        default=None,
+        metadata=CompareBehavior.Exclude.meta(),
+    )
+    database: Optional[str] = field(
+        default=None,
+        metadata=CompareBehavior.Exclude.meta(),
+    )
+    schema: Optional[str] = field(
+        default=None,
+        metadata=CompareBehavior.Exclude.meta(),
+    )
+    meta: Dict[str, Any] = field(
+        default_factory=dict,
+        metadata=MergeBehavior.Update.meta(),
+    )
+    tags: Union[List[str], str] = field(
+        default_factory=list_str,
+        metadata=metas(ShowBehavior.Hide, MergeBehavior.Append, CompareBehavior.Exclude),
+    )
 
 
 @dataclass

core/dbt/contracts/graph/unparsed.py

@@ -242,6 +242,7 @@ class Quoting(dbtClassMixin, Mergeable):
 
 @dataclass
 class UnparsedSourceTableDefinition(HasColumnTests, HasTests):
+    config: Dict[str, Any] = field(default_factory=dict)
     loaded_at_field: Optional[str] = None
     identifier: Optional[str] = None
     quoting: Quoting = field(default_factory=Quoting)
@@ -322,6 +323,7 @@ class SourcePatch(dbtClassMixin, Replaceable):
     path: Path = field(
         metadata=dict(description="The path to the patch-defining yml file"),
     )
+    config: Dict[str, Any] = field(default_factory=dict)
     description: Optional[str] = None
     meta: Optional[Dict[str, Any]] = None
     database: Optional[str] = None

core/dbt/parser/manifest.py

@@ -465,6 +465,8 @@ class ManifestLoader:
                     else:
                         dct = block.file.dict_from_yaml
                     parser.parse_file(block, dct=dct)
+                    # Came out of here with UnpatchedSourceDefinition containing configs at the source level
+                    # and not configs at the table level (as expected)
                 else:
                     parser.parse_file(block)
                 project_parsed_path_count += 1

core/dbt/parser/sources.py

@@ -1,6 +1,6 @@
 import itertools
 from pathlib import Path
-from typing import Iterable, Dict, Optional, Set, List, Any
+from typing import Iterable, Dict, Optional, Set, Any
 from dbt.adapters.factory import get_adapter
 from dbt.config import RuntimeConfig
 from dbt.context.context_config import (
@@ -137,15 +137,13 @@ class SourcePatcher:
         tags = sorted(set(itertools.chain(source.tags, table.tags)))
 
         config = self._generate_source_config(
-            fqn=target.fqn,
+            target=target,
             rendered=True,
-            project_name=target.package_name,
         )
 
         unrendered_config = self._generate_source_config(
-            fqn=target.fqn,
+            target=target,
             rendered=False,
-            project_name=target.package_name,
         )
 
         if not isinstance(config, SourceConfig):
@@ -261,7 +259,7 @@ class SourcePatcher:
         )
         return node
 
-    def _generate_source_config(self, fqn: List[str], rendered: bool, project_name: str):
+    def _generate_source_config(self, target: UnpatchedSourceDefinition, rendered: bool):
         generator: BaseContextConfigGenerator
         if rendered:
             generator = ContextConfigGenerator(self.root_project)
@@ -270,10 +268,11 @@ class SourcePatcher:
 
         return generator.calculate_node_config(
             config_call_dict={},
-            fqn=fqn,
+            fqn=target.fqn,
             resource_type=NodeType.Source,
-            project_name=project_name,
+            project_name=target.package_name,
             base=False,
+            patch_config_dict=target.source.config,
         )
 
     def _get_relation_name(self, node: ParsedSourceDefinition):

tests/functional/permission/test_permissions.py

@@ -64,7 +64,6 @@ class TestPermissions:
         project,
     ):
         # now it should work!
-        # breakpoint()
         project.run_sql("grant create on database {} to noaccess".format(project.database))
         project.run_sql(
             'grant usage, create on schema "{}" to noaccess'.format(project.test_schema)

tests/functional/source_overrides/test_simple_source_override.py

@@ -1,7 +1,8 @@
 from datetime import datetime, timedelta
 import pytest
+from dbt.contracts.graph.model_config import SourceConfig
 
-from dbt.tests.util import run_dbt, update_config_file
+from dbt.tests.util import run_dbt, update_config_file, get_manifest
 from dbt.tests.tables import TableComparison
 from dbt.tests.fixtures.project import write_project_files
 from tests.functional.source_overrides.fixtures import (  # noqa: F401
@@ -17,6 +18,7 @@ class TestSourceOverride:
     @pytest.fixture(scope="class", autouse=True)
     def setUp(self, project_root, local_dependency):  # noqa: F811
         write_project_files(project_root, "local_dependency", local_dependency)
+        pytest._id = 101
 
     @pytest.fixture(scope="class")
     def models(self):
@@ -40,6 +42,36 @@ class TestSourceOverride:
             ]
         }
 
+    def _set_updated_at_to(self, delta, project):
+        insert_time = datetime.utcnow() + delta
+        timestr = insert_time.strftime("%Y-%m-%d %H:%M:%S")
+        # favorite_color,id,first_name,email,ip_address,updated_at
+
+        quoted_columns = ",".join(
+            project.adapter.quote(c)
+            for c in ("favorite_color", "id", "first_name", "email", "ip_address", "updated_at")
+        )
+        insert_id = pytest._id
+        pytest._id += 1
+
+        kwargs = {
+            "schema": project.test_schema,
+            "time": timestr,
+            "id": insert_id,
+            "source": project.adapter.quote("snapshot_freshness_base"),
+            "quoted_columns": quoted_columns,
+        }
+
+        raw_sql = """INSERT INTO {schema}.{source}
+            ({quoted_columns})
+        VALUES (
+            'blue',{id},'Jake','abc@example.com','192.168.1.1','{time}'
+        )""".format(
+            **kwargs
+        )
+
+        project.run_sql(raw_sql)
+
     @pytest.fixture(scope="class")
     def project_config_update(self):
         return {
@@ -61,39 +93,7 @@ class TestSourceOverride:
             },
         }
 
-    def _set_updated_at_to(self, insert_id, delta, project):
-        insert_time = datetime.utcnow() + delta
-        timestr = insert_time.strftime("%Y-%m-%d %H:%M:%S")
-        # favorite_color,id,first_name,email,ip_address,updated_at
-
-        quoted_columns = ",".join(
-            project.adapter.quote(c)
-            for c in ("favorite_color", "id", "first_name", "email", "ip_address", "updated_at")
-        )
-
-        kwargs = {
-            "schema": project.test_schema,
-            "time": timestr,
-            "id": insert_id,
-            "source": project.adapter.quote("snapshot_freshness_base"),
-            "quoted_columns": quoted_columns,
-        }
-
-        raw_sql = """INSERT INTO {schema}.{source}
-            ({quoted_columns})
-        VALUES (
-            'blue',{id},'Jake','abc@example.com','192.168.1.1','{time}'
-        )""".format(
-            **kwargs
-        )
-
-        project.run_sql(raw_sql)
-
-        return insert_id + 1
-
     def test_source_overrides(self, project):
-        insert_id = 101
-
         run_dbt(["deps"])
 
         seed_results = run_dbt(["seed"])
@@ -112,7 +112,7 @@ class TestSourceOverride:
         table_comp.assert_tables_equal("expected_result", "my_model")
 
         # set the updated_at field of this seed to last week
-        insert_id = self._set_updated_at_to(insert_id, timedelta(days=-7), project)
+        self._set_updated_at_to(timedelta(days=-7), project)
         # if snapshot-freshness fails, freshness just didn't happen!
         results = run_dbt(["source", "snapshot-freshness"], expect_pass=False)
         # we disabled my_other_source, so we only run the one freshness check
@@ -120,14 +120,14 @@ class TestSourceOverride:
         assert len(results) == 1
         # If snapshot-freshness passes, that means error_after was
         # applied from the source override but not the source table override
-        insert_id = self._set_updated_at_to(insert_id, timedelta(days=-2), project)
+        self._set_updated_at_to(timedelta(days=-2), project)
         results = run_dbt(
             ["source", "snapshot-freshness"],
             expect_pass=False,
         )
         assert len(results) == 1
 
-        insert_id = self._set_updated_at_to(insert_id, timedelta(hours=-12), project)
+        self._set_updated_at_to(timedelta(hours=-12), project)
         results = run_dbt(["source", "snapshot-freshness"], expect_pass=True)
         assert len(results) == 1
 
@@ -147,3 +147,186 @@ class TestSourceOverride:
         # not-fresh source
         results = run_dbt(["source", "snapshot-freshness"], expect_pass=False)
         assert len(results) == 2
+
+
+class SourceOverrideTest:
+    @pytest.fixture(scope="class", autouse=True)
+    def setUp(self, project_root, local_dependency):  # noqa: F811
+        write_project_files(project_root, "local_dependency", local_dependency)
+
+        pytest.expected_config = SourceConfig(
+            enabled=True,
+            quoting={"database": True, "schema": True, "identifier": True, "column": True},
+            freshness={
+                "warn_after": {"count": 1, "period": "minute"},
+                "error_after": {"count": 5, "period": "minute"},
+            },
+            loader="override_a_loader",
+            loaded_at_field="override_some_column",
+            database="override_custom_database",
+            schema="override_custom_schema",
+            meta={"languages": ["java"]},
+            tags=["override_important_tag"],
+        )
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "sources": {
+                "localdep": {
+                    "my_source": {
+                        "quoting": {
+                            "database": False,
+                            "schema": False,
+                            "identifier": False,
+                            "column": False,
+                        },
+                        "freshness": {
+                            "error_after": {"count": 24, "period": "hour"},
+                            "warn_after": {"count": 12, "period": "hour"},
+                        },
+                        "loader": "a_loader",
+                        "loaded_at_field": "some_column",
+                        "database": "custom_database",
+                        "schema": "custom_schema",
+                        "meta": {"languages": ["python"]},
+                        "tags": ["important_tag"],
+                    }
+                }
+            },
+        }
+
+    @pytest.fixture(scope="class")
+    def packages(self):
+        return {
+            "packages": [
+                {
+                    "local": "local_dependency",
+                },
+            ]
+        }
+
+
+overrides_source_level__schema_yml = """
+version: 2
+sources:
+  - name: my_source
+    overrides: localdep
+    config:
+        quoting:
+            database: True
+            schema: True
+            identifier: True
+            column: True
+        freshness:
+            error_after: {count: 1, period: minute}
+            warn_after: {count: 5, period: minute}
+        loader: "override_a_loader"
+        loaded_at_field: override_some_column
+        database: override_custom_database
+        schema: override_custom_schema
+        meta: {'languages': ['java']}
+        tags: ["override_important_tag"]
+    tables:
+      - name: my_table
+      - name: my_other_table
+      - name: snapshot_freshness
+"""
+
+
+# test overriding at the source level
+# expect fail since these are no valid configs right now
+class TestSourceLevelOverride(SourceOverrideTest):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": overrides_source_level__schema_yml}
+
+    # @pytest.mark.xfail
+    def test_source_level_overrides(self, project):
+        run_dbt(["deps"])
+
+        # this currently fails because configs fail parsing under an override
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+
+        assert "source.localdep.my_source.my_table" in manifest.sources
+        assert "source.localdep.my_source.my_other_table" in manifest.sources
+        assert "source.localdep.my_source.snapshot_freshness" in manifest.sources
+
+        config_my_table = manifest.sources.get("source.localdep.my_source.my_table").config
+        config_my_other_table = manifest.sources.get(
+            "source.localdep.my_source.my_other_table"
+        ).config
+        config_snapshot_freshness_table = manifest.sources.get(
+            "source.localdep.my_source.snapshot_freshness"
+        ).config
+
+        assert isinstance(config_my_table, SourceConfig)
+        assert isinstance(config_my_other_table, SourceConfig)
+
+        assert config_my_table == config_my_other_table
+        assert config_my_table == config_snapshot_freshness_table
+        assert config_my_table == pytest.expected_config
+
+
+overrides_source_level__schema_yml = """
+version: 2
+sources:
+  - name: my_source
+    overrides: localdep
+    tables:
+      - name: my_table
+      - name: my_other_table
+        config:
+            quoting:
+                database: True
+                schema: True
+                identifier: True
+                column: True
+            freshness:
+                error_after: {count: 1, period: minute}
+                warn_after: {count: 5, period: minute}
+            loader: "override_a_loader"
+            loaded_at_field: override_some_column
+            database: override_custom_database
+            schema: override_custom_schema
+            meta: {'languages': ['java']}
+            tags: ["override_important_tag"]
+      - name: snapshot_freshness
+"""
+
+
+# test overriding at the source table level
+# expect fail since these are no valid configs right now
+class TestSourceTableOverride(SourceOverrideTest):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": overrides_source_level__schema_yml}
+
+    # @pytest.mark.xfail
+    def test_source_table_overrides(self, project):
+        run_dbt(["deps"])
+
+        # this currently fails because configs fail parsing under an override
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+
+        assert "source.localdep.my_source.my_table" in manifest.sources
+        assert "source.localdep.my_source.my_other_table" in manifest.sources
+        assert "source.localdep.my_source.snapshot_freshness" in manifest.sources
+
+        config_my_table = manifest.sources.get("source.localdep.my_source.my_table").config
+        config_my_other_table = manifest.sources.get(
+            "source.localdep.my_source.my_other_table"
+        ).config
+        config_snapshot_freshness_table = manifest.sources.get(
+            "source.localdep.my_source.snapshot_freshness"
+        ).config
+
+        assert isinstance(config_my_table, SourceConfig)
+        assert isinstance(config_my_other_table, SourceConfig)
+
+        assert config_my_table != config_my_other_table
+        assert config_my_table == config_snapshot_freshness_table
+
+        assert config_my_other_table == pytest.expected_config

tests/functional/sources/test_source_configs.py

@@ -0,0 +1,690 @@
+import pytest
+from dbt.contracts.graph.model_config import SourceConfig
+from dbt.contracts.graph.unparsed import FreshnessThreshold, Time, TimePeriod, Quoting
+from dbt.exceptions import CompilationException
+
+from dbt.tests.util import run_dbt, update_config_file, get_manifest
+
+
+class SourceConfigTests:
+    @pytest.fixture(scope="class", autouse=True)
+    def setUp(self):
+        pytest.expected_config = SourceConfig(
+            enabled=True,
+            quoting={"database": False, "schema": False, "identifier": False, "column": False},
+            freshness={
+                "warn_after": {"count": 12, "period": "hour"},
+                "error_after": {"count": 24, "period": "hour"},
+            },
+            loader="a_loader",
+            loaded_at_field="some_column",
+            database="custom_database",
+            schema="custom_schema",
+            meta={"languages": ["python"]},
+            tags=["important_tag"],
+        )
+
+
+models__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    tables:
+      - name: test_table
+  - name: other_source
+    tables:
+      - name: test_table
+"""
+
+
+# Test enabled config in dbt_project.yml
+# expect pass, already implemented
+class TestSourceEnabledConfigProjectLevel(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": models__schema_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "sources": {
+                "test": {
+                    "test_source": {
+                        "enabled": True,
+                    },
+                }
+            }
+        }
+
+    def test_enabled_source_config_dbt_project(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+
+        new_enabled_config = {
+            "sources": {
+                "test": {
+                    "test_source": {
+                        "enabled": False,
+                    },
+                }
+            }
+        }
+        update_config_file(new_enabled_config, project.project_root, "dbt_project.yml")
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+
+        assert (
+            "source.test.test_source.test_table" not in manifest.sources
+        )  # or should it be there with enabled: false??
+        assert "source.test.other_source.test_table" in manifest.sources
+
+
+disabled_source_level__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    config:
+      enabled: False
+    tables:
+      - name: test_table
+      - name: disabled_test_table
+"""
+
+
+# Test enabled config at sources level in yml file
+# expect fail - not implemented
+class TestConfigYamlSourceLevel(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": disabled_source_level__schema_yml,
+        }
+
+    def test_source_config_yaml_source_level(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" not in manifest.sources
+        assert "source.test.test_source.disabled_test_table" not in manifest.sources
+
+
+disabled_source_table__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    tables:
+      - name: test_table
+      - name: disabled_test_table
+        config:
+            enabled: False
+"""
+
+
+# Test enabled config at source table level in yaml file
+# expect fail - not implemented
+class TestConfigYamlSourceTable(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": disabled_source_table__schema_yml,
+        }
+
+    @pytest.mark.xfail
+    def test_source_config_yaml_source_table(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.disabled_test_table" not in manifest.sources
+
+
+# Test all configs other than enabled in dbt_project.yml
+# expect fail - not implemented
+class TestAllConfigsProjectLevel(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": models__schema_yml}
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "sources": {
+                "enabled": True,
+                "quoting": {
+                    "database": False,
+                    "schema": False,
+                    "identifier": False,
+                    "column": False,
+                },
+                "freshness": {
+                    "error_after": {"count": 24, "period": "hour"},
+                    "warn_after": {"count": 12, "period": "hour"},
+                },
+                "loader": "a_loader",
+                "loaded_at_field": "some_column",
+                "database": "custom_database",
+                "schema": "custom_schema",
+                "meta": {"languages": ["python"]},
+                "tags": ["important_tag"],
+            }
+        }
+
+    @pytest.mark.xfail
+    def test_source_all_configs_dbt_project(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        config = manifest.sources.get("source.test.test_source.test_table").config
+
+        assert isinstance(config, SourceConfig)
+
+        assert config == pytest.expected_config
+
+
+configs_source_level__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    config:
+        enabled: True
+        quoting:
+            database: False
+            schema: False
+            identifier: False
+            column: False
+        freshness:
+            error_after: {count: 24, period: hour}
+            warn_after: {count: 12, period: hour}
+        loader: "a_loader"
+        loaded_at_field: some_column
+        database: custom_database
+        schema: custom_schema
+        meta: {'languages': ['python']}
+        tags: ["important_tag"]
+    tables:
+      - name: test_table
+      - name: other_test_table
+"""
+
+
+# Test configs other than enabled at sources level in yaml file
+# **currently passes since enabled is all that ends up in the
+# node.config since it's the only thing implemented
+class TestAllConfigsSourceLevel(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": configs_source_level__schema_yml}
+
+    def test_source_all_configs_source_level(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.other_test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+        config_other_test_table = manifest.sources.get(
+            "source.test.test_source.other_test_table"
+        ).config
+
+        assert isinstance(config_test_table, SourceConfig)
+        assert isinstance(config_other_test_table, SourceConfig)
+
+        assert config_test_table == config_other_test_table
+        assert config_test_table == pytest.expected_config
+
+
+configs_source_table__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    tables:
+      - name: test_table
+        config:
+            enabled: True
+            quoting:
+                database: False
+                schema: False
+                identifier: False
+                column: False
+            freshness:
+                error_after: {count: 24, period: hour}
+                warn_after: {count: 12, period: hour}
+            loader: "a_loader"
+            loaded_at_field: some_column
+            database: custom_database
+            schema: custom_schema
+            meta: {'languages': ['python']}
+            tags: ["important_tag"]
+      - name: other_test_table
+"""
+
+
+# Test configs other than enabled at source table level in yml file
+# expect fail - not implemented
+class TestSourceAllConfigsSourceTable(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": configs_source_table__schema_yml}
+
+    @pytest.mark.xfail
+    def test_source_all_configs_source_table(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.other_test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+        config_other_test_table = manifest.sources.get(
+            "source.test.test_source.other_test_table"
+        ).config
+
+        assert isinstance(config_test_table, SourceConfig)
+        assert isinstance(config_other_test_table, SourceConfig)
+
+        assert config_test_table != config_other_test_table
+        assert config_test_table == pytest.expected_config
+
+
+all_configs_everywhere__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    config:
+        tags: ["source_level_important_tag"]
+    tables:
+      - name: test_table
+        config:
+            enabled: True
+           quoting:
+                database: False
+                schema: False
+                identifier: False
+                column: False
+            freshness:
+                error_after: {count: 24, period: hour}
+                warn_after: {count: 12, period: hour}
+            loader: "a_loader"
+            loaded_at_field: some_column
+            database: custom_database
+            schema: custom_schema
+            meta: {'languages': ['python']}
+            tags: ["important_tag"]
+      - name: other_test_table
+"""
+
+
+# Test inheritence - set configs at project, source, and source-table level - expect source-table level to win
+# expect fail - not implemented
+class TestSourceConfigsInheritence1(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": all_configs_everywhere__schema_yml}
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"sources": {"tags": "project_level_important_tag"}}
+
+    @pytest.mark.xfail
+    def test_source_all_configs_source_table(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.other_test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+        config_other_test_table = manifest.sources.get(
+            "source.test.test_source.other_test_table"
+        ).config
+
+        assert isinstance(config_test_table, SourceConfig)
+        assert isinstance(config_other_test_table, SourceConfig)
+
+        assert config_test_table != config_other_test_table
+        assert config_test_table == pytest.expected_config
+
+        expected_source_level_config = SourceConfig(
+            enabled=True,
+            # "tags" = "source_level_important_tag"  #TODO: update after SourceConfigs gets updated
+        )
+
+        assert config_other_test_table == expected_source_level_config
+
+
+all_configs_not_table_schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    config:
+        enabled: True
+        quoting:
+            database: False
+            schema: False
+            identifier: False
+            column: False
+        freshness:
+            error_after: {count: 24, period: hour}
+            warn_after: {count: 12, period: hour}
+        loader: "a_loader"
+        loaded_at_field: some_column
+        database: custom_database
+        schema: custom_schema
+        meta: {'languages': ['python']}
+        tags: ["important_tag"]
+    tables:
+      - name: test_table
+      - name: other_test_table
+"""
+
+
+# Test inheritence - set configs at project and source level - expect source level to win
+# expect fail - not implemented
+class TestSourceConfigsInheritence2(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": all_configs_not_table_schema_yml}
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"sources": {"tags": "project_level_important_tag"}}
+
+    @pytest.mark.xfail
+    def test_source_two_configs_source_level(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.other_test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+        config_other_test_table = manifest.sources.get(
+            "source.test.test_source.other_test_table"
+        ).config
+
+        assert isinstance(config_test_table, SourceConfig)
+        assert isinstance(config_other_test_table, SourceConfig)
+
+        assert config_test_table == config_other_test_table
+        assert config_test_table == pytest.expected_config
+
+
+all_configs_project_source__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    tables:
+      - name: test_table
+        config:
+            enabled: True
+            quoting:
+                database: False
+                schema: False
+                identifier: False
+                column: False
+            freshness:
+                error_after: {count: 24, period: hour}
+                warn_after: {count: 12, period: hour}
+            loader: "a_loader"
+            loaded_at_field: some_column
+            database: custom_database
+            schema: custom_schema
+            meta: {'languages': ['python']}
+            tags: ["important_tag"]
+      - name: other_test_table
+"""
+
+
+# Test inheritence - set configs at project and source-table level - expect source-table level to win
+# expect fail - not implemented
+class TestSourceConfigsInheritence3(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": all_configs_project_source__schema_yml}
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"sources": {"tags": "project_level_important_tag"}}
+
+    @pytest.mark.xfail
+    def test_source_two_configs_source_table(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.other_test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+        config_other_test_table = manifest.sources.get(
+            "source.test.test_source.other_test_table"
+        ).config
+
+        assert isinstance(config_test_table, SourceConfig)
+        assert isinstance(config_other_test_table, SourceConfig)
+
+        assert config_test_table != config_other_test_table
+        assert config_test_table == pytest.expected_config
+
+        expected_project_level_config = SourceConfig(
+            enabled=True,
+            # tags = "project_level_important_tag",   # TODO: uncomment these once SourceConfig is updated
+        )
+
+        assert config_other_test_table == expected_project_level_config
+
+
+class SourceBackwardsCompatibility(SourceConfigTests):
+    @pytest.mark.xfail
+    def check_source_configs_and_properties(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+
+        # this is new functionality - but it currently passes since SourceConfig is not updated
+        # and is commented out in the setup becuse it is not updated with new configs
+        # even when properties are defined at the top level they should end up on the node.config
+        assert isinstance(config_test_table, SourceConfig)
+        assert config_test_table == pytest.expected_config
+
+        # this is existing functionality - should always pass
+        properties_test_table = manifest.sources.get("source.test.test_source.test_table")
+        assert properties_test_table.quoting == Quoting(
+            database=False, schema=False, identifier=False, column=False
+        )
+        assert properties_test_table.freshness == FreshnessThreshold(
+            warn_after=Time(count=12, period=TimePeriod.hour),
+            error_after=Time(count=24, period=TimePeriod.hour),
+            filter=None,
+        )
+        assert properties_test_table.loader == "a_loader"
+        assert properties_test_table.loaded_at_field == "some_column"
+        assert properties_test_table.database == "custom_database"
+        assert properties_test_table.schema == "custom_schema"
+        assert properties_test_table.meta == {}  # TODO: why is this empty
+        assert properties_test_table.tags == ["important_tag"]
+
+
+properties_as_configs__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    quoting:
+        database: False
+        schema: False
+        identifier: False
+        column: False
+    freshness:
+        error_after: {count: 24, period: hour}
+        warn_after: {count: 12, period: hour}
+    loader: "a_loader"
+    loaded_at_field: some_column
+    database: custom_database
+    schema: custom_schema
+    meta: {'languages': ['python']}
+    tags: ["important_tag"]
+    tables:
+      - name: test_table
+"""
+
+
+# Check backwards compatibility of setting configs as properties at top level
+# expect pass since the properties don't get copied to the node.config yet so
+# the values match since we haven't build the full SourceConfig yet
+class TestPropertiesAsConfigs(SourceBackwardsCompatibility):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": properties_as_configs__schema_yml}
+
+    @pytest.mark.xfail
+    def test_source_properties_as_configs(self, project):
+        self.check_source_configs_and_properties(project)
+
+
+configs_as_properties__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    config:
+        quoting:
+            database: False
+            schema: False
+            identifier: False
+            column: False
+        freshness:
+            error_after: {count: 24, period: hour}
+            warn_after: {count: 12, period: hour}
+        loader: "a_loader"
+        loaded_at_field: some_column
+        database: custom_database
+        schema: custom_schema
+        meta: {'languages': ['python']}
+        tags: ["important_tag"]
+    tables:
+      - name: test_table
+"""
+
+
+# Check backwards compatibility of setting configs as configs at top level and expect them to end up as properties
+# expect fail since the properties don't get copied to the node.config yet
+class TestConfigsAsProperties(SourceBackwardsCompatibility):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": configs_as_properties__schema_yml}
+
+    @pytest.mark.xfail
+    def test_source_configs_as_properties(self, project):
+        self.check_source_configs_and_properties(project)
+
+
+configs_properites__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    quoting:
+        database: False
+        schema: False
+        identifier: False
+        column: False
+    freshness:
+        error_after: {count: 24, period: hour}
+        warn_after: {count: 12, period: hour}
+    loader: "a_loader"
+    loaded_at_field: some_column
+    database: custom_database
+    schema: custom_schema
+    meta: {'languages': ['python']}
+    tags: ["important_tag"]
+    config:
+        enabled: True
+        quoting:
+            database: False
+            schema: False
+            identifier: False
+            column: False
+        freshness:
+            error_after: {count: 24, period: hour}
+            warn_after: {count: 12, period: hour}
+        loader: "a_loader"
+        loaded_at_field: some_column
+        database: custom_database
+        schema: custom_schema
+        meta: {'languages': ['python']}
+        tags: ["important_tag"]
+    tables:
+      - name: test_table
+      - name: other_test_table
+"""
+
+
+# Raise an error when properties are set at top level and also as configs
+class TestErrorSourceConfigProperty(SourceConfigTests):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": configs_properites__schema_yml}
+
+    @pytest.mark.xfail
+    def test_error_source_configs_properties(self, project):
+        # TODO: update below with correct exception/text/etc.  This is placeholder.
+        with pytest.raises(CompilationException) as exc:
+            run_dbt(["parse"])
+
+        assert "???" in str(exc.value)
+
+
+configs_properites__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    tables:
+      - name: test_table
+          config:
+            enabled: True
+            identifier: "seed"
+"""
+
+
+class TestSourceIdentifierConfig:
+    @pytest.fixture(scope="class", autouse=True)
+    def setUp(self):
+        pytest.expected_config = SourceConfig(
+            enabled=True,
+            # TODO: uncomment all this once it's added to SourceConfig, throws error right now
+            # identifier = "seed"
+        )
+
+
+identifier_config_source_table__schema_yml = """version: 2
+
+sources:
+  - name: test_source
+    tables:
+      - name: test_table
+        config:
+            identifier: "seed"
+      - name: other_test_table
+"""
+
+
+# Test identifier config at source table level in yml file.  This config differs by
+# table so should only be defined at this level.
+# expect fail - not implemented
+class TestSourceAllCTestSourceIdentifierConfig:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": identifier_config_source_table__schema_yml}
+
+    @pytest.mark.xfail
+    def test_source_identifier_config_source_table(self, project):
+        run_dbt(["parse"])
+        manifest = get_manifest(project.project_root)
+        assert "source.test.test_source.test_table" in manifest.sources
+        assert "source.test.test_source.other_test_table" in manifest.sources
+        config_test_table = manifest.sources.get("source.test.test_source.test_table").config
+        config_other_test_table = manifest.sources.get(
+            "source.test.test_source.other_test_table"
+        ).config
+
+        assert isinstance(config_test_table, SourceConfig)
+        assert isinstance(config_other_test_table, SourceConfig)
+
+        identifier_expected_config = SourceConfig(
+            enabled=True,
+            # TODO: uncomment this once it's added to SourceConfig, throws error right now
+            # identifier = "seed"
+        )
+
+        assert config_test_table != config_other_test_table
+        assert config_test_table == identifier_expected_config