test_graph.py 文件源码

def setUp(self):
        dbt.flags.STRICT_MODE = True

        def mock_write_gpickle(graph, outfile):
            self.graph_result = graph

        self.real_write_gpickle = nx.write_gpickle
        nx.write_gpickle = mock_write_gpickle

        self.graph_result = None

        self.profiles = {
            'test': {
                'outputs': {
                    'test': {
                        'type': 'postgres',
                        'threads': 4,
                        'host': 'database',
                        'port': 5432,
                        'user': 'root',
                        'pass': 'password',
                        'dbname': 'dbt',
                        'schema': 'dbt_test'
                    }
                },
                'target': 'test'
            }
        }

        self.real_dependency_projects = dbt.utils.dependency_projects
        dbt.utils.dependency_projects = MagicMock(return_value=[])

        self.mock_models = []
        self.mock_content = {}

        def mock_find_matching(root_path, relative_paths_to_search,
                               file_pattern):
            if 'sql' not in file_pattern:
                return []

            to_return = []

            if 'models' in relative_paths_to_search:
                to_return = to_return + self.mock_models

            return to_return

        self.real_find_matching = dbt.clients.system.find_matching
        dbt.clients.system.find_matching = MagicMock(
            side_effect=mock_find_matching)

        def mock_load_file_contents(path):
            return self.mock_content[path]

        self.real_load_file_contents = dbt.clients.system.load_file_contents
        dbt.clients.system.load_file_contents = MagicMock(
            side_effect=mock_load_file_contents)