Add references for new snapshot syntax in .yml files (#21)

Co-authored-by: Andrej Svec <andyswec@cisco.com>
slidoapp · Jan 20, 2022 · 20c1400 · 20c1400
1 parent ffd8aef
commit 20c1400
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+## Added
+- Support for snapshots.
 
 ## [0.1.8] - 2021-12-21
 ## Fixed

diff --git a/dbt_coverage/__init__.py b/dbt_coverage/__init__.py
@@ -82,6 +82,7 @@ class Manifest:
     sources: Dict[str, Dict[str, Dict]]
     models: Dict[str, Dict[str, Dict]]
     seeds: Dict[str, Dict[str, Dict]]
+    snapshots: Dict[str, Dict[str, Dict]]
     tests: Dict[str, Dict[str, List[Dict]]]
 
     @classmethod
@@ -101,9 +102,14 @@ def from_nodes(cls, manifest_nodes: Dict[str, Dict]) -> Manifest:
         seeds = {cls._full_table_name(table): cls._normalize_column_names(table['columns'])
                  for table in seeds}
 
+        snapshots = [table for table in manifest_nodes.values()
+                     if table['resource_type'] == 'snapshot']
+        snapshots = {cls._full_table_name(table): cls._normalize_column_names(table['columns'])
+                     for table in snapshots}
+
         tests = cls._parse_tests(manifest_nodes)
 
-        return Manifest(sources, models, seeds, tests)
+        return Manifest(sources, models, seeds, snapshots, tests)
 
     @classmethod
     def _parse_tests(cls, manifest_nodes: Dict[str, Dict]) -> Dict[str, Dict[str, List[Dict]]]:
@@ -115,7 +121,7 @@ def _parse_tests(cls, manifest_nodes: Dict[str, Dict]) -> Dict[str, Dict[str, Li
 
         id_to_table_name = {table_id: cls._full_table_name(table)
                             for table_id, table in manifest_nodes.items()
-                            if table['resource_type'] in ['source', 'model', 'seed']}
+                            if table['resource_type'] in ['source', 'model', 'seed', 'snapshot']}
 
         tests = {}
         for node in manifest_nodes.values():
@@ -513,16 +519,18 @@ def load_files(project_dir: Path) -> Catalog:
         manifest_source_table = manifest.sources.get(table_name, {})
         manifest_model_table = manifest.models.get(table_name, {})
         manifest_seed_table = manifest.seeds.get(table_name, {})
+        manifest_snapshot_table = manifest.snapshots.get(table_name, {})
         manifest_table_tests = manifest.tests.get(table_name, {})
 
         for catalog_column in catalog_table.columns.values():
             manifest_source_column = manifest_source_table.get(catalog_column.name)
             manifest_model_column = manifest_model_table.get(catalog_column.name)
             manifest_seed_column = manifest_seed_table.get(catalog_column.name)
+            manifest_snapshot_column = manifest_snapshot_table.get(catalog_column.name)
             manifest_column_tests = manifest_table_tests.get(catalog_column.name)
 
             manifest_column = manifest_source_column or manifest_model_column \
-                or manifest_seed_column or {}
+                or manifest_seed_column or manifest_snapshot_column or {}
             doc = manifest_column.get('description')
             catalog_column.doc = Column.is_valid_doc(doc)
             catalog_column.test = Column.is_valid_test(manifest_column_tests)