ElementCallable.validate() - Code Metrics - Inspection of "Elem none" - sprout42/StarStruct - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#42)

by TJ

created 2016-12-20 21:09 UTC

ElementCallable.validate() C

↳ Parent: ElementCallable

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	9.8645

Importance

Changes	3
Bugs	1	Features	0

Metric	Value
cc	8
c	3
b	1
f	0
dl	0
loc	20
ccs	9
cts	13
cp	0.6923
crap	9.8645
rs	6.6666

"""
Element callable.

Call a function to validate data.

TODO: Update the format here

.. code-block:: python

    from binascii import crc32

    ExampleMessage = Message('VarTest', [('x', 'B'), ('y', 'B')])

    CRCedMessage = Message('CRCedMessage', [
       ('data', ExampleMessage),                    # A data field that has the example message in it
       ('crc', 'I', crc32, ['data']),               # Crc the data, and give an error if we have something unexpected
       ('crc', 'I', crc32, ['data'], False),        # Crc the data, but don't give an error
   ])

Following creating this message, you have two options:

1. Specify a value. The function will be used to validate the value.

.. code-block:: python

    def adder(x, y):
        return x + y

    AdderMessage = Message('AdderMessage', [
        ('item_a', 'H'),
        ('item_b', 'B'),
        ('function_data', 'I', adder, ['item_a', 'item_b']),
    ])

    test_data = {
        'item_a': 2,
        'item_b': 5,
        'function_data': 7,
    }

    made = AdderMessage.make(test_data)
    assert made.item_a == 2
    assert made.item_b == 5
    assert made.function_data == 7

    # If you specify the wrong value, you'll get a ValueError
    test_data = {
        'item_a': 2,
        'item_b': 5,
        'function_data': 33,
    }

    try:
        made = AdderMessage.make(test_data)
    except ValueError:
        print('Told you so')

    # Unless you specify `False` in your original item, then
    # nobody will care.

2. Use the function to generate a value.

.. code-block:: python

    def adder(x, y):
        return x + y

    AdderMessage = Message('AdderMessage', [
        ('item_a', 'H'),
        ('item_b', 'B'),
        ('function_data', 'I', adder, ['item_a', 'item_b']),
    ])

    test_data = {
        'item_a': 2,
        'item_b': 5,
    }

    made = AdderMessage.make(test_data)
    assert made.item_a == 2
    assert made.item_b == 5
    assert made.function_data == 7

"""

import copy
import struct

from typing import Optional
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from starstruct.element import register, Element
from starstruct.modes import Mode


# pylint: disable=too-many-instance-attributes
@register
class ElementCallable(Element):
    """
    Initialize a StarStruct element object.

    :param field: The fields passed into the constructor of the element
    :param mode: The mode in which to pack the bytes
    :param alignment: Number of bytes to align to
    """
    accepted_mesages = (True, False)

    def __init__(self, field: list, mode: Optional[Mode]=Mode.Native, alignment: Optional[int]=1):
        # All of the type checks have already been performed by the class
        # factory
        self.name = field[0]

        # Callable elements use the normal struct packing method
        self.format = field[1]

        if isinstance(field[2], dict):
            self.ref = field[2]

            default_list = [None, None]
            self._make_func = self.ref.get('make', default_list)[0]
            self._make_args = self.ref.get('make', default_list)[1:]

            self._pack_func = self.ref.get('pack', default_list)[0]
            self._pack_args = self.ref.get('pack', default_list)[1:]

            self._unpack_func = self.ref.get('unpack', default_list)[0]
            self._unpack_args = self.ref.get('unpack', default_list)[1:]
        elif isinstance(field[2], set):
            instruction = field[2].copy().pop()
            self.ref = {'all': instruction}

            self._make_func = self._pack_func = self._unpack_func = instruction[0]
            self._make_args = self._pack_args = self._unpack_args = instruction[1:]

        if len(field) == 4:
            self._error_on_bad_result = field[3]
        else:
            self._error_on_bad_result = True

        self._elements = []

        self.update(mode, alignment)

    @property
    def _struct(self):
        return struct.Struct(self._mode.value + self.format)

    @staticmethod
    def valid(field: tuple) -> bool:
        """
        See :py:func:`starstruct.element.Element.valid`

        :param field: The items to determine the structure of the element
        """
        required_keys = {'pack', 'unpack', 'make'}

        if len(field) >= 3 and isinstance(field[0], str) and isinstance(field[1], str):
            if isinstance(field[2], dict):
                if set(field[2].keys()) <= required_keys and \
                        all(isinstance(val, tuple) for val in field[2].values()):
                    return True
            elif isinstance(field[2], set):
                return len(field[2]) == 1 and \
                    all(isinstance(val, tuple) for val in field[2])

        return False

    def validate(self, msg):
        """
        Ensure that the supplied message contains the required information for
        this element object to operate.

        All elements that are Variable must reference valid Length elements.
        """
        for action in self.ref.values():
            for arg in action[1:]:
                if arg in ElementCallable.accepted_mesages:
                    continue
                elif isinstance(arg, str):
                    pass
                elif hasattr(arg, 'decode'):
                    arg = arg.decode('utf-8')
                elif hasattr(arg, 'to_bytes'):
                    arg = arg.to_bytes((arg.bit_length() + 7) // 8, self._mode.to_byteorder()).decode('utf-8')

                if arg not in msg:
                    raise ValueError('Need all keys to be in the message, {0} was not found\nAction: {1} -> {2}'.format(arg, action, action[1:]))

    def update(self, mode=None, alignment=None):
        """change the mode of the struct format"""
        if mode:
            self._mode = mode

        if alignment:
            self._alignment = alignment

    def pack(self, msg):
        """Pack the provided values into the supplied buffer."""
        pack_values = self.call_func(msg, self._pack_func, self._pack_args)

        # Test if the object is iterable
        # If it isn't, then turn it into a list
        try:
            _ = (p for p in pack_values)
        except TypeError:
            pack_values = [pack_values]

        # Unpack the items for struct to allow for mutli-value
        # items to be passed in.
        return self._struct.pack(*pack_values)


    def unpack(self, msg, buf):
        """Unpack data from the supplied buffer using the initialized format."""
        ret = self._struct.unpack_from(buf)
        if isinstance(ret, (list, tuple)) and len(ret) == 1:
            # We only change it not to a list if we expected one value.
            # Otherwise, we keep it as a list, because that's what we would
            # expect (like for a 16I type of struct
            ret = ret[0]

        # Only check for errors if they haven't told us not to
        if self._error_on_bad_result:
            # Pretend we're getting a dictionary to make our item,
            # but it has no reference to `self`. This is so we check
            # for errors correctly.
            temp_dict = copy.deepcopy(msg._asdict())
            temp_dict.pop(self.name)
            expected_value = self.call_func(msg, self._unpack_func, self._unpack_args)

            # Check for an error
            if expected_value != ret:
                raise ValueError('Expected value was: {0}, but got: {1}'.format(
                    expected_value,
                    ret,
                ))

        if self.name in self._unpack_args:
            msg = msg._replace(**{self.name: ret})


        ret = self.call_func(msg, self._unpack_func, self._unpack_args, original=ret)

        return (ret, buf[self._struct.size:])

    def make(self, msg):
        """Return the expected "made" value"""
        # If we aren't going to error on a bad result
        # and our name is in the message, just send the value
        # No need to do extra work.
        if not self._error_on_bad_result \
                and self.name in msg \
                and msg[self.name] is not None:
            return msg[self.name]

        ret = self.call_func(msg, self._make_func, self._make_args)

        if self.name in msg:
            if ret != msg[self.name]:
                raise ValueError('Excepted value: {0}, but got: {1}'.format(ret, msg[self.name]))

        return ret

    def call_func(self, msg, func, args, original=None):
        if func is None:
            return original

        items = self.prepare_args(msg, args)
        return func(*items)


    def prepare_args(self, msg, args):
        items = []

        if hasattr(msg, '_asdict'):
            msg = msg._asdict()

        for reference in args:
            if isinstance(reference, str):
                index = reference
                attr = 'make'
            elif isinstance(reference, bytes):
                index = reference.decode('utf-8')
                attr = 'pack'
            else:
                raise ValueError('Needed str or bytes for the reference')

            items.append(
                getattr(self._elements[index], attr)(msg)
            )

        return items


Pull Request — master (#42)

ElementCallable.validate() C

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing init.py files

1		"""
2		Element callable.
3
4		Call a function to validate data.
5
6		TODO: Update the format here
7
8		.. code-block:: python
9
10		from binascii import crc32
11
12		ExampleMessage = Message('VarTest', [('x', 'B'), ('y', 'B')])
13
14		CRCedMessage = Message('CRCedMessage', [
15		('data', ExampleMessage), # A data field that has the example message in it
16		('crc', 'I', crc32, ['data']), # Crc the data, and give an error if we have something unexpected
17		('crc', 'I', crc32, ['data'], False), # Crc the data, but don't give an error
18		])
19
20		Following creating this message, you have two options:
21
22		1. Specify a value. The function will be used to validate the value.
23
24		.. code-block:: python
25
26		def adder(x, y):
27		return x + y
28
29		AdderMessage = Message('AdderMessage', [
30		('item_a', 'H'),
31		('item_b', 'B'),
32		('function_data', 'I', adder, ['item_a', 'item_b']),
33		])
34
35		test_data = {
36		'item_a': 2,
37		'item_b': 5,
38		'function_data': 7,
39		}
40
41		made = AdderMessage.make(test_data)
42		assert made.item_a == 2
43		assert made.item_b == 5
44		assert made.function_data == 7
45
46		# If you specify the wrong value, you'll get a ValueError
47		test_data = {
48		'item_a': 2,
49		'item_b': 5,
50		'function_data': 33,
51		}
52
53		try:
54		made = AdderMessage.make(test_data)
55		except ValueError:
56		print('Told you so')
57
58		# Unless you specify `False` in your original item, then
59		# nobody will care.
60
61		2. Use the function to generate a value.
62
63		.. code-block:: python
64
65		def adder(x, y):
66		return x + y
67
68		AdderMessage = Message('AdderMessage', [
69		('item_a', 'H'),
70		('item_b', 'B'),
71		('function_data', 'I', adder, ['item_a', 'item_b']),
72		])
73
74		test_data = {
75		'item_a': 2,
76		'item_b': 5,
77		}
78
79		made = AdderMessage.make(test_data)
80		assert made.item_a == 2
81		assert made.item_b == 5
82		assert made.function_data == 7
83
84		"""
85
86	1	import copy
87	1	import struct
88
89	1	from typing import Optional
		0 ignored issues – show Configuration introduced 2016-11-30 19:25 UTC by Report Bug Copy Issue Report The import `typing` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
90
91	1	from starstruct.element import register, Element
92	1	from starstruct.modes import Mode
93
94
95		# pylint: disable=too-many-instance-attributes
96	1	@register
97	1	class ElementCallable(Element):
98		"""
99		Initialize a StarStruct element object.
100
101		:param field: The fields passed into the constructor of the element
102		:param mode: The mode in which to pack the bytes
103		:param alignment: Number of bytes to align to
104		"""
105	1	accepted_mesages = (True, False)
106
107	1	def __init__(self, field: list, mode: Optional[Mode]=Mode.Native, alignment: Optional[int]=1):
108		# All of the type checks have already been performed by the class
109		# factory
110	1	self.name = field[0]
111
112		# Callable elements use the normal struct packing method
113	1	self.format = field[1]
114
115	1	if isinstance(field[2], dict):
116	1	self.ref = field[2]
117
118	1	default_list = [None, None]
119	1	self._make_func = self.ref.get('make', default_list)[0]
120	1	self._make_args = self.ref.get('make', default_list)[1:]
121
122	1	self._pack_func = self.ref.get('pack', default_list)[0]
123	1	self._pack_args = self.ref.get('pack', default_list)[1:]
124
125	1	self._unpack_func = self.ref.get('unpack', default_list)[0]
126	1	self._unpack_args = self.ref.get('unpack', default_list)[1:]
127	1	elif isinstance(field[2], set):
128	1	instruction = field[2].copy().pop()
129	1	self.ref = {'all': instruction}
130
131	1	self._make_func = self._pack_func = self._unpack_func = instruction[0]
132	1	self._make_args = self._pack_args = self._unpack_args = instruction[1:]
133
134	1	if len(field) == 4:
135	1	self._error_on_bad_result = field[3]
136		else:
137	1	self._error_on_bad_result = True
138
139	1	self._elements = []
140
141	1	self.update(mode, alignment)
142
143	1	@property
144		def _struct(self):
145	1	return struct.Struct(self._mode.value + self.format)
146
147	1	@staticmethod
148	1	def valid(field: tuple) -> bool:
149		"""
150		See :py:func:`starstruct.element.Element.valid`
151
152		:param field: The items to determine the structure of the element
153		"""
154	1	required_keys = {'pack', 'unpack', 'make'}
155
156	1	if len(field) >= 3 and isinstance(field[0], str) and isinstance(field[1], str):
157	1	if isinstance(field[2], dict):
158	1	if set(field[2].keys()) <= required_keys and \
159		all(isinstance(val, tuple) for val in field[2].values()):
160	1	return True
161	1	elif isinstance(field[2], set):
162	1	return len(field[2]) == 1 and \
163		all(isinstance(val, tuple) for val in field[2])
164
165	1	return False
166
167	1	def validate(self, msg):
168		"""
169		Ensure that the supplied message contains the required information for
170		this element object to operate.
171
172		All elements that are Variable must reference valid Length elements.
173		"""
174	1	for action in self.ref.values():
175	1	for arg in action[1:]:
176	1	if arg in ElementCallable.accepted_mesages:
177		continue
178	1	elif isinstance(arg, str):
179	1	pass
180	1	elif hasattr(arg, 'decode'):
181	1	arg = arg.decode('utf-8')
182		elif hasattr(arg, 'to_bytes'):
183		arg = arg.to_bytes((arg.bit_length() + 7) // 8, self._mode.to_byteorder()).decode('utf-8')
184
185	1	if arg not in msg:
186		raise ValueError('Need all keys to be in the message, {0} was not found\nAction: {1} -> {2}'.format(arg, action, action[1:]))
187
188	1	def update(self, mode=None, alignment=None):
189		"""change the mode of the struct format"""
190	1	if mode:
191	1	self._mode = mode
192
193	1	if alignment:
194	1	self._alignment = alignment
195
196	1	def pack(self, msg):
197		"""Pack the provided values into the supplied buffer."""
198	1	pack_values = self.call_func(msg, self._pack_func, self._pack_args)
199
200		# Test if the object is iterable
201		# If it isn't, then turn it into a list
202	1	try:
203	1	_ = (p for p in pack_values)
204	1	except TypeError:
205	1	pack_values = [pack_values]
206
207		# Unpack the items for struct to allow for mutli-value
208		# items to be passed in.
209	1	return self._struct.pack(*pack_values)
		0 ignored issues – show Coding Style introduced 2016-12-19 19:55 UTC by Report Bug Copy Issue Report Usage of `` or `` arguments should usually be done with care. Generally, there is nothing wrong with usage of `` or `**` arguments. For readability of the code base, we suggest to not over-use these language constructs though. For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect. Loading history...
210
211	1	def unpack(self, msg, buf):
212		"""Unpack data from the supplied buffer using the initialized format."""
213	1	ret = self._struct.unpack_from(buf)
214	1	if isinstance(ret, (list, tuple)) and len(ret) == 1:
215		# We only change it not to a list if we expected one value.
216		# Otherwise, we keep it as a list, because that's what we would
217		# expect (like for a 16I type of struct
218	1	ret = ret[0]
219
220		# Only check for errors if they haven't told us not to
221	1	if self._error_on_bad_result:
222		# Pretend we're getting a dictionary to make our item,
223		# but it has no reference to `self`. This is so we check
224		# for errors correctly.
225	1	temp_dict = copy.deepcopy(msg._asdict())
226	1	temp_dict.pop(self.name)
227	1	expected_value = self.call_func(msg, self._unpack_func, self._unpack_args)
228
229		# Check for an error
230	1	if expected_value != ret:
231	1	raise ValueError('Expected value was: {0}, but got: {1}'.format(
232		expected_value,
233		ret,
234		))
235
236	1	if self.name in self._unpack_args:
237	1	msg = msg._replace(**{self.name: ret})
		0 ignored issues – show Coding Style introduced 2016-12-20 21:16 UTC by Report Bug Copy Issue Report Usage of `` or `` arguments should usually be done with care. Generally, there is nothing wrong with usage of `` or `**` arguments. For readability of the code base, we suggest to not over-use these language constructs though. For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect. Loading history...
238
239	1	ret = self.call_func(msg, self._unpack_func, self._unpack_args, original=ret)
240
241	1	return (ret, buf[self._struct.size:])
242
243	1	def make(self, msg):
244		"""Return the expected "made" value"""
245		# If we aren't going to error on a bad result
246		# and our name is in the message, just send the value
247		# No need to do extra work.
248	1	if not self._error_on_bad_result \
249		and self.name in msg \
250		and msg[self.name] is not None:
251	1	return msg[self.name]
252
253	1	ret = self.call_func(msg, self._make_func, self._make_args)
254
255	1	if self.name in msg:
256	1	if ret != msg[self.name]:
257	1	raise ValueError('Excepted value: {0}, but got: {1}'.format(ret, msg[self.name]))
258
259	1	return ret
260
261	1	def call_func(self, msg, func, args, original=None):
262	1	if func is None:
263	1	return original
264
265	1	items = self.prepare_args(msg, args)
266	1	return func(*items)
		0 ignored issues – show Coding Style introduced 2016-12-20 18:46 UTC by Report Bug Copy Issue Report Usage of `` or `` arguments should usually be done with care. Generally, there is nothing wrong with usage of `` or `**` arguments. For readability of the code base, we suggest to not over-use these language constructs though. For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect. Loading history...
267
268	1	def prepare_args(self, msg, args):
269	1	items = []
270
271	1	if hasattr(msg, '_asdict'):
272	1	msg = msg._asdict()
273
274	1	for reference in args:
275	1	if isinstance(reference, str):
276	1	index = reference
277	1	attr = 'make'
278	1	elif isinstance(reference, bytes):
279	1	index = reference.decode('utf-8')
280	1	attr = 'pack'
281		else:
282		raise ValueError('Needed str or bytes for the reference')
283
284	1	items.append(
285		getattr(self._elements[index], attr)(msg)
286		)
287
288		return items
289

sprout42 / StarStruct

Pull Request — master (#42)

ElementCallable.validate() C

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files