FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
feast/sdk/python/tests/unit/cli/test_cli.py at pytorch-template · feast-dev/feast · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
feast-dev
/
feast
Public
Notifications
You must be signed in to change notification settings
Fork
1.4k
Star
7.2k
Code
Issues
218
Pull requests
191
Discussions
Actions
Security and quality
1
Insights
Additional navigation options
Code
Issues
Pull requests
Discussions
Actions
Security and quality
Insights
Expand file tree
Breadcrumbs
feast
/
sdk
/
python
/
tests
/
unit
/
cli
/
test_cli.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
229 lines (196 loc) · 8.88 KB
Breadcrumbs
feast
/
sdk
/
python
/
tests
/
unit
/
cli
/
test_cli.py
Copy path
File metadata and controls
229 lines (196 loc) · 8.88 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
import
os
import
tempfile
from
contextlib
import
contextmanager
from
pathlib
import
Path
from
textwrap
import
dedent
from
unittest
import
mock
from
assertpy
import
assertpy
from
tests
.
utils
.
cli_repo_creator
import
CliRunner
def
test_3rd_party_providers
()
->
None
:
"""
Test running apply on third party providers
"""
runner
=
CliRunner
()
# Check with incorrect built-in provider name (no dots)
with
setup_third_party_provider_repo
(
"feast123"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
1
)
assertpy
.
assert_that
(
output
).
contains
(
b"Provider 'feast123' is not implemented"
)
# Check with incorrect third-party provider name (with dots)
with
setup_third_party_provider_repo
(
"feast_foo.Provider"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
1
)
assertpy
.
assert_that
(
output
).
contains
(
b"Could not import module 'feast_foo' while attempting to load class 'Provider'"
)
# Check with incorrect third-party provider name (with dots)
with
setup_third_party_provider_repo
(
"foo.FooProvider"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
1
)
assertpy
.
assert_that
(
output
).
contains
(
b"Could not import class 'FooProvider' from module 'foo'"
)
# Check with correct third-party provider name
with
setup_third_party_provider_repo
(
"foo.provider.FooProvider"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
0
)
def
test_3rd_party_registry_store
()
->
None
:
"""
Test running apply on third party registry stores
"""
runner
=
CliRunner
()
# Check with incorrect built-in provider name (no dots)
with
setup_third_party_registry_store_repo
(
"feast123"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
1
)
assertpy
.
assert_that
(
output
).
contains
(
b'Registry store class name should end with "RegistryStore"'
)
# Check with incorrect third-party registry store name (with dots)
with
setup_third_party_registry_store_repo
(
"feast_foo.RegistryStore"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
1
)
assertpy
.
assert_that
(
output
).
contains
(
b"Could not import module 'feast_foo' while attempting to load class 'RegistryStore'"
)
# Check with incorrect third-party registry store name (with dots)
with
setup_third_party_registry_store_repo
(
"foo.FooRegistryStore"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
1
)
assertpy
.
assert_that
(
output
).
contains
(
b"Could not import class 'FooRegistryStore' from module 'foo'"
)
# Check with correct third-party registry store name
with
setup_third_party_registry_store_repo
(
"foo.registry_store.FooRegistryStore"
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
0
)
def
test_3rd_party_registry_store_with_fs_yaml_override
()
->
None
:
runner
=
CliRunner
()
fs_yaml_file
=
"test_fs.yaml"
with
setup_third_party_registry_store_repo
(
"foo.registry_store.FooRegistryStore"
,
fs_yaml_file_name
=
fs_yaml_file
)
as
repo_path
:
return_code
,
output
=
runner
.
run_with_output
(
[
"--feature-store-yaml"
,
fs_yaml_file
,
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
0
)
def
test_3rd_party_registry_store_with_fs_yaml_override_by_env_var
()
->
None
:
runner
=
CliRunner
()
fs_yaml_file
=
"test_fs.yaml"
with
setup_third_party_registry_store_repo
(
"foo.registry_store.FooRegistryStore"
,
fs_yaml_file_name
=
fs_yaml_file
)
as
repo_path
:
custom_yaml_path
=
os
.
path
.
join
(
repo_path
,
fs_yaml_file
)
with
mock
.
patch
.
dict
(
"os.environ"
, {
"FEAST_FS_YAML_FILE_PATH"
:
custom_yaml_path
},
clear
=
True
):
return_code
,
output
=
runner
.
run_with_output
([
"apply"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
0
)
@
contextmanager
def
setup_third_party_provider_repo
(
provider_name
:
str
):
with
tempfile
.
TemporaryDirectory
()
as
repo_dir_name
:
# Construct an example repo in a temporary dir
repo_path
=
Path
(
repo_dir_name
)
repo_config
=
repo_path
/
"feature_store.yaml"
repo_config
.
write_text
(
dedent
(
f"""
project: foo
registry: data/registry.db
provider:
{
provider_name
}
online_store:
path: data/online_store.db
type: sqlite
offline_store:
type: file
entity_key_serialization_version: 3
"""
)
)
(
repo_path
/
"foo"
).
mkdir
()
repo_example
=
repo_path
/
"foo/provider.py"
repo_example
.
write_text
(
(
Path
(
__file__
).
parents
[
2
]
/
"foo_provider.py"
).
read_text
()
)
yield
repo_path
@
contextmanager
def
setup_third_party_registry_store_repo
(
registry_store
:
str
,
fs_yaml_file_name
:
str
=
"feature_store.yaml"
):
with
tempfile
.
TemporaryDirectory
()
as
repo_dir_name
:
# Construct an example repo in a temporary dir
repo_path
=
Path
(
repo_dir_name
)
repo_config
=
repo_path
/
fs_yaml_file_name
repo_config
.
write_text
(
dedent
(
f"""
project: foo
registry:
registry_store_type:
{
registry_store
}
path: foobar://foo.bar
provider: local
online_store:
path: data/online_store.db
type: sqlite
offline_store:
type: file
entity_key_serialization_version: 3
"""
)
)
(
repo_path
/
"foo"
).
mkdir
()
repo_example
=
repo_path
/
"foo/registry_store.py"
repo_example
.
write_text
(
(
Path
(
__file__
).
parents
[
2
]
/
"foo_registry_store.py"
).
read_text
()
)
yield
repo_path
def
test_cli_configuration
():
"""
Unit test for the 'feast configuration' command
"""
runner
=
CliRunner
()
with
setup_third_party_provider_repo
(
"local"
)
as
repo_path
:
# Run the 'feast configuration' command
return_code
,
output
=
runner
.
run_with_output
([
"configuration"
],
cwd
=
repo_path
)
# Assertions
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
0
)
assertpy
.
assert_that
(
output
).
contains
(
b"project: foo"
)
assertpy
.
assert_that
(
output
).
contains
(
b"provider: local"
)
assertpy
.
assert_that
(
output
).
contains
(
b"type: sqlite"
)
assertpy
.
assert_that
(
output
).
contains
(
b"path: data/online_store.db"
)
assertpy
.
assert_that
(
output
).
contains
(
b"type: file"
)
assertpy
.
assert_that
(
output
).
contains
(
b"entity_key_serialization_version: 3"
)
def
test_cli_materialize_disable_event_timestamp
():
"""
Unit test for the 'feast materialize --disable-event-timestamp' command
"""
runner
=
CliRunner
()
with
setup_third_party_provider_repo
(
"local"
)
as
repo_path
:
# Test that --disable-event-timestamp flag works without timestamps
return_code
,
output
=
runner
.
run_with_output
(
[
"materialize"
,
"--disable-event-timestamp"
],
cwd
=
repo_path
)
# Should succeed (though may not have data to materialize)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
0
)
# Test that providing timestamps with --disable-event-timestamp fails
return_code
,
output
=
runner
.
run_with_output
(
[
"materialize"
,
"--disable-event-timestamp"
,
"2021-01-01T00:00:00"
,
"2021-01-02T00:00:00"
,
],
cwd
=
repo_path
,
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
2
)
# Click usage error
assertpy
.
assert_that
(
output
).
contains
(
b"Cannot specify START_TS or END_TS when --disable-event-timestamp is used"
)
# Test that missing timestamps without flag fails
return_code
,
output
=
runner
.
run_with_output
([
"materialize"
],
cwd
=
repo_path
)
assertpy
.
assert_that
(
return_code
).
is_equal_to
(
2
)
# Click usage error
assertpy
.
assert_that
(
output
).
contains
(
b"START_TS and END_TS are required unless --disable-event-timestamp is used"
)
Back
|
FazBrowse Home
|
New Git URL