264 lines
9.6 KiB
Python
264 lines
9.6 KiB
Python
"""Sync engine tests: backend/flows/*.json -> DB + LFX (LFX real, DB sqlite)."""
|
|
|
|
import asyncio
|
|
import json
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.flows import store, sync
|
|
from app.flows.lfx_client import LfxClient, LfxNotFound
|
|
from app.flows.schemas import SyncReport
|
|
from app.flows.sync import lfx_id_for
|
|
from app.settings import settings
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures" / "flows"
|
|
|
|
|
|
def load_fixture(name: str) -> dict:
|
|
return json.loads((FIXTURES / name).read_text())
|
|
|
|
|
|
def _lfx() -> LfxClient:
|
|
return LfxClient(settings.lfx_url, api_key=settings.lfx_api_key or None)
|
|
|
|
|
|
def _seed_dir(tmp_path: Path, name: str = "simple_rule.json") -> Path:
|
|
shutil.copy(FIXTURES / name, tmp_path / name)
|
|
return tmp_path
|
|
|
|
|
|
async def test_sync_adds_new_flow(session, lfx_server, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert report.added == ["simple_rule"]
|
|
assert report.errors == {}
|
|
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
|
assert definition is not None
|
|
assert definition.name == "simple_rule"
|
|
versions = store.list_versions(session, definition.id)
|
|
assert [v.version_no for v in versions] == [1]
|
|
assert versions[0].lfx_flow_id == lfx_id_for(definition.id, 1)
|
|
assert versions[0].deployed_at is not None
|
|
|
|
|
|
async def test_sync_skips_unchanged(session, lfx_server, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
first = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert first.added == ["simple_rule"]
|
|
second = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert second.skipped == ["simple_rule"]
|
|
assert second.added == second.updated == []
|
|
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
|
assert len(store.list_versions(session, definition.id)) == 1
|
|
|
|
|
|
async def test_sync_updates_on_change_and_replaces_lfx(session, lfx_server, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
|
old_lfx_id = lfx_id_for(definition.id, 1)
|
|
|
|
doc = load_fixture("simple_rule.json")
|
|
doc["description"] = "geaendert"
|
|
(tmp_path / "simple_rule.json").write_text(json.dumps(doc))
|
|
|
|
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert report.updated == ["simple_rule"]
|
|
versions = store.list_versions(session, definition.id)
|
|
assert [v.version_no for v in versions] == [1, 2]
|
|
assert versions[1].lfx_flow_id == lfx_id_for(definition.id, 2)
|
|
# Superseded LFX registration is gone.
|
|
with pytest.raises(LfxNotFound):
|
|
await _lfx().flow_info(old_lfx_id)
|
|
|
|
|
|
async def test_sync_deletes_missing_file_with_cascade(session, lfx_server, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
|
version = store.list_versions(session, definition.id)[0]
|
|
run = store.create_run(session, version, input={"a": 1}, tenant_id="default")
|
|
lfx_id = lfx_id_for(definition.id, 1)
|
|
|
|
(tmp_path / "simple_rule.json").unlink()
|
|
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert report.deleted == ["simple_rule"]
|
|
assert store.get_definition_by_source_path(session, "simple_rule.json") is None
|
|
with pytest.raises(store.NotFoundError):
|
|
store.get_run(session, run.id)
|
|
with pytest.raises(LfxNotFound):
|
|
await _lfx().flow_info(lfx_id)
|
|
|
|
|
|
async def test_sync_reports_invalid_json(session, lfx_server, tmp_path):
|
|
(tmp_path / "kaputt.json").write_text("{kein json")
|
|
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert "kaputt.json" in report.errors
|
|
assert report.added == []
|
|
assert store.list_definitions(session) == []
|
|
|
|
|
|
async def test_sync_missing_dir_raises(session, lfx_server, tmp_path):
|
|
with pytest.raises(FileNotFoundError):
|
|
await sync.sync_from_directory(session, _lfx(), tmp_path / "gibts-nicht")
|
|
|
|
|
|
async def test_sync_rename_updates_definition_name(session, lfx_server, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
doc = load_fixture("simple_rule.json")
|
|
doc["name"] = "umbenannt"
|
|
(tmp_path / "simple_rule.json").write_text(json.dumps(doc))
|
|
report = await sync.sync_from_directory(session, _lfx(), tmp_path)
|
|
assert report.updated == ["simple_rule"]
|
|
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
|
assert definition.name == "umbenannt"
|
|
|
|
|
|
async def test_sync_endpoint_returns_report(
|
|
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
|
):
|
|
await login()
|
|
override_langflow() # empty Langflow: file-only pass
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
_seed_dir(tmp_path)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["added"] == ["simple_rule"]
|
|
assert body["pulled"] == []
|
|
assert body["errors"] == {}
|
|
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.json()["skipped"] == ["simple_rule"]
|
|
|
|
|
|
def _langflow_doc(name: str, langflow_id: str) -> dict:
|
|
doc = load_fixture("simple_rule.json")
|
|
return {
|
|
"id": langflow_id,
|
|
"name": name,
|
|
"description": "aus Langflow",
|
|
"endpoint_name": None,
|
|
"data": doc["data"],
|
|
"tags": [],
|
|
"webhook": False,
|
|
"is_component": False,
|
|
}
|
|
|
|
|
|
async def test_sync_endpoint_pulls_from_langflow_then_syncs(
|
|
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
|
):
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
langflow_id = "22222222-3333-4444-8555-666666666666"
|
|
override_langflow(
|
|
flows=[{"id": langflow_id, "name": "Gehaltserhöhung"}],
|
|
docs={langflow_id: _langflow_doc("Gehaltserhöhung", langflow_id)},
|
|
)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["pulled"] == ["Gehaltserhöhung"]
|
|
assert body["added"] == ["Gehaltserhöhung"]
|
|
assert body["errors"] == {}
|
|
assert (tmp_path / "Gehaltserhöhung.json").is_file()
|
|
|
|
# Second pass: same content pulls again but syncs nothing new.
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["pulled"] == ["Gehaltserhöhung"]
|
|
assert body["skipped"] == ["Gehaltserhöhung"]
|
|
|
|
|
|
async def test_sync_endpoint_pull_reuses_same_named_file(
|
|
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
|
):
|
|
"""A pulled flow seeded from simple_rule.json updates it in place
|
|
instead of duplicating it as simple-rule.json (slug drift)."""
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
_seed_dir(tmp_path) # fixture shape, name "simple_rule"
|
|
langflow_id = "44444444-5555-4666-8777-888888888888"
|
|
override_langflow(
|
|
flows=[{"id": langflow_id, "name": "simple_rule"}],
|
|
docs={langflow_id: _langflow_doc("simple_rule", langflow_id)},
|
|
)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["pulled"] == ["simple_rule"]
|
|
assert body["added"] == ["simple_rule"]
|
|
assert body["errors"] == {}
|
|
assert not (tmp_path / "simple-rule.json").exists()
|
|
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.json()["skipped"] == ["simple_rule"]
|
|
|
|
|
|
async def test_sync_endpoint_pull_error_continues(
|
|
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
|
):
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
good_id = "33333333-4444-4555-8666-777777777777"
|
|
override_langflow(
|
|
flows=[{"id": good_id, "name": "Gut"}, {"id": "kaputt-id", "name": "Kaputt"}],
|
|
docs={good_id: _langflow_doc("Gut", good_id)},
|
|
)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["pulled"] == ["Gut"]
|
|
assert body["added"] == ["Gut"]
|
|
assert "kaputt-id" in body["errors"]
|
|
|
|
|
|
async def test_sync_endpoint_langflow_unreachable(
|
|
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
|
):
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
override_langflow(flows=[], list_status=0)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 503
|
|
|
|
|
|
async def test_sync_endpoint_langflow_error(
|
|
client, lfx_server, tmp_path, monkeypatch, login, override_langflow
|
|
):
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
override_langflow(flows=[], list_status=500)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 502
|
|
|
|
|
|
async def test_watch_forever_triggers_on_file_change(tmp_path):
|
|
import contextlib
|
|
|
|
calls: list[str] = []
|
|
stop = asyncio.Event()
|
|
|
|
def run():
|
|
calls.append("sync")
|
|
return SyncReport()
|
|
|
|
task = asyncio.create_task(sync.watch_forever(stop, tmp_path, run_sync=run))
|
|
try:
|
|
await asyncio.sleep(0.5) # let the watcher arm
|
|
(tmp_path / "neu.json").write_text("{}")
|
|
deadline = time.time() + 15
|
|
while not calls and time.time() < deadline:
|
|
await asyncio.sleep(0.1)
|
|
assert calls == ["sync"]
|
|
finally:
|
|
stop.set()
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|