struct generation

selfdrive/car/tests/test_car_interfaces.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 import math
 import unittest
-import hypothesis.strategies as st
 from hypothesis import given, settings
 import importlib
 from parameterized import parameterized
@@ -10,36 +9,13 @@ from cereal import car
 from selfdrive.car import gen_empty_fingerprint
 from selfdrive.car.car_helpers import interfaces
 from selfdrive.car.fingerprints import _FINGERPRINTS as FINGERPRINTS, all_known_cars
-
-def get_random_car_control():
-  def actuators():
-    return st.fixed_dictionaries({
-      'gas': st.floats(min_value=0.0, max_value=1.0, width=32),
-      'brake': st.floats(min_value=0.0, max_value=1.0, width=32),
-      'steer': st.floats(min_value=-1.0, max_value=1.0, width=32),
-      'steerOutputCan': st.floats(width=32),
-      'steeringAngleDeg': st.floats(width=32),
-      'curvature' : st.floats(width=32),
-      'speed' : st.floats(width=32),
-      'accel' : st.floats(width=32)
-      })
-
-  return st.fixed_dictionaries({
-    'enabled': st.booleans(),
-    'latActive': st.booleans(),
-    'longActive': st.booleans(),
-    'actuators': actuators(),
-    'actuatorsOutput': actuators(),
-    'orientationNED': st.lists(st.floats(width=32)),
-    'angularVelocity': st.lists(st.floats(width=32)),
-    'cruiseControl': st.fixed_dictionaries({'cancel': st.booleans(), 'resume': st.booleans(), 'override': st.booleans()})
-    })
+from selfdrive.test.process_replay.fuzzy_generation import get_random_msg
 
 class TestCarInterfaces(unittest.TestCase):
 
   @parameterized.expand([(car,) for car in all_known_cars()])
   @settings(max_examples=5)
-  @given(cc_msg=get_random_car_control())
+  @given(cc_msg=get_random_msg(car.CarControl))
   def test_car_interfaces(self, car_name, cc_msg):
     if car_name in FINGERPRINTS:
       fingerprint = FINGERPRINTS[car_name][0]

selfdrive/test/process_replay/fuzzy_generation.py

@@ -0,0 +1,60 @@
+import hypothesis.strategies as st
+import random
+
+def generate_native_type(field):
+  if field == 'bool':
+    return st.booleans()
+  elif field == 'int8':
+    return st.integers(min_value=-2**7, max_value=2**7-1)
+  elif field == 'int16':
+    return st.integers(min_value=-2**15, max_value=2**15-1)
+  elif field == 'int32':
+    return st.integers(min_value=-2**31, max_value=2**31-1)
+  elif field == 'int64':
+    return st.integers(min_value=-2**63, max_value=2**63-1)
+  elif field == 'uint8':
+    return st.integers(min_value=0, max_value=2**8-1)
+  elif field == 'uint16':
+    return st.integers(min_value=0, max_value=2**15-1)
+  elif field == 'uint32':
+    return st.integers(min_value=0, max_value=2**31-1)
+  elif field == 'uint64':
+    return st.integers(min_value=0, max_value=2**63-1)
+  elif field == 'float32':
+    return st.floats(width=32)
+  elif field == 'float64':
+    return st.floats(width=64)
+  elif field == 'text':
+    return st.text(max_size=1000)
+  elif field == 'data':
+    return st.text(max_size=1000)
+  elif field == 'anyPointer':
+    return st.text()
+  else:
+    raise NotImplementedError(f'Invalid type : {field}')
+
+def generate_field(field):
+  def rec(field_type):
+    if field_type.which() == 'struct':
+      return generate_struct(field.schema.elementType if base_type == 'list' else field.schema)
+    elif field_type.which() == 'list':
+      return st.lists(rec(field_type.list.elementType))
+    elif field_type.which() == 'enum':
+      schema = field.schema.elementType if base_type == 'list' else field.schema
+      return st.sampled_from(list(schema.enumerants.keys()))
+    else:
+      return generate_native_type(field_type.which())
+
+  if 'slot' in field.proto.to_dict():
+    base_type = field.proto.slot.type.which()
+    return rec(field.proto.slot.type)
+  else:
+    return generate_struct(field.schema)
+
+def generate_struct(schema):
+  full_fill = list(schema.non_union_fields) if schema.non_union_fields else []
+  single_fill = [random.choice(schema.union_fields)] if schema.union_fields else []
+  return st.fixed_dictionaries(dict((field, generate_field(schema.fields[field])) for field in full_fill + single_fill))
+
+def get_random_msg(struct):
+  return generate_struct(struct.schema)