-
Notifications
You must be signed in to change notification settings - Fork 7
/
test_custom_online_store.py
48 lines (37 loc) · 1.31 KB
/
test_custom_online_store.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
import os
import re
from datetime import datetime
from feast import FeatureStore
from feature_repo.repo import driver, driver_hourly_stats_view
def test_end_to_end():
fs = FeatureStore("feature_repo/")
# apply repository
fs.apply([driver, driver_hourly_stats_view])
# load data into online store
fs.materialize_incremental(end_date=datetime.now())
# Read features from online store
feature_vector = fs.get_online_features(
features=["driver_hourly_stats:conv_rate"], entity_rows=[{"driver_id": 1001}]
).to_dict()
conv_rate = feature_vector["conv_rate"][0]
assert conv_rate > 0
# tear down feature store
fs.teardown()
def test_cli():
os.system(
"PYTHONPATH=$PYTHONPATH:/$(pwd) feast -c feature_repo apply"
)
os.system(
"PYTHONPATH=$PYTHONPATH:/$(pwd) feast -c feature_repo materialize-incremental 2021-08-19T22:29:28 > output"
)
with open("output", "r") as f:
output = f.read()
try:
if "feast_custom_online_store.mysql_online_store.MySQLOnlineStore" not in output:
raise Exception(
f'Failed to successfully use online store from CLI. Output:\n\n {output}'
)
finally:
os.system(
"PYTHONPATH=$PYTHONPATH:/$(pwd) feast -c feature_repo teardown"
)