from feast import Entity, FeatureService, FeatureStore, FeatureView, Field, FileSource

from feast.api.registry.rest.rest_registry_server import RestRegistryServer

from feast.data_source import RequestSource

from feast.infra.offline_stores.file_source import SavedDatasetFileStorage

from feast.on_demand_feature_view import on_demand_feature_view

from feast.repo_config import RepoConfig

from feast.saved_dataset import SavedDataset

from feast.types import Float64, Int64

storage=saved_dataset_storage,

tags={"environment": "test", "version": "1.0"},

)

input_request = RequestSource(

name="input_request_source",

schema=[

Field(name="request_feature", dtype=Float64),

],

)

@on_demand_feature_view(

sources=[user_profile_feature_view, input_request],

schema=[

Field(name="combined_feature", dtype=Float64),

],

description="On-demand feature view with request source for testing",

)

def test_on_demand_feature_view(features_df: pd.DataFrame) -> pd.DataFrame:

df = pd.DataFrame()

df["combined_feature"] = features_df["age"] + features_df["request_feature"]

return df

# Apply objects

store.apply(

user_behavior_feature_view,

user_preferences_feature_view,

user_feature_service,

test_on_demand_feature_view,

]

)

store._registry.apply_saved_dataset(test_saved_dataset, "demo_project")

for fv in data["featureViews"]:

assert "type" in fv

assert fv["type"] is not None

assert fv["type"] == "featureView"

assert fv["type"] in ["featureView", "onDemandFeatureView"]

# Test single endpoint

response = fastapi_test_app.get("/feature_views/user_profile?project=demo_project")

data_source_filtered_views = data["featureViews"]

assert len(data_source_filtered_views) <= len(all_feature_views)

# Test filtering on-demand feature views by request source data source

response = fastapi_test_app.get(

"/feature_views?project=demo_project&data_source=input_request_source"

)

assert response.status_code == 200

data = response.json()

assert "featureViews" in data

odfv_data_source_filtered_views = data["featureViews"]

# Should find the on-demand feature view that uses the request source

assert len(odfv_data_source_filtered_views) > 0

odfv_found = False

for fv in odfv_data_source_filtered_views:

if fv["type"] == "onDemandFeatureView":

odfv_found = True

break

assert odfv_found, (

"On-demand feature view should be found when filtering by request source data source"

)

response = fastapi_test_app.get(

"/feature_views?project=demo_project&feature_service=user_service"

)