Added values support to runtime #17

dtfabric/runtime/data_maps.py

@@ -1520,8 +1520,6 @@ def _CompositeMapByteStream(
       data_type_map = self._data_type_maps[attribute_index]
       member_definition = self._data_type_definition.members[attribute_index]
 
-      # TODO: determine how to apply a condition on a non-member data type
-      # definition.
       condition = getattr(member_definition, 'condition', None)
       if condition:
         namespace = dict(subcontext.values)
@@ -1556,6 +1554,12 @@ def _CompositeMapByteStream(
       except Exception as exception:
         raise errors.MappingError(exception)
 
+      supported_values = getattr(member_definition, 'values', None)
+      if supported_values and value not in supported_values:
+        raise errors.MappingError(
+            'Value: {0!s} not in supported values: {1:s}'.format(
+                value, ', '.join(supported_values)))
+
       byte_offset += subcontext.byte_size
       members_data_size += subcontext.byte_size
 
@@ -1691,10 +1695,22 @@ def _LinearMapByteStream(
 
     try:
       struct_tuple = self._operation.ReadFrom(byte_stream[byte_offset:])
-      values = [
-          self._data_type_maps[index].MapValue(value)
-          for index, value in enumerate(struct_tuple)]
-      mapped_value = self._structure_values_class(*values)
+      struct_values = []
+      for attribute_index, value in enumerate(struct_tuple):
+        data_type_map = self._data_type_maps[attribute_index]
+        member_definition = self._data_type_definition.members[attribute_index]
+
+        value = data_type_map.MapValue(value)
+
+        supported_values = getattr(member_definition, 'values', None)
+        if supported_values and value not in supported_values:
+          raise errors.MappingError(
+              'Value: {0!s} not in supported values: {1:s}'.format(
+                  value, ', '.join(supported_values)))
+
+        struct_values.append(value)
+
+      mapped_value = self._structure_values_class(*struct_values)
 
     except Exception as exception:
       error_string = (

test_data/structure_with_values.yaml

@@ -20,14 +20,18 @@ members:
   element_data_type: byte
   elements_data_size: 4
   value: "test"
-- name: data1
+- name: data_size
   data_type: uint32
+- name: data
+  type: stream
+  element_data_type: byte
+  elements_data_size: structure_with_value.data_size
 ---
 name: structure_with_values
 type: structure
 members:
 - name: format_version
   data_type: uint32
   values: [2, 3]
-- name: data1
+- name: data_size
   data_type: uint32

tests/reader.py

@@ -1272,10 +1272,10 @@ def testReadFileObjectStructureWithValues(self):
     self.assertIsInstance(data_type_definition, data_types.StructureDefinition)
     self.assertEqual(data_type_definition.name, 'structure_with_value')
 
-    self.assertEqual(len(data_type_definition.members), 2)
+    self.assertEqual(len(data_type_definition.members), 3)
 
     byte_size = data_type_definition.GetByteSize()
-    self.assertEqual(byte_size, 8)
+    self.assertIsNone(byte_size)
 
     data_type_definition = definitions_registry.GetDefinitionByName(
         'structure_with_values')

tests/runtime/data_maps.py

@@ -1315,6 +1315,48 @@ def testMapByteStreamWithSequenceWithCondition(self):
     self.assertIsNone(structure_with_condition.conditional_data2)
     self.assertEqual(structure_with_condition.data3, 4)
 
+  @test_lib.skipUnlessHasTestFile(['structure_with_values.yaml'])
+  def testMapByteStreamWithSequenceWithValues(self):
+    """Tests the MapByteStream function with a sequence with values."""
+    definitions_file = self._GetTestFilePath(['structure_with_values.yaml'])
+    definitions_registry = self._CreateDefinitionRegistryFromFile(
+        definitions_file)
+
+    data_type_definition = definitions_registry.GetDefinitionByName(
+        'structure_with_values')
+    data_type_map = data_maps.StructureMap(data_type_definition)
+
+    byte_values = [0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]
+    byte_stream = bytes(bytearray(byte_values))
+
+    structure_with_values = data_type_map.MapByteStream(byte_stream)
+    self.assertEqual(structure_with_values.format_version, 2)
+    self.assertEqual(structure_with_values.data_size, 0)
+
+    byte_values = [0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]
+    byte_stream = bytes(bytearray(byte_values))
+
+    with self.assertRaises(errors.MappingError):
+      data_type_map.MapByteStream(byte_stream)
+
+    data_type_definition = definitions_registry.GetDefinitionByName(
+        'structure_with_value')
+    data_type_map = data_maps.StructureMap(data_type_definition)
+
+    byte_values = [0x74, 0x65, 0x73, 0x74, 0x00, 0x00, 0x00, 0x00]
+    byte_stream = bytes(bytearray(byte_values))
+
+    structure_with_values = data_type_map.MapByteStream(byte_stream)
+    self.assertEqual(structure_with_values.signature, b'test')
+    self.assertEqual(structure_with_values.data_size, 0)
+    self.assertEqual(structure_with_values.data, b'')
+
+    byte_values = [0x54, 0x45, 0x53, 0x54, 0x00, 0x00, 0x00, 0x00]
+    byte_stream = bytes(bytearray(byte_values))
+
+    with self.assertRaises(errors.MappingError):
+      data_type_map.MapByteStream(byte_stream)
+
   @test_lib.skipUnlessHasTestFile(['structure.yaml'])
   def testMapByteStreamWithSequenceWithExpression(self):
     """Tests the MapByteStream function with a sequence with expression."""