257 lines
8.9 KiB
Python
257 lines
8.9 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 LfxNotFound
|
|
from app.flows.schemas import SyncReport
|
|
from app.flows.sync import lfx_id_for
|
|
from app.settings import settings
|
|
|
|
from .conftest import FIXTURES_DIR, langflow_export_doc, load_flow_fixture
|
|
|
|
|
|
def _seed_dir(tmp_path: Path, name: str = "simple_rule.json") -> Path:
|
|
shutil.copy(FIXTURES_DIR / name, tmp_path / name)
|
|
return tmp_path
|
|
|
|
|
|
async def test_sync_adds_new_flow(session, lfx_server, lfx_client, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
report = await sync.sync_from_directory(session, lfx_client, 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_ignored_files(session, lfx_server, lfx_client, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
(tmp_path / ".flowsignore").write_text("simple_rule\n", encoding="utf-8")
|
|
report = await sync.sync_from_directory(session, lfx_client, tmp_path)
|
|
assert report.added == [] and report.errors == {}
|
|
assert store.get_definition_by_source_path(session, "simple_rule.json") is None
|
|
|
|
|
|
async def test_sync_skips_unchanged(session, lfx_server, lfx_client, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
first = await sync.sync_from_directory(session, lfx_client, tmp_path)
|
|
assert first.added == ["simple_rule"]
|
|
second = await sync.sync_from_directory(session, lfx_client, 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, lfx_client, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
await sync.sync_from_directory(session, lfx_client, tmp_path)
|
|
definition = store.get_definition_by_source_path(session, "simple_rule.json")
|
|
old_lfx_id = lfx_id_for(definition.id, 1)
|
|
|
|
doc = load_flow_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_client, 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_client.flow_info(old_lfx_id)
|
|
|
|
|
|
async def test_sync_deletes_missing_file_with_cascade(session, lfx_server, lfx_client, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
await sync.sync_from_directory(session, lfx_client, 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_client, 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_client.flow_info(lfx_id)
|
|
|
|
|
|
async def test_sync_reports_invalid_json(session, lfx_server, lfx_client, tmp_path):
|
|
(tmp_path / "kaputt.json").write_text("{kein json")
|
|
report = await sync.sync_from_directory(session, lfx_client, 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, lfx_client, tmp_path):
|
|
with pytest.raises(FileNotFoundError):
|
|
await sync.sync_from_directory(session, lfx_client, tmp_path / "gibts-nicht")
|
|
|
|
|
|
async def test_sync_rename_updates_definition_name(session, lfx_server, lfx_client, tmp_path):
|
|
_seed_dir(tmp_path)
|
|
await sync.sync_from_directory(session, lfx_client, tmp_path)
|
|
doc = load_flow_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_client, 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"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("seed", "flow_name", "flow_id", "extra_flows", "extra_docs", "expected_file"),
|
|
[
|
|
(
|
|
None,
|
|
"Gehaltserhöhung",
|
|
"22222222-3333-4444-8555-666666666666",
|
|
[],
|
|
{},
|
|
"Gehaltserhöhung.json",
|
|
),
|
|
(
|
|
# Seeded from simple_rule.json: updates in place instead of
|
|
# duplicating as simple-rule.json (slug drift).
|
|
"simple_rule.json",
|
|
"simple_rule",
|
|
"44444444-5555-4666-8777-888888888888",
|
|
[],
|
|
{},
|
|
"simple_rule.json",
|
|
),
|
|
(
|
|
None,
|
|
"Gut",
|
|
"33333333-4444-4555-8666-777777777777",
|
|
[{"id": "kaputt-id", "name": "Kaputt"}],
|
|
{},
|
|
"Gut.json",
|
|
),
|
|
],
|
|
ids=["pull-new", "pull-reuse", "pull-error-continues"],
|
|
)
|
|
async def test_sync_endpoint_pull_matrix(
|
|
client,
|
|
lfx_server,
|
|
tmp_path,
|
|
monkeypatch,
|
|
login,
|
|
override_langflow,
|
|
seed,
|
|
flow_name,
|
|
flow_id,
|
|
extra_flows,
|
|
extra_docs,
|
|
expected_file,
|
|
):
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
if seed is not None:
|
|
_seed_dir(tmp_path, seed)
|
|
override_langflow(
|
|
flows=[{"id": flow_id, "name": flow_name}, *extra_flows],
|
|
docs={flow_id: langflow_export_doc(flow_name, flow_id), **extra_docs},
|
|
)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["pulled"] == [flow_name]
|
|
assert body["added"] == [flow_name]
|
|
assert (tmp_path / expected_file).is_file()
|
|
if flow_name == "Gut":
|
|
assert "kaputt-id" in body["errors"]
|
|
else:
|
|
assert body["errors"] == {}
|
|
assert not (tmp_path / "simple-rule.json").exists()
|
|
|
|
# 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"] == [flow_name]
|
|
assert body["skipped"] == [flow_name]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("list_status", "expected_status"),
|
|
[(0, 503), (500, 502)],
|
|
)
|
|
async def test_sync_endpoint_langflow_failures(
|
|
client,
|
|
lfx_server,
|
|
tmp_path,
|
|
monkeypatch,
|
|
login,
|
|
override_langflow,
|
|
list_status,
|
|
expected_status,
|
|
):
|
|
"""Listing failures map to 503/502 (parametrized)."""
|
|
await login()
|
|
monkeypatch.setattr(settings, "flows_dir", str(tmp_path))
|
|
override_langflow(flows=[], list_status=list_status)
|
|
response = await client.post("/api/flows/sync")
|
|
assert response.status_code == expected_status
|
|
|
|
|
|
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
|