forked from feast-dev/feast
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_data_sources.py
More file actions
34 lines (27 loc) · 1.37 KB
/
test_data_sources.py
File metadata and controls
34 lines (27 loc) · 1.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
import pytest
from feast import ValueType
from feast.data_source import PushSource, RequestDataSource, RequestSource
from feast.infra.offline_stores.bigquery_source import BigQuerySource
def test_push_with_batch():
push_source = PushSource(
name="test",
schema={"user_id": ValueType.INT64, "ltv": ValueType.INT64},
batch_source=BigQuerySource(table="test.test"),
)
push_source_proto = push_source.to_proto()
assert push_source_proto.push_options is not None
assert push_source_proto.push_options.HasField("batch_source")
push_source_unproto = PushSource.from_proto(push_source_proto)
assert push_source.name == push_source_unproto.name
assert push_source.schema == push_source_unproto.schema
assert push_source.batch_source.name == push_source_unproto.batch_source.name
def test_request_data_source_deprecation():
with pytest.warns(DeprecationWarning):
request_data_source = RequestDataSource(
name="vals_to_add",
schema={"val_to_add": ValueType.INT64, "val_to_add_2": ValueType.INT64},
)
request_data_source_proto = request_data_source.to_proto()
returned_request_source = RequestSource.from_proto(request_data_source_proto)
assert returned_request_source.name == request_data_source.name
assert returned_request_source.schema == request_data_source.schema