diff --git a/server/src/ledgrab/api/routes/automations.py b/server/src/ledgrab/api/routes/automations.py index 7c3e3fb..eb630d0 100644 --- a/server/src/ledgrab/api/routes/automations.py +++ b/server/src/ledgrab/api/routes/automations.py @@ -23,6 +23,7 @@ from ledgrab.storage.automation import ( ApplicationRule, DisplayStateRule, HomeAssistantRule, + HTTPPollRule, MQTTRule, Rule, StartupRule, @@ -75,6 +76,11 @@ def _rule_from_schema(s: RuleSchema) -> Rule: state=s.state or "", match_mode=s.match_mode or "exact", ), + "http_poll": lambda: HTTPPollRule( + value_source_id=s.value_source_id or "", + operator=s.operator or "equals", + value=s.value or "", + ), } factory = _SCHEMA_TO_RULE.get(s.rule_type) if factory is None: diff --git a/server/src/ledgrab/api/schemas/automations.py b/server/src/ledgrab/api/schemas/automations.py index 75d8897..b840735 100644 --- a/server/src/ledgrab/api/schemas/automations.py +++ b/server/src/ledgrab/api/schemas/automations.py @@ -46,6 +46,24 @@ class RuleSchema(BaseModel): None, description="HA entity ID, e.g. 'binary_sensor.front_door' (for home_assistant rule)", ) + # HTTP poll rule fields + value_source_id: Optional[str] = Field( + None, + description=( + "Value source ID (for http_poll rule). The referenced " + "ValueSource must be of source_type='http'." + ), + ) + operator: Optional[str] = Field( + None, + description=( + "Comparison operator for http_poll rule: " + "'equals', 'not_equals', 'contains', 'regex', 'gt', 'lt', 'exists'." + ), + ) + value: Optional[str] = Field( + None, description="Expected value (for http_poll rule; ignored for 'exists')" + ) # Backward-compatible alias diff --git a/server/src/ledgrab/static/js/features/automations.ts b/server/src/ledgrab/static/js/features/automations.ts index 83b3330..869e50c 100644 --- a/server/src/ledgrab/static/js/features/automations.ts +++ b/server/src/ledgrab/static/js/features/automations.ts @@ -5,6 +5,7 @@ import { apiKey, _automationsLoading, set_automationsLoading, automationsCacheObj, scenePresetsCache, _cachedHASources, haSourcesCache, + _cachedValueSources, valueSourcesCache, getHAEntityFriendlyName, setHAEntityNames, } from '../core/state.ts'; import { prefetchHAEntities } from './home-assistant-sources.ts'; @@ -26,6 +27,7 @@ import { registerIconEntityType, makeSimpleIconAdapter } from './icon-picker.ts' import { getBaseOrigin } from './settings.ts'; import { IconSelect } from '../core/icon-select.ts'; import { EntitySelect } from '../core/entity-palette.ts'; +import { enhanceMiniSelects } from '../core/mini-select.ts'; import { attachProcessPicker } from '../core/process-picker.ts'; import { TreeNav } from '../core/tree-nav.ts'; import { csScenes, createSceneCard, initScenePresetDelegation } from './scene-presets.ts'; @@ -106,6 +108,12 @@ class AutomationEditorModal extends Modal { onForceClose() { if (_automationTagsInput) { _automationTagsInput.destroy(); _automationTagsInput = null; } + // Tear down any per-rule portal widgets (http_poll EntitySelect / + // IconSelect) the rule rows attached. Walks every rule row in the + // modal, no-op for rows that didn't stash widgets. + document + .querySelectorAll('#automation-rules-list .rule-fields-container') + .forEach(c => _disposeHTTPPollWidgets(c)); } snapshotValues() { @@ -226,6 +234,7 @@ export async function loadAutomations() { automationsCacheObj.fetch(), scenePresetsCache.fetch(), haSourcesCache.fetch(), + valueSourcesCache.fetch(), ]); const sceneMap = new Map(scenes.map(s => [s.id, s])); @@ -345,8 +354,67 @@ const RULE_CHIP_RENDERERS: Record = { title: tooltip, }; }, + http_poll: (c) => { + const vsId = c.value_source_id || ''; + const vs = (_cachedValueSources || []).find(v => v.id === vsId); + const vsLabel = vs?.name || (vsId ? vsId : t('automations.rule.http_poll.no_source')); + const op = c.operator || 'equals'; + const opGlyph = _httpOpGlyph(op); + const rhs = op === 'exists' ? '' : ` ${c.value ?? ''}`; + return { + icon: _icon(P.globe), + text: `${vsLabel} ${opGlyph}${rhs}`, + title: t('automations.rule.http_poll'), + }; + }, }; +type HTTPPollOp = 'equals' | 'not_equals' | 'contains' | 'regex' | 'gt' | 'lt' | 'exists'; + +const HTTP_OP_KEYS: HTTPPollOp[] = [ + 'equals', 'not_equals', 'contains', 'regex', 'gt', 'lt', 'exists', +]; + +const _HTTP_OP_GLYPHS: Record = { + equals: '=', + not_equals: '≠', + contains: '∈', + regex: '/.../', + gt: '>', + lt: '<', + exists: '?', +}; + +const _HTTP_OP_ICON_PATHS: Record = { + equals: P.check, + not_equals: P.circleOff, + contains: P.search, + regex: P.code, + gt: P.chevronUp, + lt: P.chevronDown, + exists: P.zap, +}; + +function _httpOpGlyph(op: string): string { + return _HTTP_OP_GLYPHS[op as HTTPPollOp] ?? '='; +} + +function _httpOpIconPath(op: string): string { + return _HTTP_OP_ICON_PATHS[op as HTTPPollOp] ?? P.check; +} + +/** Destroy any EntitySelect / IconSelect widgets that the http_poll rule + * branch attached to *container*. Safe to call when there are none — + * it just clears the stash. Called both before re-rendering the rule's + * fields (rule-type change) and before removing the rule row entirely. */ +function _disposeHTTPPollWidgets(container: HTMLElement): void { + const stash = (container as any)._httpPollWidgets; + if (!stash) return; + try { stash.vsEntitySelect?.destroy?.(); } catch { /* widget already gone */ } + try { stash.opIconSelect?.destroy?.(); } catch { /* widget already gone */ } + delete (container as any)._httpPollWidgets; +} + /** Render a chain-arrow separator span. `+` between AND-rules, * the localised OR label between OR-rules, and `→` for the * rule-chain → scene-activation transition. */ @@ -681,11 +749,12 @@ export function addAutomationRule() { _autoGenerateAutomationName(); } -const RULE_TYPE_KEYS = ['startup', 'application', 'time_of_day', 'system_idle', 'display_state', 'mqtt', 'webhook', 'home_assistant']; +const RULE_TYPE_KEYS = ['startup', 'application', 'time_of_day', 'system_idle', 'display_state', 'mqtt', 'webhook', 'home_assistant', 'http_poll']; const RULE_TYPE_ICONS = { startup: P.power, application: P.smartphone, time_of_day: P.clock, system_idle: P.moon, display_state: P.monitor, mqtt: P.radio, webhook: P.globe, home_assistant: P.home, + http_poll: P.globe, }; const MATCH_TYPE_KEYS = ['running', 'topmost', 'topmost_fullscreen', 'fullscreen']; @@ -777,7 +846,7 @@ function addAutomationRuleRow(rule: any) { - + `; @@ -794,14 +863,29 @@ function addAutomationRuleRow(rule: any) { const typeSelect = row.querySelector('.rule-type-select') as HTMLSelectElement; const container = row.querySelector('.rule-fields-container') as HTMLElement; + // Remove button — dispose any widgets the rule body stashed (portal + // overlays would otherwise leak) before pulling the row from the DOM. + const removeBtn = row.querySelector('.btn-remove-rule') as HTMLButtonElement; + removeBtn.addEventListener('click', () => { + _disposeHTTPPollWidgets(container); + row.remove(); + const autoGen = (window as any)._autoGenerateAutomationName; + if (typeof autoGen === 'function') autoGen(); + }); + // Attach IconSelect to the rule type dropdown const ruleIconSelect = new IconSelect({ target: typeSelect, items: _buildRuleTypeItems(), - columns: 4, + columns: 3, } as any); function renderFields(type: any, data: any) { + // Tear down any widgets the previous renderFields call attached + // (EntitySelect/IconSelect portal overlays to document.body, so + // a bare ``container.innerHTML = …`` leaves them in the registry). + _disposeHTTPPollWidgets(container); + if (type === 'startup') { container.innerHTML = `${t('automations.rule.startup.hint')}`; return; @@ -880,6 +964,7 @@ function addAutomationRuleRow(rule: any) { `; + enhanceMiniSelects(container, 'select.rule-display-state'); return; } if (type === 'mqtt') { @@ -905,6 +990,7 @@ function addAutomationRuleRow(rule: any) { `; + enhanceMiniSelects(container, 'select.rule-mqtt-match-mode'); return; } if (type === 'home_assistant') { @@ -987,6 +1073,87 @@ function addAutomationRuleRow(rule: any) { return; } + if (type === 'http_poll') { + const vsId = data.value_source_id || ''; + const operator = data.operator || 'equals'; + const valueStr = data.value || ''; + + container.innerHTML = ` +
+ ${t('automations.rule.http_poll.hint')} +
+ + +
+
+ + +
+
+ + +
+
`; + + // Pull only HTTP value sources (source_type === 'http') + const httpVs = (_cachedValueSources || []).filter((v: any) => v.source_type === 'http'); + + // Wire EntitySelect for the value source picker + const vsSelect = container.querySelector('.rule-http-value-source') as HTMLSelectElement; + // Pre-populate the option so EntitySelect can sync display text. + vsSelect.innerHTML = `` + + httpVs.map((v: any) => ``).join(''); + vsSelect.value = vsId || ''; + const vsEntitySelect = new EntitySelect({ + target: vsSelect, + getItems: () => (_cachedValueSources || []) + .filter((v: any) => v.source_type === 'http') + .map((v: any) => ({ + value: v.id, + label: v.name, + icon: _icon(P.globe), + desc: v.json_path || t('automations.rule.http_poll.raw_body'), + })), + placeholder: t('palette.search'), + }); + + // Wire IconSelect for operator + const opSelect = container.querySelector('.rule-http-operator') as HTMLSelectElement; + const opItems = HTTP_OP_KEYS.map(k => ({ + value: k, + icon: _icon(_httpOpIconPath(k)), + label: t('automations.rule.http_poll.operator.' + k), + desc: t('automations.rule.http_poll.operator.' + k + '.desc'), + })); + const opIconSelect = new IconSelect({ + target: opSelect, + items: opItems, + columns: 3, + onChange: (newOp: string) => { + // Hide the value field when operator is 'exists' + const valField = container.querySelector('.rule-http-value-field') as HTMLElement; + if (valField) valField.style.display = newOp === 'exists' ? 'none' : ''; + }, + }); + // Sync initial visibility based on the operator we just loaded. + const valField = container.querySelector('.rule-http-value-field') as HTMLElement; + if (valField) valField.style.display = operator === 'exists' ? 'none' : ''; + + // Stash both widgets so they can be destroyed when the row's + // rule type changes (renderFields re-entry) or the row is + // removed (button.btn-remove-rule onclick — calls + // _disposeHTTPPollWidgets via the row's data hook). + (container as any)._httpPollWidgets = { + vsEntitySelect, + opIconSelect, + }; + + return; + } if (type === 'webhook') { if (data.token) { const webhookUrl = getBaseOrigin() + '/api/v1/webhooks/' + data.token; @@ -1102,6 +1269,18 @@ function getAutomationEditorRules() { state: (row.querySelector('.rule-ha-state') as HTMLInputElement).value, match_mode: (row.querySelector('.rule-ha-match-mode') as HTMLSelectElement).value || 'exact', }); + } else if (ruleType === 'http_poll') { + const op = (row.querySelector('.rule-http-operator') as HTMLSelectElement).value || 'equals'; + const r: any = { + rule_type: 'http_poll', + value_source_id: (row.querySelector('.rule-http-value-source') as HTMLSelectElement).value, + operator: op, + }; + // The 'exists' operator has no comparison value. + if (op !== 'exists') { + r.value = (row.querySelector('.rule-http-value') as HTMLInputElement).value; + } + rules.push(r); } else { const matchType = (row.querySelector('.rule-match-type') as HTMLSelectElement).value; const appsText = (row.querySelector('.rule-apps') as HTMLTextAreaElement).value.trim(); @@ -1114,6 +1293,8 @@ function getAutomationEditorRules() { export async function saveAutomationEditor() { const idInput = document.getElementById('automation-editor-id') as HTMLInputElement; + if (automationModal.closeIfPristine(idInput.value)) return; + const nameInput = document.getElementById('automation-editor-name') as HTMLInputElement; const enabledInput = document.getElementById('automation-editor-enabled') as HTMLInputElement; const logicSelect = document.getElementById('automation-editor-logic') as HTMLSelectElement; diff --git a/server/src/ledgrab/storage/automation.py b/server/src/ledgrab/storage/automation.py index 24a0725..2f0fbf1 100644 --- a/server/src/ledgrab/storage/automation.py +++ b/server/src/ledgrab/storage/automation.py @@ -201,6 +201,49 @@ class HomeAssistantRule(Rule): ) +@dataclass +class HTTPPollRule(Rule): + """Activate based on a value extracted by an HTTP value source. + + The extraction (URL, auth, json_path, cadence) lives on an + ``HTTPValueSource``; this rule just references the value source and + compares its current raw value to ``value`` using ``operator``. + """ + + rule_type: str = "http_poll" + value_source_id: str = "" # references an HTTPValueSource + operator: str = "equals" # equals | not_equals | contains | regex | gt | lt | exists + value: str = "" + + def to_dict(self) -> dict: + d = super().to_dict() + d["value_source_id"] = self.value_source_id + d["operator"] = self.operator + d["value"] = self.value + return d + + @classmethod + def from_dict(cls, data: dict) -> "HTTPPollRule": + # Accept legacy ``http_source_id`` + ``json_path`` payloads from any + # in-flight v1 data and ignore them — the new shape only needs the + # value_source_id. Log a warning so a dev DB that still has the old + # form on disk is visible: such a row will load with + # ``value_source_id=""`` and the rule will silently evaluate to + # False forever otherwise. + if "http_source_id" in data or "json_path" in data: + logger.warning( + "Migrating legacy http_poll rule (had keys: %s). " + "value_source_id is empty; re-point the rule at an HTTPValueSource " + "and re-save to clear this warning.", + sorted(k for k in ("http_source_id", "json_path") if k in data), + ) + return cls( + value_source_id=data.get("value_source_id", ""), + operator=data.get("operator", "equals"), + value=data.get("value", ""), + ) + + _RULE_MAP: Dict[str, Type[Rule]] = { "application": ApplicationRule, "time_of_day": TimeOfDayRule, @@ -210,6 +253,7 @@ _RULE_MAP: Dict[str, Type[Rule]] = { "webhook": WebhookRule, "startup": StartupRule, "home_assistant": HomeAssistantRule, + "http_poll": HTTPPollRule, # Legacy: "always" maps to StartupRule for migration "always": StartupRule, } diff --git a/server/tests/core/test_automation_engine.py b/server/tests/core/test_automation_engine.py index da5bae3..7abbbfc 100644 --- a/server/tests/core/test_automation_engine.py +++ b/server/tests/core/test_automation_engine.py @@ -284,3 +284,284 @@ class TestEngineLifecycle: @pytest.mark.asyncio async def test_stop_without_start_is_safe(self, engine): await engine.stop() # no-op + + +# --------------------------------------------------------------------------- +# HTTP poll rule — operator helpers + integration with mocked value stream +# --------------------------------------------------------------------------- + + +class TestApplyOperator: + """Unit tests for the rule-comparison operator dispatch.""" + + def test_operator_equals(self): + from ledgrab.core.automations.automation_engine import _apply_operator + + assert _apply_operator("equals", "ok", "ok") is True + assert _apply_operator("equals", 42, "42") is True # str coercion + assert _apply_operator("equals", "ok", "fail") is False + + def test_operator_contains(self): + from ledgrab.core.automations.automation_engine import _apply_operator + + assert _apply_operator("contains", "hello world", "world") is True + assert _apply_operator("contains", "hello", "xyz") is False + + def test_operator_regex(self): + from ledgrab.core.automations.automation_engine import _apply_operator + + assert _apply_operator("regex", "abc123", r"\d+") is True + assert _apply_operator("regex", "abc", r"\d+") is False + assert _apply_operator("regex", "abc", "[") is False # invalid regex + + def test_operator_gt_lt(self): + from ledgrab.core.automations.automation_engine import _apply_operator + + assert _apply_operator("gt", 5, "3") is True + assert _apply_operator("gt", 2, "3") is False + assert _apply_operator("lt", 2, "3") is True + assert _apply_operator("gt", "not-a-number", "3") is False + + def test_operator_not_equals(self): + from ledgrab.core.automations.automation_engine import _apply_operator + + assert _apply_operator("not_equals", "a", "b") is True + assert _apply_operator("not_equals", "a", "a") is False + + +class TestHTTPPollRuleEvaluation: + """_evaluate_http_poll behavior with a mocked ValueStreamManager. + + Rule now points at an HTTPValueSource via ``value_source_id``; the + engine reads the cached raw value from the corresponding ValueStream + and compares with ``_apply_operator``. + """ + + def _engine_with_vsm(self, mock_store, mock_manager, vsm): + with patch("ledgrab.core.automations.automation_engine.PlatformDetector"): + return AutomationEngine( + automation_store=mock_store, + processor_manager=mock_manager, + poll_interval=0.1, + value_stream_manager=vsm, + ) + + def _vsm_with_stream(self, vs_id: str, raw_value): + """Build a mock ValueStreamManager whose ``peek(vs_id)`` returns a + stream whose ``get_raw_value()`` yields *raw_value*.""" + vsm = MagicMock() + stream = MagicMock() + stream.get_raw_value.return_value = raw_value + vsm.peek.side_effect = lambda q: stream if q == vs_id else None + return vsm + + def test_no_manager_returns_false(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + with patch("ledgrab.core.automations.automation_engine.PlatformDetector"): + eng = AutomationEngine( + automation_store=mock_store, + processor_manager=mock_manager, + poll_interval=0.1, + ) + rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok") + assert eng._evaluate_http_poll(rule) is False + + def test_no_value_source_id_returns_false(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = MagicMock() + vsm.peek.return_value = None + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="", operator="equals", value="ok") + assert eng._evaluate_http_poll(rule) is False + + def test_stream_missing_returns_false(self, mock_store, mock_manager): + """If the value stream hasn't been acquired yet, the rule is False.""" + from ledgrab.storage.automation import HTTPPollRule + + vsm = MagicMock() + vsm.peek.return_value = None # nothing acquired + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok") + assert eng._evaluate_http_poll(rule) is False + + def test_raw_value_none_returns_false(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = self._vsm_with_stream("vs_x", None) + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok") + assert eng._evaluate_http_poll(rule) is False + + def test_equals_match(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = self._vsm_with_stream("vs_x", "playing") + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="playing") + assert eng._evaluate_http_poll(rule) is True + + def test_exists_when_value_present(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = self._vsm_with_stream("vs_x", 1) + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="exists") + assert eng._evaluate_http_poll(rule) is True + + def test_exists_when_value_none(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = self._vsm_with_stream("vs_x", None) + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="exists") + assert eng._evaluate_http_poll(rule) is False + + def test_gt_numeric(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = self._vsm_with_stream("vs_x", 72.5) + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="gt", value="70") + assert eng._evaluate_http_poll(rule) is True + + def test_stream_without_raw_value_attr_returns_false(self, mock_store, mock_manager): + """A non-HTTP stream type that has no ``get_raw_value`` short-circuits.""" + from ledgrab.storage.automation import HTTPPollRule + + plain_stream = object() # no get_raw_value attribute + vsm = MagicMock() + vsm.peek.return_value = plain_stream + eng = self._engine_with_vsm(mock_store, mock_manager, vsm) + rule = HTTPPollRule(value_source_id="vs_x", operator="equals", value="ok") + assert eng._evaluate_http_poll(rule) is False + + +class TestSyncValueStreamRefs: + """Engine acquire/release diffing against the ValueStreamManager.""" + + def test_no_manager_is_noop(self, mock_store, mock_manager): + with patch("ledgrab.core.automations.automation_engine.PlatformDetector"): + eng = AutomationEngine( + automation_store=mock_store, + processor_manager=mock_manager, + poll_interval=0.1, + ) + eng._sync_value_stream_refs() # should not raise + + def test_acquires_referenced_sources(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = MagicMock() + + with patch("ledgrab.core.automations.automation_engine.PlatformDetector"): + eng = AutomationEngine( + automation_store=mock_store, + processor_manager=mock_manager, + poll_interval=0.1, + value_stream_manager=vsm, + ) + + mock_store.create_automation( + name="poll1", + enabled=True, + rule_logic="or", + rules=[ + HTTPPollRule( + value_source_id="vs_abc", + operator="equals", + value="ok", + ), + ], + scene_preset_id=None, + deactivation_mode="none", + deactivation_scene_preset_id=None, + ) + + eng._sync_value_stream_refs() + vsm.acquire.assert_called_once_with("vs_abc") + assert eng._value_sources_acquired == {"vs_abc"} + + def test_releases_unreferenced_sources(self, mock_store, mock_manager): + """When a rule is removed, the engine releases its formerly-held source.""" + vsm = MagicMock() + + with patch("ledgrab.core.automations.automation_engine.PlatformDetector"): + eng = AutomationEngine( + automation_store=mock_store, + processor_manager=mock_manager, + poll_interval=0.1, + value_stream_manager=vsm, + ) + + eng._value_sources_acquired = {"vs_old"} + eng._sync_value_stream_refs() # no rules reference it + vsm.release.assert_called_once_with("vs_old") + assert eng._value_sources_acquired == set() + + def test_disabled_automation_ignored(self, mock_store, mock_manager): + from ledgrab.storage.automation import HTTPPollRule + + vsm = MagicMock() + + with patch("ledgrab.core.automations.automation_engine.PlatformDetector"): + eng = AutomationEngine( + automation_store=mock_store, + processor_manager=mock_manager, + poll_interval=0.1, + value_stream_manager=vsm, + ) + + mock_store.create_automation( + name="off", + enabled=False, + rule_logic="or", + rules=[ + HTTPPollRule( + value_source_id="vs_xyz", + operator="equals", + value="", + ), + ], + scene_preset_id=None, + deactivation_mode="none", + deactivation_scene_preset_id=None, + ) + + eng._sync_value_stream_refs() + vsm.acquire.assert_not_called() + + +class TestHTTPValueStreamExtraction: + """Unit tests for the dot-path extractor used by HTTPValueStream.""" + + def test_empty_path_returns_body_text(self): + from ledgrab.core.processing.value_stream import _extract_simple_path + + assert _extract_simple_path(None, "", "raw body") == "raw body" + + def test_simple_key(self): + from ledgrab.core.processing.value_stream import _extract_simple_path + + assert _extract_simple_path({"status": "ok"}, "status", "") == "ok" + + def test_nested_with_array_index(self): + from ledgrab.core.processing.value_stream import _extract_simple_path + + body = {"MediaContainer": {"size": 2, "Metadata": [{"title": "Show"}]}} + assert _extract_simple_path(body, "MediaContainer.size", "") == 2 + assert _extract_simple_path(body, "MediaContainer.Metadata[0].title", "") == "Show" + + def test_chained_indices(self): + from ledgrab.core.processing.value_stream import _extract_simple_path + + body = {"matrix": [[1, 2, 3], [4, 5, 6]]} + assert _extract_simple_path(body, "matrix[0][1]", "") == 2 + assert _extract_simple_path(body, "matrix[1][2]", "") == 6 + + def test_missing_returns_none(self): + from ledgrab.core.processing.value_stream import _extract_simple_path + + assert _extract_simple_path({"a": 1}, "b", "") is None + assert _extract_simple_path({"a": [1]}, "a[5]", "") is None