49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
|
"""Feature registry tests."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import feature_registrations # noqa: F401
|
||
|
|
from auth import AUTH_HEADER
|
||
|
|
from db import get_connection
|
||
|
|
from feature_registry import get_registered_features, sync_features_to_db
|
||
|
|
from tests.factories import provision_user_in_tenant
|
||
|
|
|
||
|
|
|
||
|
|
def test_feature_registration_required_fields():
|
||
|
|
keys = {f.key for f in get_registered_features()}
|
||
|
|
assert keys == {
|
||
|
|
"kairo.feature.registry",
|
||
|
|
"kairo.config.registry",
|
||
|
|
"kairo.prompt.registry",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_feature_sync_idempotent():
|
||
|
|
assert sync_features_to_db() == 0
|
||
|
|
assert sync_features_to_db() == 0
|
||
|
|
conn = get_connection()
|
||
|
|
try:
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute("SELECT COUNT(*) FROM features WHERE is_active = TRUE")
|
||
|
|
assert cur.fetchone()[0] == 3
|
||
|
|
finally:
|
||
|
|
conn.close()
|
||
|
|
|
||
|
|
|
||
|
|
def test_features_in_entitlements(client):
|
||
|
|
user = provision_user_in_tenant(tenant_role="member", portal_role="user")
|
||
|
|
login = client.post(
|
||
|
|
"/api/auth/login",
|
||
|
|
json={"email": user["email"], "password": user["password"]},
|
||
|
|
)
|
||
|
|
token = login.json()["token"]
|
||
|
|
res = client.get("/api/me/entitlements", headers={AUTH_HEADER: token})
|
||
|
|
features = res.json()["features"]
|
||
|
|
assert "kairo.prompt.registry" in features
|
||
|
|
assert features["kairo.prompt.registry"]["enabled"] is True
|
||
|
|
assert features["kairo.prompt.registry"]["module"] == "prompt"
|
||
|
|
|
||
|
|
|
||
|
|
def test_features_api_requires_auth(client):
|
||
|
|
assert client.get("/api/features").status_code == 401
|