181 lines
6.4 KiB
Python
181 lines
6.4 KiB
Python
"""The rename mapper, and the registries that keep it complete."""
|
|
|
|
import ast
|
|
import dataclasses
|
|
import inspect
|
|
import re
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow import namespace, pipeline, state
|
|
from fluksio.flow import queue as queue_module
|
|
from fluksio.flow.alerts import dedup_key
|
|
from fluksio.flow.namespace import FAMILIES, ITEM_FIELDS, Rename, touches
|
|
from fluksio.flow.nodes import Node, base
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.flow.queue import RedisWorkQueue, WorkItem, done_marker
|
|
from fluksio.flow.state import MemoryState, RedisState
|
|
|
|
HEAT = Rename("heat", "warm")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("key", "moved"),
|
|
[
|
|
("heat.temp", "warm.temp"),
|
|
("heat.room.temp", "warm.room.temp"),
|
|
("heating.temp", None),
|
|
("__version__:heat.temp", "__version__:warm.temp"),
|
|
("__version__:heating.temp", None),
|
|
("__history__:heat.room.temp", "__history__:warm.room.temp"),
|
|
("__flush__:heat.valve", "__flush__:warm.valve"),
|
|
("__last_seen__:heat.valve:hall.temp", "__last_seen__:warm.valve:hall.temp"),
|
|
("__last_seen__:hall.lamp:heat.temp", "__last_seen__:hall.lamp:warm.temp"),
|
|
("__last_seen__:heat.valve:heat.temp", "__last_seen__:warm.valve:warm.temp"),
|
|
("__in_ts__:heating.valve:heat.temp", "__in_ts__:heating.valve:warm.temp"),
|
|
("__in_ts__:heating.valve:heating.temp", None),
|
|
# A node's own key is its data: only the node part moves.
|
|
("__node__:heat.limit:wait:heat.temp", "__node__:warm.limit:wait:heat.temp"),
|
|
("__node__:hall.limit:wait:heat.temp", None),
|
|
("__names__", None),
|
|
# Already under the new key.
|
|
("warm.temp", None),
|
|
("__last_seen__:warm.valve:warm.temp", None),
|
|
],
|
|
)
|
|
def test_state_keys_move_by_the_names_they_embed(key, moved):
|
|
assert HEAT.state_key(key) == moved
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"key",
|
|
["__bogus__:heat.temp", "__version__", "__names__:heat.temp", "__in_ts__:heat.v"],
|
|
)
|
|
def test_a_key_of_no_registered_family_is_refused(key):
|
|
with pytest.raises(ValueError, match="no registered family"):
|
|
HEAT.state_key(key)
|
|
|
|
|
|
def test_every_builder_maps_onto_itself_under_the_new_key():
|
|
"""The families agree with the builders that write them."""
|
|
pipe = Pipeline(nodes=[], state=MemoryState())
|
|
redis_state = RedisState(namespace="pipeline")
|
|
|
|
def node_key(flow, key):
|
|
node = Node(f=lambda params: None, name="limit")
|
|
node.assign_flow(flow, "limit")
|
|
return node._state_key(key)
|
|
|
|
builders = [
|
|
lambda f: pipe._version_key(f"{f}.temp"),
|
|
lambda f: pipe._timestamp_key(f"{f}.temp"),
|
|
lambda f: pipe._held_key(f"{f}.temp"),
|
|
lambda f: pipe._flush_key(f"{f}.valve"),
|
|
lambda f: pipe._last_seen_key(f"{f}.valve", f"{f}.temp"),
|
|
lambda f: pipe._delivered_key("hall.lamp", f"{f}.temp"),
|
|
lambda f: node_key(f, "wait:heat.temp"),
|
|
lambda f: redis_state._history_key(f"{f}.temp").removeprefix("pipeline:"),
|
|
]
|
|
|
|
for build in builders:
|
|
assert HEAT.state_key(build("heat")) == build("warm")
|
|
redis_state.close()
|
|
|
|
|
|
def test_touches_reads_the_same_parts():
|
|
assert touches("heat", "__last_seen__:hall.lamp:heat.temp")
|
|
assert not touches("heat", "__node__:hall.limit:wait:heat.temp")
|
|
assert not touches("heat", "heating.temp")
|
|
assert not touches("heat", "__names__")
|
|
|
|
|
|
def test_work_items_move_their_node_flow_and_output_names():
|
|
own = WorkItem(
|
|
kind="flush", node="heat.valve", flow="heat", guard_key="wait:heat.temp"
|
|
)
|
|
reaching = WorkItem(
|
|
kind="cascade",
|
|
node="hall.lamp",
|
|
flow="hall",
|
|
outputs={"heat.target": 21, "hall.on": True},
|
|
)
|
|
|
|
assert HEAT.item(own) == dataclasses.replace(own, node="warm.valve", flow="warm")
|
|
assert HEAT.item(reaching) == dataclasses.replace(
|
|
reaching, outputs={"warm.target": 21, "hall.on": True}
|
|
)
|
|
for item in (own, reaching):
|
|
assert HEAT.fields(item.to_fields()) == HEAT.item(item).to_fields()
|
|
|
|
|
|
def test_work_items_without_the_old_names_stay():
|
|
# A run item names its flow by id, and the first ids were the keys.
|
|
run = WorkItem(kind="run", node="", flow="heat", run_id="r1")
|
|
other = WorkItem(kind="cascade", node="heating.valve", flow="heating")
|
|
fields = {**other.to_fields(), "not_before": "0"}
|
|
|
|
assert HEAT.item(run) is None
|
|
assert HEAT.fields(run.to_fields()) is None
|
|
assert HEAT.item(other) is None
|
|
assert HEAT.fields(fields) is None
|
|
|
|
|
|
def test_alert_dedup_keys_follow_the_node_or_flow():
|
|
def key(**event):
|
|
return HEAT.dedup_key(dedup_key(event))
|
|
|
|
assert key(type="node_error", node="heat.valve", flow="heat") == (
|
|
"node_error:warm.valve"
|
|
)
|
|
assert key(type="flow_quarantined", flow="heat") == "flow_quarantined:warm"
|
|
assert key(type="flow_quarantined", flow="heating") == "flow_quarantined:heating"
|
|
assert key(type="engine_degraded") == "engine_degraded:"
|
|
|
|
|
|
def test_done_markers_read_back():
|
|
queue = SimpleNamespace(_ns="queue")
|
|
for entry_id in ("1700000000000-0", "due:1700000000000-0"):
|
|
key = RedisWorkQueue._done_key(queue, entry_id, "heat.valve")
|
|
assert done_marker(key) == (entry_id, "heat.valve")
|
|
|
|
|
|
FAMILY_LITERAL = re.compile(r"(__[a-z][a-z_]*__)(?::|$)")
|
|
#: What Python itself names this way, such as ``getattr(f, "__name__")``.
|
|
PYTHON = set(dir(type)) | set(dir(lambda: None))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("module", "registry"),
|
|
[
|
|
(pipeline, FAMILIES),
|
|
(base, FAMILIES),
|
|
(state, FAMILIES),
|
|
(queue_module, queue_module.STRUCTURES),
|
|
],
|
|
)
|
|
def test_every_family_in_the_key_builders_is_registered(module, registry):
|
|
"""A key family written as a literal here, not registered in `namespace`
|
|
(the queue's in `queue.STRUCTURES`), would be left behind by a rename."""
|
|
found = {
|
|
family
|
|
for node in ast.walk(ast.parse(inspect.getsource(module)))
|
|
if isinstance(node, ast.Constant) and isinstance(node.value, str)
|
|
for family in FAMILY_LITERAL.findall(node.value)
|
|
} - PYTHON
|
|
|
|
assert found <= set(registry), found - set(registry)
|
|
|
|
|
|
def test_every_work_item_field_is_classified():
|
|
fields = {field.name for field in dataclasses.fields(WorkItem)}
|
|
|
|
assert fields == set(ITEM_FIELDS)
|
|
assert set(WorkItem(kind="cascade", node="a.b", flow="a").to_fields()) <= fields
|
|
assert set(ITEM_FIELDS.values()) <= {
|
|
None,
|
|
namespace.NAME,
|
|
namespace.FLOW,
|
|
namespace.OUTPUTS,
|
|
}
|