"""Unit tests for canonical flow files (export.py). No DB or LFX needed.""" import json from app.flows import export def test_filename_stem_keeps_original_names(): assert export.filename_stem("simple_rule") == "simple_rule" assert export.filename_stem("test_flow") == "test_flow" assert export.filename_stem("Gehaltserhöhung") == "Gehaltserhöhung" assert export.filename_stem("Dachsanierung 200k") == "Dachsanierung 200k" assert export.filename_stem(" A/B\\C ") == "A-B-C" assert export.filename_stem('a:b*c?d"eg|h') == "a-b-c-d-e-f-g-h" assert export.filename_stem("trailing...") == "trailing" assert export.filename_stem("") == "flow" def test_canonical_bytes_stable_under_key_reorder(): doc = {"name": "x", "data": {"nodes": [], "edges": []}, "tags": ["b", "a"]} shuffled = {"tags": ["b", "a"], "data": {"edges": [], "nodes": []}, "name": "x"} assert export.canonical_bytes(shuffled) == export.canonical_bytes(doc) assert export.content_hash(doc) == export.content_hash(shuffled) def test_write_and_read_roundtrip_is_byte_stable(tmp_path): doc = {"name": "My Flow", "data": {"nodes": [], "edges": []}} path = export.write_flow_file(tmp_path, doc) assert path == tmp_path / "My Flow.json" assert json.loads(path.read_text()) == doc # Rewriting identical content is byte-identical (no phantom git diffs). before = path.read_bytes() export.write_flow_file(tmp_path, doc) assert path.read_bytes() == before def test_iter_flow_files_only_top_level_json(tmp_path): (tmp_path / "a.json").write_text("{}") (tmp_path / "b.txt").write_text("{}") sub = tmp_path / "sub" sub.mkdir() (sub / "c.json").write_text("{}") assert export.iter_flow_files(tmp_path) == [tmp_path / "a.json"] def test_find_reuse_path_matches_identical_content_first(tmp_path): doc = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}} other = tmp_path / "other-name.json" other.write_text(export.canonical_bytes(doc).decode()) path, identical = export.find_reuse_path(tmp_path, dict(doc)) assert (path, identical) == (other, True) def test_find_reuse_path_matches_same_name(tmp_path): old = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}, "extra": 1} (tmp_path / "simple_rule.json").write_text(json.dumps(old)) new = {"name": "Simple Rule", "data": {"nodes": [], "edges": []}} path, identical = export.find_reuse_path(tmp_path, new) assert (path, identical) == (tmp_path / "simple_rule.json", False) def test_find_reuse_path_no_match_and_missing_dir(tmp_path): (tmp_path / "a.json").write_text(json.dumps({"name": "Other"})) assert export.find_reuse_path(tmp_path, {"name": "Neu"}) == (None, False) assert export.find_reuse_path(tmp_path / "gibts-nicht", {"name": "Neu"}) == (None, False) def test_find_reuse_path_skips_unreadable_files(tmp_path): (tmp_path / "kaputt.json").write_text("{kein json") assert export.find_reuse_path(tmp_path, {"name": "Neu"}) == (None, False) def test_write_flow_file_is_world_readable(tmp_path): import os import stat doc = {"name": "Lesbar", "data": {"nodes": [], "edges": []}} path = export.write_flow_file(tmp_path, doc) mode = stat.S_IMODE(os.stat(path).st_mode) # Read bits for user/group/other (mkstemp defaults to 0o600; on Windows # chmod maps to 0o666, so only assert the read bits, not the exact mode). assert mode & 0o444 == 0o444, f"expected world-readable, got {oct(mode)}"