"""Tests for template value source API: CRUD, validate-template, delete-protection.""" from unittest.mock import MagicMock import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from ledgrab.api import dependencies as deps from ledgrab.api.routes.value_sources import router from ledgrab.storage.value_source_store import ValueSourceStore @pytest.fixture def _route_db(tmp_path): from ledgrab.storage.database import Database db = Database(tmp_path / "test.db") yield db db.close() @pytest.fixture def store(_route_db): return ValueSourceStore(_route_db) @pytest.fixture def client(store): app = FastAPI() app.include_router(router) from ledgrab.api.auth import verify_api_key app.dependency_overrides[verify_api_key] = lambda: "test-user" app.dependency_overrides[deps.get_value_source_store] = lambda: store app.dependency_overrides[deps.get_processor_manager] = lambda: MagicMock() app.dependency_overrides[deps.get_output_target_store] = lambda: MagicMock( get_all_targets=lambda: [] ) deps._deps["processor_manager"] = MagicMock() return TestClient(app, raise_server_exceptions=False) def _create(client, **over): body = { "source_type": "template", "name": "Combo", "template": "min(a * 2, 1)", "inputs": [{"name": "a", "value_source_id": ""}], "default_value": 0.2, } body.update(over) return client.post("/api/v1/value-sources", json=body) class TestCRUD: def test_create_get_list_roundtrip(self, client): r = _create(client) assert r.status_code == 201, r.text body = r.json() assert body["source_type"] == "template" assert body["return_type"] == "float" assert body["template"] == "min(a * 2, 1)" assert body["inputs"] == [{"name": "a", "value_source_id": ""}] assert body["default_value"] == 0.2 sid = body["id"] got = client.get(f"/api/v1/value-sources/{sid}").json() assert got["template"] == "min(a * 2, 1)" lst = client.get("/api/v1/value-sources").json() assert any(s["id"] == sid and s["source_type"] == "template" for s in lst["sources"]) def test_update(self, client): sid = _create(client).json()["id"] r = client.put( f"/api/v1/value-sources/{sid}", json={"source_type": "template", "template": "clamp(a * 3)"}, ) assert r.status_code == 200, r.text assert r.json()["template"] == "clamp(a * 3)" def test_create_compile_error_returns_400(self, client): r = _create(client, template="a +") assert r.status_code == 400 def test_create_reserved_name_returns_400(self, client): r = _create(client, inputs=[{"name": "min", "value_source_id": ""}]) assert r.status_code == 400 class TestDeleteProtection: def test_delete_blocked_when_referenced(self, client): base = client.post( "/api/v1/value-sources", json={"source_type": "static", "name": "Base", "value": 0.5}, ).json() _create( client, name="Uses", template="b", inputs=[{"name": "b", "value_source_id": base["id"]}], ) r = client.delete(f"/api/v1/value-sources/{base['id']}") assert r.status_code == 400 assert "referenced by" in r.json()["detail"] class TestValidateEndpoint: def _validate(self, client, **body): return client.post("/api/v1/value-sources/validate-template", json=body) def test_valid_expression(self, client): r = self._validate( client, template="min(a, b)", inputs=[{"name": "a", "value_source_id": ""}, {"name": "b", "value_source_id": ""}], ) assert r.status_code == 200 data = r.json() assert data["valid"] is True assert set(data["variables"]) == {"a", "b"} def test_compile_error(self, client): r = self._validate(client, template="a +", inputs=[]) data = r.json() assert data["valid"] is False assert data["error"] def test_reserved_name(self, client): r = self._validate( client, template="min(0,1)", inputs=[{"name": "raw", "value_source_id": ""}] ) assert r.json()["valid"] is False def test_missing_input_is_warning_not_error(self, client): r = self._validate( client, template="a", inputs=[{"name": "a", "value_source_id": "vs_nope"}] ) data = r.json() assert data["valid"] is True assert data["warnings"] def test_unbound_variable_is_error(self, client): # Typo: expression uses 'ha_enti' but the input is named 'ha_entity'. r = self._validate( client, template="ha_enti", inputs=[{"name": "ha_entity", "value_source_id": ""}] ) data = r.json() assert data["valid"] is False assert any("unbound" in e for e in data["errors"]) def test_cycle_detected_with_id(self, client): t1 = _create(client, name="T1", template="clamp(0.5)", inputs=[]).json() t2 = _create( client, name="T2", template="x", inputs=[{"name": "x", "value_source_id": t1["id"]}], ).json() # Editing t1 to point at t2 would close a cycle. r = self._validate( client, template="x", inputs=[{"name": "x", "value_source_id": t2["id"]}], id=t1["id"] ) assert r.json()["valid"] is False class TestResponseMapCoverage: def test_template_in_response_map(self): from ledgrab.api.routes.value_sources import _RESPONSE_MAP from ledgrab.storage.value_source import TemplateValueSource assert TemplateValueSource in _RESPONSE_MAP def test_template_in_all_unions(self): from ledgrab.api.schemas import value_sources as sch for union_name in ("ValueSourceResponse", "ValueSourceCreate", "ValueSourceUpdate"): src = repr(getattr(sch, union_name)) assert "template" in src.lower() or "Template" in src