forked from feast-dev/feast
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_cli_local.py
More file actions
183 lines (150 loc) · 6.37 KB
/
test_cli_local.py
File metadata and controls
183 lines (150 loc) · 6.37 KB
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
import tempfile
from contextlib import contextmanager
from pathlib import Path
from textwrap import dedent
import assertpy
import pytest
from feast.feature_store import FeatureStore
from tests.cli_utils import CliRunner
from tests.online_read_write_test import basic_rw_test
@pytest.mark.integration
def test_workflow() -> None:
"""
Test running apply on a sample repo, and make sure the infra gets created.
"""
runner = CliRunner()
with tempfile.TemporaryDirectory() as repo_dir_name, tempfile.TemporaryDirectory() as data_dir_name:
# Construct an example repo in a temporary dir
repo_path = Path(repo_dir_name)
data_path = Path(data_dir_name)
repo_config = repo_path / "feature_store.yaml"
repo_config.write_text(
dedent(
f"""
project: foo
registry: {data_path / "registry.db"}
provider: local
online_store:
path: {data_path / "online_store.db"}
offline_store:
type: bigquery
"""
)
)
repo_example = repo_path / "example.py"
repo_example.write_text(
(Path(__file__).parent / "example_feature_repo_1.py").read_text()
)
result = runner.run(["apply"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
# entity & feature view list commands should succeed
result = runner.run(["entities", "list"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
result = runner.run(["feature-views", "list"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
# entity & feature view describe commands should succeed when objects exist
result = runner.run(["entities", "describe", "driver"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
result = runner.run(
["feature-views", "describe", "driver_locations"], cwd=repo_path
)
assertpy.assert_that(result.returncode).is_equal_to(0)
# entity & feature view describe commands should fail when objects don't exist
result = runner.run(["entities", "describe", "foo"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(1)
result = runner.run(["feature-views", "describe", "foo"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(1)
# Doing another apply should be a no op, and should not cause errors
result = runner.run(["apply"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
basic_rw_test(
FeatureStore(repo_path=str(repo_path), config=None),
view_name="driver_locations",
)
result = runner.run(["teardown"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
@pytest.mark.integration
def test_non_local_feature_repo() -> None:
"""
Test running apply on a sample repo, and make sure the infra gets created.
"""
runner = CliRunner()
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(
"""
project: foo
registry: data/registry.db
provider: local
online_store:
path: data/online_store.db
offline_store:
type: bigquery
"""
)
)
repo_example = repo_path / "example.py"
repo_example.write_text(
(Path(__file__).parent / "example_feature_repo_1.py").read_text()
)
result = runner.run(["apply"], cwd=repo_path)
assertpy.assert_that(result.returncode).is_equal_to(0)
fs = FeatureStore(repo_path=str(repo_path))
assertpy.assert_that(fs.list_feature_views()).is_length(3)
result = runner.run(["teardown"], cwd=repo_path)
assertpy.assert_that(result.returncode).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
"""
)
)
(repo_path / "foo").mkdir()
repo_example = repo_path / "foo/provider.py"
repo_example.write_text((Path(__file__).parent / "foo_provider.py").read_text())
yield repo_path
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 Provider module 'feast_foo'"
)
# 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 Provider '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)