forked from dbt-labs/dbt-core
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_version.py
176 lines (154 loc) · 7.02 KB
/
test_version.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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
import unittest
from unittest.mock import patch, MagicMock
import dbt.main
import dbt.version
class VersionTest(unittest.TestCase):
@patch("dbt.version.__version__", "0.10.0")
@patch('dbt.version._get_dbt_plugins_info', autospec=True)
@patch('dbt.version.requests.get')
def test_versions_equal(self, mock_get, mock_get_dbt_plugins_info):
mock_get.return_value.json.return_value = {
'info': {'version': '0.10.0'}
}
mock_get_dbt_plugins_info.return_value = [
('dbt-postgres', '0.10.0'),
('dbt-redshift', '0.10.0'),
('dbt-bigquery', '0.10.0'),
('dbt-snowflake', '0.10.0')
]
latest_version = dbt.version.get_latest_version()
installed_version = dbt.version.get_installed_version()
version_information = dbt.version.get_version_information()
expected_version_information = "installed version: 0.10.0\n" \
" latest version: 0.10.0\n\n" \
"Up to date!\n\n" \
"Plugins:\n" \
" - dbt-postgres: 0.10.0\n" \
" - dbt-redshift: 0.10.0\n" \
" - dbt-bigquery: 0.10.0\n" \
" - dbt-snowflake: 0.10.0\n"
self.assertEqual(latest_version, installed_version)
self.assertEqual(latest_version, installed_version)
self.assertMultiLineEqual(version_information,
expected_version_information)
@patch("dbt.version.__version__", "0.10.2-a1")
@patch('dbt.version._get_dbt_plugins_info', autospec=True)
@patch('dbt.version.requests.get')
def test_installed_version_greater(self, mock_get, mock_get_dbt_plugins_info):
mock_get.return_value.json.return_value = {
'info': {'version': '0.10.1'}
}
mock_get_dbt_plugins_info.return_value = [
('dbt-postgres', '0.10.0'),
('dbt-redshift', '0.10.0'),
('dbt-bigquery', '0.10.0'),
('dbt-snowflake', '0.10.0')
]
latest_version = dbt.version.get_latest_version()
installed_version = dbt.version.get_installed_version()
version_information = dbt.version.get_version_information()
expected_version_information = "installed version: 0.10.2-a1\n" \
" latest version: 0.10.1\n\n" \
"Your version of dbt is ahead of the latest release!\n\n" \
"Plugins:\n" \
" - dbt-postgres: 0.10.0\n" \
" - dbt-redshift: 0.10.0\n" \
" - dbt-bigquery: 0.10.0\n" \
" - dbt-snowflake: 0.10.0\n"
assert installed_version > latest_version
self.assertMultiLineEqual(version_information,
expected_version_information)
@patch("dbt.version.__version__", "0.9.5")
@patch('dbt.version._get_dbt_plugins_info', autospec=True)
@patch('dbt.version.requests.get')
def test_installed_version_lower(self, mock_get, mock_get_dbt_plugins_info):
mock_get.return_value.json.return_value = {
'info': {'version': '0.10.0'}
}
mock_get_dbt_plugins_info.return_value = [
('dbt-postgres', '0.10.0'),
('dbt-redshift', '0.10.0'),
('dbt-bigquery', '0.10.0'),
('dbt-snowflake', '0.10.0')
]
latest_version = dbt.version.get_latest_version()
installed_version = dbt.version.get_installed_version()
version_information = dbt.version.get_version_information()
expected_version_information = "installed version: 0.9.5\n" \
" latest version: 0.10.0\n\n" \
"Your version of dbt is out of date! " \
"You can find instructions for upgrading here:\n" \
"https://docs.getdbt.com/docs/installation\n\n" \
"Plugins:\n" \
" - dbt-postgres: 0.10.0\n" \
" - dbt-redshift: 0.10.0\n" \
" - dbt-bigquery: 0.10.0\n" \
" - dbt-snowflake: 0.10.0\n"
assert installed_version < latest_version
self.assertMultiLineEqual(version_information,
expected_version_information)
# suppress having version info printed to the screen during tests.
@patch('sys.stderr')
@patch('dbt.version.requests.get')
def test_dbt_version_flag(self, mock_get, stderr):
mock_get.return_value.json.return_value = {
'info': {'version': '0.10.1'}
}
with self.assertRaises(SystemExit) as exc:
dbt.main.handle_and_check(['--version'])
self.assertEqual(exc.exception.code, 0)
@patch('glob.glob', autospec=True)
@patch('importlib.util.find_spec', autospec=True)
def test_get_adapter_plugin_names(self, mock_find_spec, mock_glob):
mock_submodule = unittest.mock.MagicMock()
mock_find_spec.return_value = mock_submodule
mock_submodule.submodule_search_locations = ['/tmp/dbt/adapters']
def glob_side_effect(path: str) -> str:
return [
path.replace('*', 'postgres'),
path.replace('*', 'snowflake'),
]
mock_glob.side_effect = glob_side_effect
self.assertEqual(
list(dbt.version._get_adapter_plugin_names()),
['postgres', 'snowflake'],
)
mock_find_spec.assert_called_once_with('dbt.adapters')
@patch('glob.glob', autospec=True)
@patch('importlib.util.find_spec', autospec=True)
def test_get_adapter_plugin_names(self, mock_find_spec, mock_glob):
mock_submodule = unittest.mock.MagicMock()
mock_find_spec.return_value = mock_submodule
mock_submodule.submodule_search_locations = [
'/tmp/dbt/plugins/postgres/dbt/adapters/', '/tmp/dbt/plugins/snowflake/dbt/adapters/'
]
def glob_side_effect(path: str) -> str:
assert 'postgres' in path or 'snowflake' in path
if 'postgres' in path:
return [path.replace('*', 'postgres')]
elif 'snowflake' in path:
return [path.replace('*', 'snowflake')]
mock_glob.side_effect = glob_side_effect
self.assertEqual(
list(dbt.version._get_adapter_plugin_names()),
['postgres', 'snowflake'],
)
mock_find_spec.assert_called_once_with('dbt.adapters')
@patch('dbt.version._get_adapter_plugin_names', autospec=True)
@patch('importlib.import_module', autospec=True)
def test_get_dbt_plugins_info_with_version_info(
self, mock_mod, mock_get_plugin_names
):
mock_get_plugin_names.return_value = ['postgres', 'snowflake']
mod_version = unittest.mock.Mock()
mock_mod.return_value = mod_version
mod_version.version = '1.0'
self.assertEqual(
list(dbt.version._get_dbt_plugins_info()),
[('postgres', '1.0'), ('snowflake', '1.0')]
)
mock_get_plugin_names.assert_called_once_with()
mock_mod.assert_has_calls([
unittest.mock.call('dbt.adapters.postgres.__version__'),
unittest.mock.call('dbt.adapters.snowflake.__version__')
])