Params Registry Checks

openpilot/common/tests/test_params_registry_usage.py

@@ -0,0 +1,76 @@
+import ast
+from pathlib import Path
+import re
+
+
+ROOT = Path(__file__).resolve().parents[3]
+KEY_METHODS = {
+  'get', 'get_bool', 'get_int', 'get_float', 'put', 'put_bool', 'put_int', 'put_float',
+  'put_nonblocking', 'put_bool_nonblocking', 'remove', 'get_default_value', 'check_key',
+  'get_type', 'get_param_path', 'cpp2python',
+}
+
+
+def literal_params_calls(source):
+  tree = ast.parse(source)
+  owners = set()
+  mappings = set()
+  for node in ast.walk(tree):
+    if not isinstance(node, (ast.Assign, ast.AnnAssign)):
+      continue
+    targets = node.targets if isinstance(node, ast.Assign) else [node.target]
+    value = node.value
+    if isinstance(value, ast.Call) and ast.unparse(value.func).split('.')[-1] == 'Params':
+      owners.update(ast.unparse(target) for target in targets)
+    if (isinstance(value, ast.Dict) or
+        isinstance(value, ast.Attribute) and value.attr == 'query_params'):
+      mappings.update(ast.unparse(target) for target in targets)
+  for node in ast.walk(tree):
+    if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute) or node.func.attr not in KEY_METHODS:
+      continue
+    receiver = ast.unparse(node.func.value)
+    direct = isinstance(node.func.value, ast.Call) and ast.unparse(node.func.value.func).split('.')[-1] == 'Params'
+    conventional = re.search(r'(?:^|\.)(?:params\w*|\w*_params)$', receiver) is not None
+    if not direct and (receiver not in owners and not conventional or receiver in mappings and receiver not in owners):
+      continue
+    key = node.args[0] if node.args else next((keyword.value for keyword in node.keywords if keyword.arg == 'key'), None)
+    if isinstance(key, ast.Constant) and isinstance(key.value, str) and key.value:
+      yield node.lineno, key.value
+
+
+def test_literal_runtime_params_keys_are_registered():
+  registered = set(re.findall(r'^\s*\{"([^"\n]+)",', (ROOT / 'openpilot/common/params_keys.h').read_text(), re.MULTILINE))
+  assert registered
+  missing = []
+  for base in (ROOT / 'openpilot', ROOT / 'opendbc_repo/opendbc'):
+    for path in sorted(base.rglob('*.py')):
+      if {'tests', 'test', '__pycache__'} & set(path.parts):
+        continue
+      for line, key in literal_params_calls(path.read_text()):
+        if key not in registered:
+          missing.append(f'{path.relative_to(ROOT)}:{line}: {key}')
+  assert not missing, '\n'.join(missing)
+
+
+def test_audit_recognizes_direct_conventional_constructor_and_keyword_calls():
+  source = '''
+params.get("Read")
+self.params.put_bool("Write", True)
+store = Params()
+store.get_default_value("Default")
+Params().get_bool(key="Direct")
+params.get(dynamic_key)
+raw_namespace.joinpath("LegacyOnly").read_bytes()
+'''
+  assert [key for _, key in literal_params_calls(source)] == ['Read', 'Write', 'Default', 'Direct']
+
+
+def test_audit_excludes_query_dicts_and_empty_namespace_path():
+  source = '''
+params = server.query_params
+params.get("provider")
+query_params = {"provider": "test"}
+query_params.get("provider")
+Params().get_param_path("")
+'''
+  assert list(literal_params_calls(source)) == []

openpilot/starpilot/system/android_auto/supervisor.py

@@ -69,7 +69,9 @@ STATE_LABELS = {
 def _is_onroad() -> bool:
   try:
     from openpilot.common.params import Params
-    return Params().get_bool("IsOnroad")
+    from openpilot.starpilot.saved_source import read_saved
+    raw, readable = read_saved(Params(), "IsOffroad", 1)
+    return readable and raw == b"0"
   except Exception:
     return False
 

openpilot/starpilot/system/android_auto/tests/test_onroad_state.py

@@ -0,0 +1,37 @@
+import sys
+from types import ModuleType
+from unittest.mock import Mock
+
+import pytest
+
+from openpilot.starpilot.system.android_auto.supervisor import _is_onroad
+
+
+@pytest.mark.parametrize('raw,expected', [(None, False), (b'0', True), (b'1', False), (b'2', False), (b'', False), (b'garbage', False)])
+def test_auto_connect_uses_exact_registered_offroad_state(tmp_path, monkeypatch, raw, expected):
+  path = tmp_path / 'IsOffroad'
+  if raw is not None:
+    path.write_bytes(raw)
+  params = Mock()
+  def param_path(key):
+    if key != 'IsOffroad':
+      raise KeyError(key)
+    return str(path)
+  params.get_param_path.side_effect = param_path
+  module = ModuleType('openpilot.common.params')
+  module.Params = Mock(return_value=params)
+  monkeypatch.setitem(sys.modules, module.__name__, module)
+  assert _is_onroad() is expected
+  params.get_param_path.assert_called_once_with('IsOffroad')
+  params.get_bool.assert_not_called()
+
+
+def test_unreadable_or_failed_params_cannot_trigger_auto_connect(tmp_path, monkeypatch):
+  module = ModuleType('openpilot.common.params')
+  module.Params = Mock(side_effect=RuntimeError('unavailable'))
+  monkeypatch.setitem(sys.modules, module.__name__, module)
+  assert _is_onroad() is False
+  path = tmp_path / 'IsOffroad'
+  path.mkdir()
+  module.Params = Mock(return_value=Mock(get_param_path=lambda key: str(path)))
+  assert _is_onroad() is False