"""hive.config_store — document shape, layer targeting, and the hash contract."""
import pytest
from hive.config_store import (
ConfigError,
content_hash,
effective_hash,
effective_layers,
validate_layers,
)
def test_validate_accepts_a_minimal_layer():
layers = validate_layers([{"name": "g", "config": {"settings": {"a": 1}}}])
assert layers == [{"name": "g", "config": {"settings": {"a": 1}}, "match": {}, "enabled": True}]
def test_validate_rejects_duplicate_names():
with pytest.raises(ConfigError, match="duplicate"):
validate_layers(
[
{"name": "g", "config": {}},
{"name": "g", "config": {}},
]
)
def test_validate_rejects_non_object_config():
with pytest.raises(ConfigError):
validate_layers([{"name": "g", "config": []}])
def test_validate_rejects_unknown_match_keys():
with pytest.raises(ConfigError, match="unknown match keys"):
validate_layers([{"name": "g", "config": {}, "match": {"hostnames": ["x"]}}])
def test_validate_rejects_non_bool_enabled():
with pytest.raises(ConfigError):
validate_layers([{"name": "g", "config": {}, "enabled": "yes"}])
def test_empty_match_lists_mean_global():
layers = validate_layers(
[{"name": "g", "config": {}, "match": {"tags": [], "names": []}}]
)
assert layers[0]["match"] == {}
assert effective_layers(layers) == [{"name": "g", "config": {}}]
def test_effective_layers_keep_document_order():
layers = validate_layers(
[
{"name": "yuki", "match": {"names": ["yuki"]}, "config": {"a": 2}},
{"name": "global", "config": {"a": 1}},
]
)
got = effective_layers(layers, name="yuki")
assert [layer["name"] for layer in got] == ["yuki", "global"]
def test_disabled_layer_is_skipped_but_kept_in_history():
layers = validate_layers(
[
{"name": "g", "config": {"a": 1}},
{"name": "off", "config": {"b": 2}, "enabled": False},
]
)
assert effective_layers(layers) == [{"name": "g", "config": {"a": 1}}]
assert len(layers) == 2
def test_instance_id_matching():
layers = validate_layers(
[{"name": "pinned", "match": {"instance_ids": ["abc"]}, "config": {"a": 1}}]
)
assert effective_layers(layers, instance_id="abc") == [{"name": "pinned", "config": {"a": 1}}]
assert effective_layers(layers, instance_id="other") == []
def test_hash_is_over_the_layers_the_instance_receives():
layers = validate_layers(
[
{"name": "global", "config": {"a": 1}},
{"name": "lab", "match": {"tags": ["lab"]}, "config": {"b": 2}},
]
)
plain = effective_layers(layers)
lab = effective_layers(layers, tags=["lab"])
assert effective_hash(layers) == content_hash(plain)
assert effective_hash(layers, tags=["lab"]) == content_hash(lab)
assert effective_hash(layers) != effective_hash(layers, tags=["lab"])
def test_hash_ignores_match_and_enabled_metadata():
# Targeting metadata is not part of what an instance receives: a machine
# that gets the same layer content hashes the same whether the layer was
# global or addressed to it. Otherwise every re-targeting publish would
# make unaffected machines re-ack identical content.
a = validate_layers([{"name": "g", "config": {"a": 1}}])
b = validate_layers(
[{"name": "g", "config": {"a": 1}, "match": {"tags": ["x"]}}]
)
assert effective_hash(a) == effective_hash(b, tags=["x"])
def test_hash_function_matches_the_instance_side():
"""The hive and the instance must agree byte for byte on 'same config'."""
from navi.overlay.util import content_hash as navi_hash
value = [{"name": "g", "config": {"settings": {"x": [1, 2], "y": {"z": "ю"}}}}]
assert content_hash(value) == navi_hash(value)