-
Notifications
You must be signed in to change notification settings - Fork 2
/
fixtures.py
131 lines (102 loc) · 2.7 KB
/
fixtures.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""The pytest fixtures."""
from __future__ import annotations
import dataclasses
import os
import dbt.tracking
import pytest
from _pytest.fixtures import SubRequest
from dbt.clients.jinja import MacroGenerator
from dbt.config.runtime import RuntimeConfig
from dbt.context import providers
from dbt.contracts.graph.manifest import Manifest
from dbt.parser.manifest import ManifestLoader
from dbt.tracking import User
from .session import _SparkConnectionManager
from dbt.adapters.factory import ( # isort:skip
AdapterContainer,
get_adapter,
register_adapter,
)
dbt.tracking.active_user = User(os.getcwd())
@dataclasses.dataclass(frozen=True)
class Args:
"""
The arguments.
dbt is written as command line tool, therefore the entrypoints of dbt expect
(parsed) arguments. To reuse dbt's entrypoints we mock the (minimally)
arguments here.
"""
project_dir: str = os.getcwd()
@pytest.fixture
def config() -> RuntimeConfig:
"""
Get the (runtime) config.
Returns
-------
RuntimeConfig
The runtime config.
"""
# requires a profile in your project wich also exists in your profiles file
config = RuntimeConfig.from_args(Args())
return config
@pytest.fixture
def adapter(config: RuntimeConfig) -> AdapterContainer:
"""
Get the adapter.
Parameters
----------
config : RuntimeConfig
The runtime config.
Returns
-------
AdapterContainer
The adapter.
"""
register_adapter(config)
adapter = get_adapter(config)
connection_manager = _SparkConnectionManager(adapter.config)
adapter.connections = connection_manager
adapter.acquire_connection()
return adapter
@pytest.fixture
def manifest(
adapter: AdapterContainer,
) -> Manifest:
"""
Get the dbt manifest.
Parameters
----------
adapter : AdapterContainer
The adapter.
Returns
-------
Manifest
The manifest.
"""
manifest = ManifestLoader.get_full_manifest(adapter.config)
return manifest
@pytest.fixture
def macro_generator(
request: SubRequest, config: RuntimeConfig, manifest: Manifest
) -> MacroGenerator:
"""
Get a macro generator.
Parameters
----------
request : SubRequest
The pytest request containing the macro name.
config : RuntimeConfig
The runtime config.
manifest : Manifest
The manifest.
Returns
-------
MacroGenerator
The macro generator.
"""
macro = manifest.macros[request.param]
context = providers.generate_runtime_macro_context(
macro, config, manifest, macro.package_name
)
macro_generator = MacroGenerator(macro, context)
return macro_generator