114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
"""Tests for flow.core.yaml."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from flow.core.errors import ConfigError
|
|
from flow.core.yaml import (
|
|
list_yaml_files,
|
|
load_yaml_documents,
|
|
load_yaml_file,
|
|
load_yaml_source,
|
|
load_yaml_sources,
|
|
merge_yaml_values,
|
|
)
|
|
|
|
|
|
class TestLoadYamlFile:
|
|
def test_loads_mapping(self, tmp_path):
|
|
f = tmp_path / "a.yaml"
|
|
f.write_text("key: value\n")
|
|
assert load_yaml_file(f) == {"key": "value"}
|
|
|
|
def test_empty_file_returns_empty_dict(self, tmp_path):
|
|
f = tmp_path / "empty.yaml"
|
|
f.write_text("")
|
|
assert load_yaml_file(f) == {}
|
|
|
|
def test_non_mapping_raises(self, tmp_path):
|
|
f = tmp_path / "list.yaml"
|
|
f.write_text("- one\n- two\n")
|
|
with pytest.raises(ConfigError, match="mapping at root"):
|
|
load_yaml_file(f)
|
|
|
|
def test_invalid_yaml_raises(self, tmp_path):
|
|
f = tmp_path / "bad.yaml"
|
|
f.write_text(":\n :\n [invalid")
|
|
with pytest.raises(ConfigError, match="Invalid YAML"):
|
|
load_yaml_file(f)
|
|
|
|
|
|
class TestMergeYamlValues:
|
|
def test_dict_merge(self):
|
|
base = {"a": 1, "b": {"x": 10}}
|
|
overlay = {"b": {"y": 20}, "c": 3}
|
|
result = merge_yaml_values(base, overlay)
|
|
assert result == {"a": 1, "b": {"x": 10, "y": 20}, "c": 3}
|
|
|
|
def test_list_concat(self):
|
|
assert merge_yaml_values([1, 2], [3, 4]) == [1, 2, 3, 4]
|
|
|
|
def test_scalar_override(self):
|
|
assert merge_yaml_values("old", "new") == "new"
|
|
|
|
def test_overlay_wins_type_mismatch(self):
|
|
assert merge_yaml_values({"a": 1}, "scalar") == "scalar"
|
|
|
|
|
|
class TestListYamlFiles:
|
|
def test_lists_sorted(self, tmp_path):
|
|
(tmp_path / "b.yaml").write_text("b: 1\n")
|
|
(tmp_path / "a.yml").write_text("a: 1\n")
|
|
(tmp_path / "c.txt").write_text("ignored")
|
|
files = list_yaml_files(tmp_path)
|
|
assert [f.name for f in files] == ["a.yml", "b.yaml"]
|
|
|
|
def test_missing_dir_returns_empty(self, tmp_path):
|
|
assert list_yaml_files(tmp_path / "nope") == []
|
|
|
|
|
|
class TestLoadYamlSource:
|
|
def test_file(self, tmp_path):
|
|
f = tmp_path / "config.yaml"
|
|
f.write_text("key: val\n")
|
|
assert load_yaml_source(f) == {"key": "val"}
|
|
|
|
def test_directory_merges(self, tmp_path):
|
|
(tmp_path / "01.yaml").write_text("a: 1\n")
|
|
(tmp_path / "02.yaml").write_text("b: 2\n")
|
|
result = load_yaml_source(tmp_path)
|
|
assert result == {"a": 1, "b": 2}
|
|
|
|
def test_missing_returns_empty(self, tmp_path):
|
|
assert load_yaml_source(tmp_path / "gone") == {}
|
|
|
|
|
|
class TestLoadYamlDocuments:
|
|
def test_single_file(self, tmp_path):
|
|
f = tmp_path / "doc.yaml"
|
|
f.write_text("x: 1\n")
|
|
docs = load_yaml_documents(f)
|
|
assert docs == [{"x": 1}]
|
|
|
|
def test_directory(self, tmp_path):
|
|
(tmp_path / "a.yaml").write_text("a: 1\n")
|
|
(tmp_path / "b.yaml").write_text("b: 2\n")
|
|
docs = load_yaml_documents(tmp_path)
|
|
assert docs == [{"a": 1}, {"b": 2}]
|
|
|
|
def test_missing_returns_empty(self, tmp_path):
|
|
assert load_yaml_documents(tmp_path / "gone") == []
|
|
|
|
|
|
class TestLoadYamlSources:
|
|
def test_merges_multiple_paths(self, tmp_path):
|
|
d1 = tmp_path / "d1"
|
|
d2 = tmp_path / "d2"
|
|
d1.mkdir()
|
|
d2.mkdir()
|
|
(d1 / "a.yaml").write_text("a: 1\n")
|
|
(d2 / "b.yaml").write_text("b: 2\n")
|
|
result = load_yaml_sources(d1, d2)
|
|
assert result == {"a": 1, "b": 2}
|