text: import hashlib
  import json
  import os
  from datetime import datetime
  from unittest.mock import ANY, patch
  
  from pytest import mark
  from tests.integration.base import DBTIntegrationTest, use_profile, AnyFloat, \
      AnyString, AnyStringWith, normalize, Normalized
  
  import dbt.tracking
  import dbt.version
  from dbt.exceptions import CompilationException
  
  
  def _read_file(path):
      with open(path, 'r') as fp:
          return fp.read().replace('\r', '').replace('\\r', '')
  
  
  class LineIndifferent:
      def __init__(self, expected):
          self.expected = expected.replace('\r', '')
  
      def __eq__(self, other):
          return self.expected == other.replace('\r', '')
  
      def __repr__(self):
          return 'LineIndifferent({!r})'.format(self.expected)
  
      def __str__(self):
          return self.__repr__()
  
  
  class OneOf:
      def __init__(self, *options):
          self.options = options
  
      def __eq__(self, other):
          return any(o == other for o in self.options)
  
      def __repr__(self):
          return 'OneOf(*{!r})'.format(self.options)
  
      def __str__(self):
          return self.__repr__()
  
  
  def _read_json(path):
      # read json generated by dbt.
      with open(path) as fp:
          return json.load(fp)
  
  
  def walk_files(path):
      for root, dirs, files in os.walk(path):
          for basename in files:
              yield os.path.join(root, basename)
  
  
  class TestDocsGenerate(DBTIntegrationTest):
      setup_alternate_db = True
  
      def adapter_case(self, value):
          return value.upper()
  
      def setUp(self):
          super().setUp()
          self.maxDiff = None
          self.alternate_schema = self.unique_schema() + '_test'
          self.alternate_schema = self.alternate_schema.upper()
  
          self._created_schemas.add(self.alternate_schema)
          os.environ['DBT_ENV_CUSTOM_ENV_env_key'] = 'env_value'
  
      def tearDown(self):
          super().tearDown()
          del os.environ['DBT_ENV_CUSTOM_ENV_env_key']
  
      @property
      def schema(self):
          return 'docs_generate'
  
      @staticmethod
      def dir(path):
          return normalize(path)
  
      @property
      def models(self):
          return self.dir("models")
  
      @property
      def project_config(self):
          return {
              'config-version': 2,
              'quoting': {
                  'identifier': False
              }
          }
  
      def run_and_generate(self, extra=None, seed_count=1, model_count=2, alternate_db=None, args=None):
          if alternate_db is None:
              alternate_db = self.alternative_database
          project = {
              "seed-paths": [self.dir("seed")],
              'macro-paths': [self.dir('macros')],
              'snapshot-paths': [self.dir('snapshot')],
              'vars': {
                  'alternate_db': alternate_db,
                  'alternate_schema': self.alternate_schema,
              },
              'seeds': {
                  'quote_columns': True,
              },
          }
          if extra:
              project.update(extra)
          self.use_default_project(project)
  
          vars_arg = '--vars={{test_schema: {}}}'.format(self.unique_schema())
  
          self.assertEqual(len(self.run_dbt(["seed", vars_arg])), seed_count)
          self.assertEqual(len(self.run_dbt(['run', vars_arg])), model_count)
          os.remove(normalize('target/manifest.json'))
          os.remove(normalize('target/run_results.json'))
          self.generate_start_time = datetime.utcnow()
          base_args = ['docs', 'generate', vars_arg]
          if args:
              base_args.extend(args)
          self.run_dbt(base_args)
  
      def _no_stats(self):
          return {
              'has_stats': {
                  'id': 'has_stats',
                  'label': 'Has Stats?',
                  'value': False,
                  'description': 'Indicates whether there are statistics for this table',
                  'include': False,
              },
          }
  
      def _snowflake_stats(self):
          return {
              'has_stats': {
                  'id': 'has_stats',
                  'label': 'Has Stats?',
                  'value': True,
                  'description': 'Indicates whether there are statistics for this table',
                  'include': False,
              },
              'bytes': {
                  'id': 'bytes',
                  'label': 'Approximate Size',
                  'value': AnyFloat(),
                  'description': 'Approximate size of the table as reported by Snowflake',
                  'include': True,
              },
              'last_modified': {
                  'id': 'last_modified',
                  'label': 'Last Modified',
                  'value': AnyString(),
                  'description': 'The timestamp for last update/change',
                  'include': True,
              },
              'row_count': {
                  'id': 'row_count',
                  'label': 'Row Count',
                  'value': 1.0,
                  'description': 'An approximate count of rows in this table',
                  'include': True,
              }
          }
  
      def _expected_catalog(self, id_type, text_type, time_type, view_type,
                            table_type, model_stats, seed_stats=None, case=None,
                            case_columns=False, model_database=None):
          if case is None:
              def case(x): return x
          col_case = case if case_columns else lambda x: x
  
          if seed_stats is None:
              seed_stats = model_stats
  
          if model_database is None:
              model_database = self.default_database
          my_schema_name = self.unique_schema()
          role = self.get_role()
          expected_cols = {
              col_case('id'): {
                  'name': col_case('id'),
                  'index': 1,
                  'type': id_type,
                  'comment': None,
              },
              col_case('first_name'): {
                  'name': col_case('first_name'),
                  'index': 2,
                  'type': text_type,
                  'comment': None,
              },
              col_case('email'): {
                  'name': col_case('email'),
                  'index': 3,
                  'type': text_type,
                  'comment': None,
              },
              col_case('ip_address'): {
                  'name': col_case('ip_address'),
                  'index': 4,
                  'type': text_type,
                  'comment': None,
              },
              col_case('updated_at'): {
                  'name': col_case('updated_at'),
                  'index': 5,
                  'type': time_type,
                  'comment': None,
              },
          }
          return {
              'nodes': {
                  'model.test.model': {
                      'unique_id': 'model.test.model',
                      'metadata': {
                          'schema': my_schema_name,
                          'database': model_database,
                          'name': case('model'),
                          'type': view_type,
                          'comment': None,
                          'owner': role,
                      },
                      'stats': model_stats,
                      'columns': expected_cols,
                  },
                  'model.test.second_model': {
                      'unique_id': 'model.test.second_model',
                      'metadata': {
                          'schema': self.alternate_schema,
                          'database': self.default_database,
                          'name': case('second_model'),
                          'type': view_type,
                          'comment': None,
                          'owner': role,
                      },
                      'stats': model_stats,
                      'columns': expected_cols,
                  },
                  'seed.test.seed': {
                      'unique_id': 'seed.test.seed',
                      'metadata': {
                          'schema': my_schema_name,
                          'database': self.default_database,
                          'name': case('seed'),
                          'type': table_type,
                          'comment': None,
                          'owner': role,
                      },
                      'stats': seed_stats,
                      'columns': expected_cols,
                  },
              },
              'sources': {
                  'source.test.my_source.my_table': {
                      'unique_id': 'source.test.my_source.my_table',
                      'metadata': {
                          'schema': my_schema_name,
                          'database': self.default_database,
                          'name': case('seed'),
                          'type': table_type,
                          'comment': None,
                          'owner': role,
                      },
                      'stats': seed_stats,
                      'columns': expected_cols,
                  },
              },
          }
  
      def get_role(self):
          return self.run_sql('select current_role()', fetch='one')[0]
  
      def expected_snowflake_catalog(self, case_columns=False):
          return self._expected_catalog(
              id_type='NUMBER',
              text_type='TEXT',
              time_type='TIMESTAMP_NTZ',
              view_type='VIEW',
              table_type='BASE TABLE',
              model_stats=self._no_stats(),
              seed_stats=self._snowflake_stats(),
              case=lambda x: x.upper(),
              model_database=self.alternative_database,
              case_columns=case_columns,
          )
  
      def verify_catalog(self, expected):
          self.assertTrue(os.path.exists('./target/catalog.json'))
  
          catalog = _read_json('./target/catalog.json')
  
          assert set(catalog) == {'errors', 'metadata', 'nodes', 'sources'}
  
          self.verify_metadata(
              catalog['metadata'], 'https://schemas.getdbt.com/dbt/catalog/v1.json')
          assert not catalog['errors']
  
          for key in 'nodes', 'sources':
              assert catalog[key] == expected[key]
  
      def verify_manifest_macros(self, manifest, expected=None):
          self.assertIn('macros', manifest)
          if expected is None:
              self._verify_generic_macro_structure(manifest)
              return
          for unique_id, expected_macro in expected.items():
              self.assertIn(unique_id, manifest['macros'])
              actual_macro = manifest['macros'][unique_id]
              self.assertEqual(expected_macro, actual_macro)
  
      def rendered_model_config(self, **updates):
          result = {
              'database': None,
              'schema': None,
              'alias': None,
              'enabled': True,
              'materialized': 'view',
              'pre-hook': [],
              'post-hook': [],
              'column_types': {},
              'quoting': {},
              'tags': [],
              'persist_docs': {},
              'full_refresh': None,
              'on_schema_change': 'ignore',
              'meta': {},
          }
          result.update(updates)
          return result
  
      def unrendered_model_config(self, **updates):
          return updates
  
      def rendered_seed_config(self, **updates):
          result = {
              'enabled': True,
              'materialized': 'seed',
              'persist_docs': {},
              'pre-hook': [],
              'post-hook': [],
              'column_types': {},
              'quoting': {},
              'tags': [],
              'quote_columns': True,
              'full_refresh': None,
              'on_schema_change': 'ignore',
              'database': None,
              'schema': None,
              'alias': None,
              'meta': {},
          }
          result.update(updates)
          return result
  
      def unrendered_seed_config(self, **updates):
          result = {'quote_columns': True}
          result.update(updates)
          return result
  
      def rendered_snapshot_config(self, **updates):
          result = {
              'database': None,
              'schema': None,
              'alias': None,
              'enabled': True,
              'materialized': 'snapshot',
              'pre-hook': [],
              'post-hook': [],
              'column_types': {},
              'quoting': {},
              'tags': [],
              'persist_docs': {},
              'full_refresh': None,
              'on_schema_change': 'ignore',
              'strategy': 'check',
              'check_cols': 'all',
              'unique_key': 'id',
              'target_schema': None,
              'meta': {},
          }
          result.update(updates)
          return result
  
      def unrendered_snapshot_config(self, **updates):
          result = {
              'check_cols': 'all',
              'strategy': 'check',
              'target_schema': None,
              'unique_key': 'id'
          }
          result.update(updates)
          return result
  
      def rendered_tst_config(self, **updates):
          result = {
              'enabled': True,
              'materialized': 'test',
              'tags': [],
              'severity': 'ERROR',
              'store_failures': None,
              'warn_if': '!= 0',
              'error_if': '!= 0',
              'fail_calc': 'count(*)',
              'where': None,
              'limit': None,
              'database': None,
              'schema': 'dbt_test__audit',
              'alias': None,
              'meta': {},
          }
          result.update(updates)
          return result
  
      def unrendered_tst_config(self, **updates):
          result = {}
          result.update(updates)
          return result
  
      def _verify_generic_macro_structure(self, manifest):
          # just test a known global macro to avoid having to update this every
          # time they change.
          self.assertIn('macro.dbt.column_list', manifest['macros'])
          macro = manifest['macros']['macro.dbt.column_list']
          self.assertEqual(
              set(macro),
              {
                  'path', 'original_file_path', 'package_name',
                  'root_path', 'name', 'unique_id', 'tags', 'resource_type',
                  'depends_on', 'meta', 'description', 'patch_path', 'arguments',
                  'macro_sql', 'docs', 'created_at',
              }
          )
          # Don't compare the sql, just make sure it exists
          self.assertTrue(len(macro['macro_sql']) > 10)
          without_sql = {
              k: v for k, v in macro.items()
              if k not in {'macro_sql'}
          }
          # Windows means we can't hard-code these.
          helpers_path = Normalized('macros/materializations/helpers.sql')
          root_path = Normalized(os.path.join(
              self.dbt_core_install_root, 'include', 'global_project'
          ))
          self.assertEqual(
              {
                  'path': helpers_path,
                  'original_file_path': helpers_path,
                  'package_name': 'dbt',
                  'root_path': root_path,
                  'name': 'column_list',
                  'unique_id': 'macro.dbt.column_list',
                  'created_at': ANY,
                  'tags': [],
                  'resource_type': 'macro',
                  'depends_on': {'macros': []},
                  'description': '',
                  'docs': {'show': True},
                  'patch_path': None,
                  'meta': {},
                  'arguments': [],
              },
              without_sql,
          )
  
      def expected_seeded_manifest(self, model_database=None, quote_model=False):
          models_path = self.dir('models')
          model_sql_path = os.path.join(models_path, 'model.sql')
          second_model_sql_path = os.path.join(models_path, 'second_model.sql')
          model_schema_yml_path = os.path.join(models_path, 'schema.yml')
          seed_schema_yml_path = os.path.join(self.dir('seed'), 'schema.yml')
          seed_path = self.dir(os.path.join('seed', 'seed.csv'))
          snapshot_path = self.dir(os.path.join('snapshot', 'snapshot_seed.sql'))
  
          my_schema_name = self.unique_schema()
          test_audit_schema = my_schema_name + '_dbt_test__audit'
  
          if model_database is None:
              model_database = self.alternative_database
  
          model_config = self.rendered_model_config(database=model_database)
          second_config = self.rendered_model_config(
              schema=self.alternate_schema[-4:])
  
          unrendered_model_config = self.unrendered_model_config(
              database=model_database, materialized='view')
          unrendered_second_config = self.unrendered_model_config(
              schema=self.alternate_schema[-4:], materialized='view')
  
          seed_config = self.rendered_seed_config()
          unrendered_seed_config = self.unrendered_seed_config()
  
          test_config = self.rendered_tst_config()
          unrendered_test_config = self.unrendered_tst_config()
  
          snapshot_config = self.rendered_snapshot_config(
              target_schema=self.alternate_schema)
          unrendered_snapshot_config = self.unrendered_snapshot_config(
              target_schema=self.alternate_schema
          )
  
          quote_database = quote_schema = self.adapter_type != 'snowflake'
          relation_name_node_format = self._relation_name_format(
              quote_database, quote_schema, quote_model
          )
          relation_name_source_format = self._relation_name_format(
              quote_database, quote_schema, quote_identifier=True
          )
  
          return {
              'dbt_schema_version': 'https://schemas.getdbt.com/dbt/manifest/v3.json',
              'dbt_version': dbt.version.__version__,
              'nodes': {
                  'model.test.model': {
                      'compiled_path': Normalized('target/compiled/test/models/model.sql'),
                      'build_path': None,
                      'created_at': ANY,
                      'name': 'model',
                      'root_path': self.test_root_realpath,
                      'relation_name': relation_name_node_format.format(
                          model_database, my_schema_name, 'model'
                      ),
                      'resource_type': 'model',
                      'path': 'model.sql',
                      'original_file_path': model_sql_path,
                      'package_name': 'test',
                      'raw_sql': LineIndifferent(_read_file(model_sql_path).rstrip('\r\n')),
                      'refs': [['seed']],
                      'sources': [],
                      'depends_on': {'nodes': ['seed.test.seed'], 'macros': []},
                      'unique_id': 'model.test.model',
                      'fqn': ['test', 'model'],
                      'tags': [],
                      'meta': {},
                      'config': model_config,
                      'schema': my_schema_name,
                      'database': model_database,
                      'deferred': False,
                      'alias': 'model',
                      'description': 'The test model',
                      'columns': {
                          'id': {
                              'name': 'id',
                              'description': 'The user ID number',
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'first_name': {
                              'name': 'first_name',
                              'description': "The user's first name",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'email': {
                              'name': 'email',
                              'description': "The user's email",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'ip_address': {
                              'name': 'ip_address',
                              'description': "The user's IP address",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'updated_at': {
                              'name': 'updated_at',
                              'description': "The last time this user's email was updated",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                      },
                      'patch_path': 'test://' + model_schema_yml_path,
                      'docs': {'show': False},
                      'compiled': True,
                      'compiled_sql': ANY,
                      'extra_ctes_injected': True,
                      'extra_ctes': [],
                      'checksum': self._checksum_file(model_sql_path),
                      'unrendered_config': unrendered_model_config,
                  },
                  'model.test.second_model': {
                      'compiled_path': Normalized('target/compiled/test/models/second_model.sql'),
                      'build_path': None,
                      'created_at': ANY,
                      'name': 'second_model',
                      'root_path': self.test_root_realpath,
                      'relation_name': relation_name_node_format.format(
                          self.default_database, self.alternate_schema,
                          'second_model'
                      ),
                      'resource_type': 'model',
                      'path': 'second_model.sql',
                      'original_file_path': second_model_sql_path,
                      'package_name': 'test',
                      'raw_sql': LineIndifferent(_read_file(second_model_sql_path).rstrip('\r\n')),
                      'refs': [['seed']],
                      'sources': [],
                      'depends_on': {'nodes': ['seed.test.seed'], 'macros': []},
                      'unique_id': 'model.test.second_model',
                      'fqn': ['test', 'second_model'],
                      'tags': [],
                      'meta': {},
                      'config': second_config,
                      'schema': self.alternate_schema,
                      'database': self.default_database,
                      'deferred': False,
                      'alias': 'second_model',
                      'description': 'The second test model',
                      'columns': {
                          'id': {
                              'name': 'id',
                              'description': 'The user ID number',
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'first_name': {
                              'name': 'first_name',
                              'description': "The user's first name",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'email': {
                              'name': 'email',
                              'description': "The user's email",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'ip_address': {
                              'name': 'ip_address',
                              'description': "The user's IP address",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'updated_at': {
                              'name': 'updated_at',
                              'description': "The last time this user's email was updated",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                      },
                      'patch_path': 'test://' + model_schema_yml_path,
                      'docs': {'show': False},
                      'compiled': True,
                      'compiled_sql': ANY,
                      'extra_ctes_injected': True,
                      'extra_ctes': [],
                      'checksum': self._checksum_file(second_model_sql_path),
                      'unrendered_config': unrendered_second_config
                  },
                  'seed.test.seed': {
                      'compiled_path': None,
                      'build_path': None,
                      'created_at': ANY,
                      'compiled': True,
                      'compiled_sql': '',
                      'config': seed_config,
                      'patch_path': 'test://' + seed_schema_yml_path,
                      'path': 'seed.csv',
                      'name': 'seed',
                      'root_path': self.test_root_realpath,
                      'relation_name': relation_name_node_format.format(
                          self.default_database, my_schema_name, 'seed'
                      ),
                      'resource_type': 'seed',
                      'raw_sql': '',
                      'package_name': 'test',
                      'original_file_path': seed_path,
                      'refs': [],
                      'sources': [],
                      'depends_on': {'nodes': [], 'macros': []},
                      'unique_id': 'seed.test.seed',
                      'fqn': ['test', 'seed'],
                      'tags': [],
                      'meta': {},
                      'schema': my_schema_name,
                      'database': self.default_database,
                      'alias': 'seed',
                      'deferred': False,
                      'description': 'The test seed',
                      'columns': {
                          'id': {
                              'name': 'id',
                              'description': 'The user ID number',
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'first_name': {
                              'name': 'first_name',
                              'description': "The user's first name",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'email': {
                              'name': 'email',
                              'description': "The user's email",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'ip_address': {
                              'name': 'ip_address',
                              'description': "The user's IP address",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                          'updated_at': {
                              'name': 'updated_at',
                              'description': "The last time this user's email was updated",
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          },
                      },
                      'docs': {'show': True},
                      'compiled': True,
                      'compiled_sql': '',
                      'extra_ctes_injected': True,
                      'extra_ctes': [],
                      'checksum': self._checksum_file(seed_path),
                      'unrendered_config': unrendered_seed_config,
                  },
                  'test.test.not_null_model_id.d01cc630e6': {
                      'alias': 'not_null_model_id',
                      'compiled_path': Normalized('target/compiled/test/models/schema.yml/not_null_model_id.sql'),
                      'build_path': None,
                      'created_at': ANY,
                      'column_name': 'id',
                      'columns': {},
                      'config': test_config,
                      'sources': [],
                      'depends_on': {
                          'macros': ['macro.dbt.test_not_null', 'macro.dbt.get_where_subquery'],
                          'nodes': ['model.test.model'],
                      },
                      'deferred': False,
                      'description': '',
                      'fqn': ['test', 'not_null_model_id'],
                      'name': 'not_null_model_id',
                      'original_file_path': model_schema_yml_path,
                      'package_name': 'test',
                      'patch_path': None,
                      'path': Normalized('not_null_model_id.sql'),
                      'raw_sql': "{{ test_not_null(**_dbt_generic_test_kwargs) }}",
                      'refs': [['model']],
                      'relation_name': None,
                      'resource_type': 'test',
                      'root_path': self.test_root_realpath,
                      'schema': test_audit_schema,
                      'database': self.default_database,
                      'tags': [],
                      'meta': {},
                      'unique_id': 'test.test.not_null_model_id.d01cc630e6',
                      'docs': {'show': True},
                      'compiled': True,
                      'compiled_sql': AnyStringWith('where id is null'),
                      'extra_ctes_injected': True,
                      'extra_ctes': [],
                      'test_metadata': {
                          'namespace': None,
                          'name': 'not_null',
                          'kwargs': {
                              'column_name': 'id',
                              'model': "{{ get_where_subquery(ref('model')) }}",
                          },
                      },
                      'checksum': {'name': 'none', 'checksum': ''},
                      'unrendered_config': unrendered_test_config,
                  },
                  'snapshot.test.snapshot_seed': {
                      'alias': 'snapshot_seed',
                      'compiled_path': None,
                      'build_path': None,
                      'created_at': ANY,
                      'checksum': self._checksum_file(snapshot_path),
                      'columns': {},
                      'compiled': True,
                      'compiled_sql': ANY,
                      'config': snapshot_config,
                      'database':  self.default_database,
                      'deferred': False,
                      'depends_on': {
                          'macros': [],
                          'nodes': ['seed.test.seed'],
                      },
                      'description': '',
                      'docs': {'show': True},
                      'extra_ctes': [],
                      'extra_ctes_injected': True,
                      'fqn': ['test', 'snapshot_seed', 'snapshot_seed'],
                      'meta': {},
                      'name': 'snapshot_seed',
                      'original_file_path': snapshot_path,
                      'package_name': 'test',
                      'patch_path': None,
                      'path': normalize('snapshot_seed.sql'),
                      'raw_sql': LineIndifferent(
                          _read_file(snapshot_path)
                              .replace('{% snapshot snapshot_seed %}', '')
                              .replace('{% endsnapshot %}', '')),
                      'refs': [['seed']],
                      'relation_name': relation_name_node_format.format(
                          self.default_database, self.alternate_schema,
                          'snapshot_seed'
                      ),
                      'resource_type': 'snapshot',
                      'root_path': self.test_root_realpath,
                      'schema': self.alternate_schema,
                      'sources': [],
                      'tags': [],
                      'unique_id': 'snapshot.test.snapshot_seed',
                      'unrendered_config': unrendered_snapshot_config,
                  },
                  'test.test.test_nothing_model_.5d38568946': {
                      'alias': 'test_nothing_model_',
                      'compiled_path': Normalized('target/compiled/test/models/schema.yml/test_nothing_model_.sql'),
                      'build_path': None,
                      'created_at': ANY,
                      'column_name': None,
                      'columns': {},
                      'config': test_config,
                      'sources': [],
                      'depends_on': {
                          'macros': ['macro.test.test_nothing', 'macro.dbt.get_where_subquery'],
                          'nodes': ['model.test.model'],
                      },
                      'deferred': False,
                      'description': '',
                      'fqn': ['test', 'test_nothing_model_'],
                      'name': 'test_nothing_model_',
                      'original_file_path': model_schema_yml_path,
                      'package_name': 'test',
                      'patch_path': None,
                      'path': normalize('test_nothing_model_.sql'),
                      'raw_sql': "{{ test.test_nothing(**_dbt_generic_test_kwargs) }}",
                      'refs': [['model']],
                      'relation_name': None,
                      'resource_type': 'test',
                      'root_path': self.test_root_realpath,
                      'schema': test_audit_schema,
                      'database': self.default_database,
                      'tags': [],
                      'meta': {},
                      'unique_id': 'test.test.test_nothing_model_.5d38568946',
                      'docs': {'show': True},
                      'compiled': True,
                      'compiled_sql': AnyStringWith('select 0'),
                      'extra_ctes_injected': True,
                      'extra_ctes': [],
                      'test_metadata': {
                          'namespace': 'test',
                          'name': 'nothing',
                          'kwargs': {
                              'model': "{{ get_where_subquery(ref('model')) }}",
                          },
                      },
                      'checksum': {'name': 'none', 'checksum': ''},
                      'unrendered_config': unrendered_test_config,
                  },
                  'test.test.unique_model_id.67b76558ff': {
                      'alias': 'unique_model_id',
                      'compiled_path': Normalized('target/compiled/test/models/schema.yml/unique_model_id.sql'),
                      'build_path': None,
                      'created_at': ANY,
                      'column_name': 'id',
                      'columns': {},
                      'config': test_config,
                      'sources': [],
                      'depends_on': {
                          'macros': ['macro.dbt.test_unique', 'macro.dbt.get_where_subquery'],
                          'nodes': ['model.test.model'],
                      },
                      'deferred': False,
                      'description': '',
                      'fqn': ['test', 'unique_model_id'],
                      'name': 'unique_model_id',
                      'original_file_path': model_schema_yml_path,
                      'package_name': 'test',
                      'patch_path': None,
                      'path': normalize('unique_model_id.sql'),
                      'raw_sql': "{{ test_unique(**_dbt_generic_test_kwargs) }}",
                      'refs': [['model']],
                      'relation_name': None,
                      'resource_type': 'test',
                      'root_path': self.test_root_realpath,
                      'schema': test_audit_schema,
                      'database': self.default_database,
                      'tags': [],
                      'meta': {},
                      'unique_id': 'test.test.unique_model_id.67b76558ff',
                      'docs': {'show': True},
                      'compiled': True,
                      'compiled_sql': AnyStringWith('count(*)'),
                      'extra_ctes_injected': True,
                      'extra_ctes': [],
                      'test_metadata': {
                          'namespace': None,
                          'name': 'unique',
                          'kwargs': {
                              'column_name': 'id',
                              'model': "{{ get_where_subquery(ref('model')) }}",
                          },
                      },
                      'checksum': {'name': 'none', 'checksum': ''},
                      'unrendered_config': unrendered_test_config,
                  },
              },
              'sources': {
                  'source.test.my_source.my_table': {
                      'created_at': ANY,
                      'columns': {
                          'id': {
                              'description': 'An ID field',
                              'name': 'id',
                              'data_type': None,
                              'meta': {},
                              'quote': None,
                              'tags': [],
                          }
                      },
                      'config': {
                          'enabled': True,
                      },
                      'quoting': {
                          'database': None,
                          'schema': None,
                          'identifier': True,
                          'column': None,
                      },
                      'database': self.default_database,
                      'description': 'My table',
                      'external': None,
                      'freshness': {'error_after': {'count': None, 'period': None}, 'warn_after': {'count': None, 'period': None}, 'filter': None},
                      'identifier': 'seed',
                      'loaded_at_field': None,
                      'loader': 'a_loader',
                      'meta': {},
                      'name': 'my_table',
                      'original_file_path': self.dir('models/schema.yml'),
                      'package_name': 'test',
                      'path': self.dir('models/schema.yml'),
                      'patch_path': None,
                      'relation_name': relation_name_source_format.format(
                          self.default_database, my_schema_name, 'seed'
                      ),
                      'resource_type': 'source',
                      'root_path': self.test_root_realpath,
                      'schema': my_schema_name,
                      'source_description': 'My source',
                      'source_name': 'my_source',
                      'source_meta': {},
                      'tags': [],
                      'unique_id': 'source.test.my_source.my_table',
                      'fqn': ['test', 'my_source', 'my_table'],
                      'unrendered_config': {},
                  },
              },
              'exposures': {
                  'exposure.test.notebook_exposure': {
                      'created_at': ANY,
                      'depends_on': {
                          'macros': [],
                          'nodes': ['model.test.model', 'model.test.second_model']
                      },
                      'description': 'A description of the complex exposure\n',
                      'fqn': ['test', 'notebook_exposure'],
                      'maturity': 'medium',
                      'meta': {'tool': 'my_tool', 'languages': ['python']},
                      'tags': ['my_department'],
                      'name': 'notebook_exposure',
                      'original_file_path': self.dir('models/schema.yml'),
                      'owner': {
                          'email': 'something@example.com',
                          'name': 'Some name'
                      },
                      'package_name': 'test',
                      'path': 'schema.yml',
                      'refs': [['model'], ['second_model']],
                      'resource_type': 'exposure',
                      'root_path': self.test_root_realpath,
                      'sources': [],
                      'type': 'notebook',
                      'unique_id': 'exposure.test.notebook_exposure',
                      'url': 'http://example.com/notebook/1'
                  },
                  'exposure.test.simple_exposure': {
                      'created_at': ANY,
                      'depends_on': {
                          'macros': [],
                          'nodes': [
                              'source.test.my_source.my_table',
                              'model.test.model'
                          ],
                      },
                      'description': '',
                      'fqn': ['test', 'simple_exposure'],
                      'name': 'simple_exposure',
                      'original_file_path': self.dir('models/schema.yml'),
                      'owner': {
                          'email': 'something@example.com',
                          'name': None,
                      },
                      'package_name': 'test',
                      'path': 'schema.yml',
                      'refs': [['model']],
                      'resource_type': 'exposure',
                      'root_path': self.test_root_realpath,
                      'sources': [['my_source', 'my_table']],
                      'type': 'dashboard',
                      'unique_id': 'exposure.test.simple_exposure',
                      'url': None,
                      'maturity': None,
                      'meta': {},
                      'tags': []
                  }
              },
              'selectors': {},
              'parent_map': {
                  'model.test.model': ['seed.test.seed'],
                  'model.test.second_model': ['seed.test.seed'],
                  'exposure.test.notebook_exposure': ['model.test.model', 'model.test.second_model'],
                  'exposure.test.simple_exposure': ['model.test.model', 'source.test.my_source.my_table'],
                  'seed.test.seed': [],
                  'snapshot.test.snapshot_seed': ['seed.test.seed'],
                  'source.test.my_source.my_table': [],
                  'test.test.not_null_model_id.d01cc630e6': ['model.test.model'],
                  'test.test.test_nothing_model_.5d38568946': ['model.test.model'],
                  'test.test.unique_model_id.67b76558ff': ['model.test.model'],
              },
              'child_map': {
                  'model.test.model': [
                      'exposure.test.notebook_exposure',
                      'exposure.test.simple_exposure',
                      'test.test.not_null_model_id.d01cc630e6',
                      'test.test.test_nothing_model_.5d38568946',
                      'test.test.unique_model_id.67b76558ff',
                  ],
                  'model.test.second_model': ['exposure.test.notebook_exposure'],
                  'exposure.test.notebook_exposure': [],
                  'exposure.test.simple_exposure': [],
                  'seed.test.seed': ['model.test.model',
                                     'model.test.second_model',
                                     'snapshot.test.snapshot_seed'],
                  'snapshot.test.snapshot_seed': [],
                  'source.test.my_source.my_table': ['exposure.test.simple_exposure'],
                  'test.test.not_null_model_id.d01cc630e6': [],
                  'test.test.test_nothing_model_.5d38568946': [],
                  'test.test.unique_model_id.67b76558ff': [],
              },
              'docs': {
                  'dbt.__overview__': ANY,
                  'test.macro_info': ANY,
                  'test.macro_arg_info': ANY,
              },
              'disabled': {},
          }
  
      def _checksum_file(self, path):
          """windows has silly git behavior that adds newlines, and python does
          silly things if we just open(..., 'r').encode('utf-8').
          """
          with open(self.dir(path), 'rb') as fp:
              hashed = hashlib.sha256(fp.read()).hexdigest()
          return {
              'name': 'sha256',
              'checksum': hashed,
          }
  
      def _path_to(self, searched_path: str, relative_path: str):
          return {
              'searched_path': normalize(searched_path),
              'relative_path': normalize(relative_path),
              'project_root': normalize(self.test_root_dir),
          }
  
      def _absolute_path_to(self, searched_path: str, relative_path: str):
          return os.path.join(
              normalize(self.test_root_dir),
              normalize(searched_path),
              normalize(relative_path)
          )
  
      def _relation_name_format(self, quote_database: bool, quote_schema: bool,
                                quote_identifier: bool):
          return ".".join((
              self._quote("{0}") if quote_database else '{0}',
              self._quote("{1}") if quote_schema else '{1}',
              self._quote("{2}") if quote_identifier else '{2}',
          ))
  
      def verify_metadata(self, metadata, dbt_schema_version):
          assert 'generated_at' in metadata
          self.assertBetween(metadata['generated_at'],
                             start=self.generate_start_time)
          assert 'dbt_version' in metadata
          assert metadata['dbt_version'] == dbt.version.__version__
          assert 'dbt_schema_version' in metadata
          assert metadata['dbt_schema_version'] == dbt_schema_version
          assert metadata['invocation_id'] == dbt.tracking.active_user.invocation_id
          key = 'env_key'
          if os.name == 'nt':
              key = key.upper()
          assert metadata['env'] == {
              key: 'env_value'
          }
  
      def verify_manifest(self, expected_manifest):
          self.assertTrue(os.path.exists('./target/manifest.json'))
  
          manifest = _read_json('./target/manifest.json')
  
          manifest_keys = frozenset({
              'nodes', 'sources', 'macros', 'parent_map', 'child_map',
              'docs', 'metadata', 'docs', 'disabled', 'exposures', 'selectors',
          })
  
          self.assertEqual(frozenset(manifest), manifest_keys)
  
          for key in manifest_keys:
              if key == 'macros':
                  self.verify_manifest_macros(
                      manifest, expected_manifest.get('macros'))
              elif key == 'metadata':
                  metadata = manifest['metadata']
                  self.verify_metadata(
                      metadata, 'https://schemas.getdbt.com/dbt/manifest/v3.json')
                  assert 'project_id' in metadata and metadata[
                      'project_id'] == '098f6bcd4621d373cade4e832627b4f6'
                  assert 'send_anonymous_usage_stats' in metadata and metadata[
                      'send_anonymous_usage_stats'] is False
                  assert 'adapter_type' in metadata and metadata['adapter_type'] == self.adapter_type
              else:
                  self.assertIn(key, expected_manifest)  # sanity check
                  self.assertEqual(manifest[key], expected_manifest[key])
  
      def _quote(self, value):
          quote_char = '"'
          return '{0}{1}{0}'.format(quote_char, value)
  
      def expected_run_results(self):
          """
          The expected results of this run.
          """
  
          return [
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'model.test.model',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'model.test.second_model',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'seed.test.seed',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'snapshot.test.snapshot_seed',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'test.test.not_null_model_id.d01cc630e6',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'test.test.test_nothing_model_.5d38568946',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
              {
                  'status': 'success',
                  'message': None,
                  'execution_time': AnyFloat(),
                  'unique_id': 'test.test.unique_model_id.67b76558ff',
                  'adapter_response': ANY,
                  'thread_id': ANY,
                  'timing': [ANY, ANY],
                  'failures': ANY,
              },
          ]
  
      def verify_run_results(self, expected_run_results):
          run_results = _read_json('./target/run_results.json')
          assert 'metadata' in run_results
          self.verify_metadata(
              run_results['metadata'], 'https://schemas.getdbt.com/dbt/run-results/v3.json')
          self.assertIn('elapsed_time', run_results)
          self.assertGreater(run_results['elapsed_time'], 0)
          self.assertTrue(
              isinstance(run_results['elapsed_time'], float),
              "run_results['elapsed_time'] is of type {}, expected float".format(
                  str(type(run_results['elapsed_time'])))
          )
  
          assert 'args' in run_results
          # sort the results so we can make reasonable assertions
          run_results['results'].sort(key=lambda r: r['unique_id'])
          assert run_results['results'] == expected_run_results
          set(run_results) == {'elapsed_time', 'results', 'metadata'}
  
      @use_profile('snowflake')
      def test__snowflake__run_and_generate(self):
          self.run_and_generate()
  
          self.verify_catalog(self.expected_snowflake_catalog())
          self.verify_manifest(self.expected_seeded_manifest())
          self.verify_run_results(self.expected_run_results())
  
      @use_profile('snowflake')
      def test__snowflake__run_and_generate_ignore_quoting_parameter(self):
          # with optional adapters, this package could easily just not exist!
          # accordingly, only run it when we think snowflake things should work
          from dbt.adapters.snowflake import connections as snowflake_conn
          old_connect = snowflake_conn.snowflake.connector.connect
  
          def connect(*args, **kwargs):
              kwargs['session_parameters'] = {
                  'QUOTED_IDENTIFIERS_IGNORE_CASE': True
              }
              return old_connect(*args, **kwargs)
  
          with patch.object(snowflake_conn.snowflake.connector, 'connect', connect):
              self.run_and_generate({
                  'quoting': {
                      'identifier': True,
                  }
              })
  
          self.verify_catalog(self.expected_snowflake_catalog(case_columns=True))
          self.verify_manifest(self.expected_seeded_manifest(quote_model=True))
          self.verify_run_results(self.expected_run_results())
  