The HA suite rejects service calls at runtime, but only on exercised paths. Parse the integration source and fail on service calls or registration, actuator method names, service-name strings, or any platform other than sensor/binary_sensor (AGENTS.md safety rule 1). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
60 lines
2.9 KiB
Python
60 lines
2.9 KiB
Python
"""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()
|