diff --git a/tests/test_shadow_invariant.py b/tests/test_shadow_invariant.py new file mode 100644 index 0000000..33c8d41 --- /dev/null +++ b/tests/test_shadow_invariant.py @@ -0,0 +1,59 @@ +"""Static tripwire for the observation-only rule (AGENTS.md safety rule 1). + +The HA suite rejects service calls at runtime; this check also covers code paths +no test reaches. A failure here means actuator code was added: that needs the +separately approved control milestone, not an edit to this list. +""" +import ast +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] / 'custom_components' / 'heating_mpc' + +# Calling, registering or forwarding HA services, and entity write methods. +FORBIDDEN_ATTRIBUTES = { + 'services', 'async_call', 'async_register', 'call_service', + 'async_turn_on', 'async_turn_off', 'async_toggle', 'turn_on', 'turn_off', 'toggle', + 'async_set_temperature', 'async_set_hvac_mode', 'async_set_preset_mode', + 'set_temperature', 'set_hvac_mode', 'set_preset_mode', +} +FORBIDDEN_STRINGS = { + 'turn_on', 'turn_off', 'toggle', 'set_temperature', 'set_hvac_mode', + 'set_preset_mode', 'call_service', +} +READ_ONLY_PLATFORMS = {'SENSOR', 'BINARY_SENSOR'} + + +class ShadowOnlyInvariant(unittest.TestCase): + def test_no_actuator_code_in_integration(self): + problems = [] + for path in sorted(ROOT.glob('*.py')): + for node in ast.walk(ast.parse(path.read_text(), str(path))): + where = f'{path.name}:{getattr(node, "lineno", "?")}' + if isinstance(node, ast.Attribute) and node.attr in FORBIDDEN_ATTRIBUTES: + problems.append(f'{where} uses .{node.attr}') + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in FORBIDDEN_ATTRIBUTES: + problems.append(f'{where} defines {node.name}()') + elif isinstance(node, ast.Constant) and node.value in FORBIDDEN_STRINGS: + problems.append(f'{where} names service {node.value!r}') + elif (isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name) + and node.value.id == 'Platform' and node.attr not in READ_ONLY_PLATFORMS): + problems.append(f'{where} forwards writable platform {node.attr}') + self.assertEqual(problems, []) + + def test_tripwire_detects_a_service_call(self): + tree = ast.parse('await hass.services.async_call("switch", "turn_on", {})') + attrs = {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)} + strings = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant)} + self.assertTrue(attrs & FORBIDDEN_ATTRIBUTES) + self.assertTrue(strings & FORBIDDEN_STRINGS) + + def test_only_read_only_entity_platform_files(self): + platforms = {'sensor', 'binary_sensor', 'switch', 'climate', 'number', 'select', + 'button', 'water_heater', 'valve', 'fan', 'light'} + present = {p.stem for p in ROOT.glob('*.py')} & platforms + self.assertEqual(present, {'sensor', 'binary_sensor'}) + + +if __name__ == '__main__': + unittest.main()