6738061857

Committed 02 Nov 2023 08:47PM UTC coverage: 91.889% (+2.1%) from 89.783%

Build # 6738061857

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #77 from haydenroche5/cobs

Run Details

504 of 504 new or added lines in 4 files covered. (100.0%)

2028 of 2207 relevant lines covered (91.89%)

4.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.43

/test/test_binary_helpers.py

import os
import sys
import pytest
from unittest.mock import MagicMock, patch

sys.path.insert(0,
                os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))

import notecard  # noqa: E402
from notecard.cobs import cobs_encode  # noqa: E402
from notecard.binary_helpers import (  # noqa: E402
    binary_store_decoded_length,
    binary_store_reset,
    binary_store_receive,
    binary_store_transmit,
    _md5_hash,
    BINARY_RETRIES
)


@pytest.fixture
def arrange_test():
    def _arrange_test():
        card = notecard.Notecard()
        card.Transaction = MagicMock()
        card.lock = MagicMock()
        card.unlock = MagicMock()

        return card

    yield _arrange_test


@pytest.fixture
def arrange_rx_test(arrange_test):
    def _arrange_rx_test(bad_md5=False):
        card = arrange_test()

        # Set up receive.
        rx_data = bytearray.fromhex('deadbeef0a')
        card.receive = MagicMock(return_value=rx_data)

        # Set up Transaction.
        if bad_md5:
            rsp = {'status': 'abc'}
        else:
            # This is the MD5 of 0xdeadbeef. Note that 0x0a is omitted -- that's
            # the newline that terminates the binary payload. It isn't included
            # in the MD5 calculation.
            rsp = {'status': '2f249230a8e7c2bf6005ccd2679259ec'}
        card.Transaction.return_value = rsp

        # Don't actually do any COBS decoding. Just return the received data.
        # data minus the newline.
        notecard.binary_helpers.cobs_decode = MagicMock(
            return_value=rx_data[:-1])

        return card

    yield _arrange_rx_test


@pytest.fixture
def arrange_tx_test(arrange_test):
    def _arrange_tx_test(transmit_exception=None, **kwargs):
        card = arrange_test()
        card.transmit = MagicMock()

        CARD_BINARY_PRE_TRANSMIT = 0
        CARD_BINARY_PUT = 1
        CARD_BINARY_POST_TRANSMIT = 2
        DONE = 3

        class TransactionSideEffect:
            '''Iterable for the Transaction method that uses a state machine for
               various binary_store_transmit scenarios exercised in these unit
               tests.'''
            def __init__(self, pre_transmit_err=None, maximum=1024, length=0,
                         card_binary_put_err=None,
                         card_binary_put_exception=None, post_transmit_err=None,
                         retry_transmit_forever=False):
                # 'err' field value in the pre-transmission card.binary
                # response.
                self.pre_transmit_err = pre_transmit_err
                # 'max' field value in the pre-transmission card.binary
                # response.
                self.maximum = maximum
                # 'length' field value in the pre-transmission card.binary
                # response.
                self.length = length
                # 'err' field value in the card.binary.put response.
                self.card_binary_put_err = card_binary_put_err
                # Raise an exception on the card.binary.put request, using the
                # string `card_binary_put_exception`.
                self.card_binary_put_exception = card_binary_put_exception
                # 'err' field value in the post-transmission card.binary
                # response.
                self.post_transmit_err = post_transmit_err
                # If this is true, post_transmit_err will not be cleared on a
                # transmission retry. This results in continuous retrying.
                self.retry_transmit_forever = retry_transmit_forever

            def __iter__(self):
                self.state = CARD_BINARY_PRE_TRANSMIT

            def __next__(self):
                # This is the aforementioned state machine. The first
                # Transaction call in binary_store_transmit is a card.binary
                # request.
                if self.state == CARD_BINARY_PRE_TRANSMIT:
                    rsp = {'length': self.length}
                    if self.maximum:
                        rsp['max'] = self.maximum
                    if self.pre_transmit_err:
                        rsp['err'] = self.pre_transmit_err
                    self.state = CARD_BINARY_PUT

                # The second call is card.binary.put.
                elif self.state == CARD_BINARY_PUT:
                    if self.card_binary_put_exception:
                        rsp = Exception(self.card_binary_put_exception)
                    elif self.card_binary_put_err:
                        rsp = {'err': self.card_binary_put_err}
                    else:
                        rsp = {}
                    self.state = CARD_BINARY_POST_TRANSMIT

                # The third call is card.binary again.
                elif self.state == CARD_BINARY_POST_TRANSMIT:
                    if self.post_transmit_err:
                        rsp = {'err': self.post_transmit_err}
                        # Unless we intend to error out on the post transmission
                        # card.binary continuously, clear the error here so that
                        # we don't hit it again on the retry.
                        if not self.retry_transmit_forever:
                            self.post_transmit_err = None

                        self.state = CARD_BINARY_PUT
                    else:
                        rsp = {}
                        self.state = DONE

                elif self.state == DONE:
                    raise StopIteration

                return rsp

        card.Transaction.side_effect = TransactionSideEffect(**kwargs)

        if transmit_exception:
            card.transmit.side_effect = Exception(transmit_exception)

        return card

    with patch('notecard.binary_helpers.cobs_encode', side_effect=cobs_encode):
        yield _arrange_tx_test


@pytest.fixture
def tx_data():
    return bytearray.fromhex(('82f9a19b7e02c95bd87b38a8ce479608830ee68ffe4d1763'
                              'ecc205681ed537a3'))


def get_card_binary_put_call(card):
    '''Find the card.binary.put request by searching through the calls to
       card.Transaction.'''
    card_binary_put_call = None
    for call in card.Transaction.call_args_list:
        req = call[0][0]
        if req['req'] == 'card.binary.put':
            card_binary_put_call = call
            break

    return card_binary_put_call


class TestBinaryStoreDecodedLength:
    # binary_store_decoded_length tests.
    def test_makes_card_binary_request(self, arrange_test):
        card = arrange_test()

        binary_store_decoded_length(card)

        card.Transaction.assert_called_once_with({'req': 'card.binary'})

    def test_returns_zero_if_no_length_field(self, arrange_test):
        card = arrange_test()
        card.Transaction.return_value = {}

        assert binary_store_decoded_length(card) == 0

    def test_returns_length_length_field_present(self, arrange_test):
        card = arrange_test()
        length = 99
        card.Transaction.return_value = {'length': length}

        assert binary_store_decoded_length(card) == length

    def test_ignores_bad_bin_errors(self, arrange_test):
        card = arrange_test()
        card.Transaction.return_value = {'err': '{bad-bin}'}

        binary_store_decoded_length(card)

    def test_raises_exception_on_non_bad_bin_error(self, arrange_test):
        card = arrange_test()
        err_msg = 'some error'
        card.Transaction.return_value = {'err': err_msg}
        exception_msg = f'Error in response to card.binary request: {err_msg}.'

        with pytest.raises(Exception, match=exception_msg):
            binary_store_decoded_length(card)


class TestBinaryStoreReset:
    def test_makes_card_binary_request_with_delete_true(
            self, arrange_test):
        card = arrange_test()

        binary_store_reset(card)

        card.Transaction.assert_called_once_with(
            {'req': 'card.binary', 'delete': True})

    def test_raises_exception_on_error_in_response(self, arrange_test):
        card = arrange_test()
        err_msg = 'some error'
        card.Transaction.return_value = {'err': err_msg}
        exception_msg = ('Error in response to card.binary delete request: '
                         f'{err_msg}.')

        with pytest.raises(Exception, match=exception_msg):
            binary_store_reset(card)


class TestBinaryStoreReceive:
    def test_maps_card_binary_get_params_correctly(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99
        expected_req = {
            'req': 'card.binary.get',
            'offset': offset,
            'length': length
        }

        binary_store_receive(card, offset, length)

        card.Transaction.assert_called_once_with(expected_req, lock=False)

    def test_locks_and_unlocks(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99

        binary_store_receive(card, offset, length)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_unlocks_after_card_binary_get_exception(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99
        exception_msg = 'Transaction failed.'
        card.Transaction.side_effect = Exception(exception_msg)

        with pytest.raises(Exception, match=exception_msg):
            binary_store_receive(card, offset, length)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_unlocks_after_receive_exception(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99
        exception_msg = 'receive failed.'
        card.receive.side_effect = Exception(exception_msg)

        with pytest.raises(Exception, match=exception_msg):
            binary_store_receive(card, offset, length)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_raises_exception_if_error_in_card_binary_get_response(
            self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99
        err_msg = 'some error'
        card.Transaction.return_value['err'] = err_msg

        with pytest.raises(Exception, match=err_msg):
            binary_store_receive(card, offset, length)

    def test_unlocks_if_error_in_card_binary_get_response(
            self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99
        err_msg = 'some error'
        card.Transaction.return_value['err'] = err_msg

        with pytest.raises(Exception, match=err_msg):
            binary_store_receive(card, offset, length)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_queues_reset_after_receive_exception(self, arrange_rx_test):
        card = arrange_rx_test()
        card._reset_required = False
        offset = 11
        length = 99
        exception_msg = 'receive failed.'
        card.receive.side_effect = Exception(exception_msg)

        with pytest.raises(Exception, match=exception_msg):
            binary_store_receive(card, offset, length)

        assert card._reset_required

    def test_raises_exception_on_bad_md5(self, arrange_rx_test):
        card = arrange_rx_test(bad_md5=True)
        offset = 11
        length = 99
        exception_msg = 'Computed MD5 does not match received MD5.'

        with pytest.raises(Exception, match=exception_msg):
            binary_store_receive(card, offset, length)

    def test_calls_cobs_decode_on_rx_data_without_newline(
            self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99

        binary_store_receive(card, offset, length)

        # First, verify what was received ends in a newline.
        assert card.receive.return_value[-1] == ord('\n')
        # Then, check that cobs_decode was called with everything but that
        # newline.
        assert card.receive.return_value[:-1] == \
               notecard.binary_helpers.cobs_decode.call_args[0][0]

    def test_calls_cobs_decode_with_newline_as_eop(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99

        binary_store_receive(card, offset, length)

        assert notecard.binary_helpers.cobs_decode.call_args[0][1] == ord('\n')

    def test_returns_cobs_decoded_data(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99

        data = binary_store_receive(card, offset, length)

        assert notecard.binary_helpers.cobs_decode.return_value == data

    def test_computes_md5_over_cobs_decoded_data(self, arrange_rx_test):
        card = arrange_rx_test()
        offset = 11
        length = 99
        md5_fn = notecard.binary_helpers._md5_hash

        with patch('notecard.binary_helpers._md5_hash',
                   side_effect=md5_fn) as md5_mock:
            binary_store_receive(card, offset, length)

            cobs_decoded_data = \
                notecard.binary_helpers.cobs_decode.call_args[0][0]
            assert md5_mock.call_args[0][0] == cobs_decoded_data


class TestBinaryStoreTransmit:
    def test_ignores_bad_bin_err_on_initial_card_binary_request(
            self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test(pre_transmit_err='{bad-bin}')

        binary_store_transmit(card, tx_data, offset)

    def test_fails_on_non_bad_bin_err_on_initial_card_binary_request(
            self, tx_data, arrange_tx_test):
        offset = 0
        err_msg = 'some error'
        card = arrange_tx_test(pre_transmit_err=err_msg)

        with pytest.raises(Exception, match=err_msg):
            binary_store_transmit(card, tx_data, offset)

    @pytest.mark.parametrize('maximum', [None, 0])
    def test_fails_on_invalid_max_on_initial_card_binary_request(
            self, tx_data, arrange_tx_test, maximum):
        offset = 0
        card = arrange_tx_test(maximum=maximum)
        exception_msg = ('Unexpected card.binary response: max is zero or not '
                         'present.')

        with pytest.raises(Exception, match=exception_msg):
            binary_store_transmit(card, tx_data, offset)

    def test_fails_if_offset_not_equal_to_length_reported_by_notecard(
            self, tx_data, arrange_tx_test):
        offset = 1
        card = arrange_tx_test()
        exception_msg = 'Notecard data length is misaligned with offset.'

        with pytest.raises(Exception, match=exception_msg):
            binary_store_transmit(card, tx_data, offset)

    def test_fails_if_data_length_exceeds_available_space_on_notecard(
            self, tx_data, arrange_tx_test):
        offset = 5
        card = arrange_tx_test(maximum=offset, length=offset)
        exception_msg = ('Data to transmit won\'t fit in the Notecard\'s binary'
                         ' store.')

        with pytest.raises(Exception, match=exception_msg):
            binary_store_transmit(card, tx_data, offset)

    def test_calls_cobs_encode_on_tx_data(self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test()

        binary_store_transmit(card, tx_data, offset)

        assert notecard.binary_helpers.cobs_encode.call_args[0][0] == tx_data

    def test_calls_cobs_encode_with_newline_as_eop(
            self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test()

        binary_store_transmit(card, tx_data, offset)

        assert notecard.binary_helpers.cobs_encode.call_args[0][1] == ord('\n')

    def test_maps_card_binary_put_params_correctly(
            self, tx_data, arrange_tx_test):
        offset = 1
        card = arrange_tx_test(length=offset)

        binary_store_transmit(card, tx_data, offset)

        card_binary_put_call = get_card_binary_put_call(card)
        # If we didn't find a card.binary.put request, something's wrong.
        assert card_binary_put_call is not None

        req = card_binary_put_call[0][0]
        assert req['cobs'] == len(cobs_encode(tx_data, ord('\n')))
        assert req['status'] == _md5_hash(tx_data)
        assert req['offset'] == offset

    def test_sends_card_binary_put_with_lock_false(
            self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test()

        binary_store_transmit(card, tx_data, offset)

        card_binary_put_call = get_card_binary_put_call(card)
        # If we didn't find a card.binary.put request, something's wrong.
        assert card_binary_put_call is not None

        assert not card_binary_put_call[1]['lock']

    def test_calls_transmit_with_delay_false(self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test()

        binary_store_transmit(card, tx_data, offset)

        assert not card.transmit.call_args[1]['delay']

    def test_calls_transmit_with_cobs_encoded_data_plus_newline(
            self, tx_data, arrange_tx_test):
        expected_data = cobs_encode(bytearray(tx_data), ord('\n')) + b'\n'
        offset = 0
        card = arrange_tx_test()

        binary_store_transmit(card, tx_data, offset)

        assert card.transmit.call_args[0][0] == expected_data

    def test_raises_exception_on_card_binary_put_error(
            self, tx_data, arrange_tx_test):
        offset = 0
        err_msg = 'some error'
        card = arrange_tx_test(card_binary_put_err=err_msg)

        with pytest.raises(Exception, match=err_msg):
            binary_store_transmit(card, tx_data, offset)

    def test_locks_and_unlocks(self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test()

        binary_store_transmit(card, tx_data, offset)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_unlocks_after_card_binary_put_exception(
            self, tx_data, arrange_tx_test):
        offset = 0
        exception_msg = 'card.binary.put failed'
        card = arrange_tx_test(
            card_binary_put_exception=exception_msg)

        with pytest.raises(Exception, match=exception_msg):
            binary_store_transmit(card, tx_data, offset)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_unlocks_after_card_binary_put_error(
            self, tx_data, arrange_tx_test):
        offset = 0
        err_msg = 'some error'
        card = arrange_tx_test(card_binary_put_err=err_msg)

        with pytest.raises(Exception, match=err_msg):
            binary_store_transmit(card, tx_data, offset)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_unlocks_after_transmit_exception(self, tx_data, arrange_tx_test):
        offset = 0
        exception_msg = 'transmit failed'
        card = arrange_tx_test(
            transmit_exception=exception_msg)

        with pytest.raises(Exception, match=exception_msg):
            binary_store_transmit(card, tx_data, offset)

        card.lock.assert_called_once()
        card.unlock.assert_called_once()

    def test_retries_on_bad_bin_error_after_transmit(
            self, tx_data, arrange_tx_test):
        offset = 0
        card = arrange_tx_test(post_transmit_err='{bad-bin}')

        binary_store_transmit(card, tx_data, offset)

        # transmit should've been called once and then retried once due to the
        # {bad-bin} error.
        assert card.transmit.call_count == 2

    def test_fails_on_card_binary_error_after_transmission(
            self, tx_data, arrange_tx_test):
        offset = 0
        err_msg = 'some error'
        card = arrange_tx_test(post_transmit_err=err_msg)

        with pytest.raises(Exception, match=err_msg):
            binary_store_transmit(card, tx_data, offset)

    def test_fails_after_running_out_of_retries(
            self, tx_data, arrange_tx_test):
        offset = 0
        exception_msg = 'Failed to transmit binary data.'
        card = arrange_tx_test(post_transmit_err='{bad-bin}',
                               retry_transmit_forever=True)

        with pytest.raises(Exception, match=exception_msg):
            binary_store_transmit(card, tx_data, offset)

        # transmit should've been called once and then retried BINARY_RETRIES
        # times.
        assert card.transmit.call_count == BINARY_RETRIES + 1

1	import os	5✔
2	import sys	5✔
3	import pytest	5✔
4	from unittest.mock import MagicMock, patch	5✔
5
6	sys.path.insert(0,	5✔
7	os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
8
9	import notecard # noqa: E402	5✔
10	from notecard.cobs import cobs_encode # noqa: E402	5✔
11	from notecard.binary_helpers import ( # noqa: E402	5✔
12	binary_store_decoded_length,
13	binary_store_reset,
14	binary_store_receive,
15	binary_store_transmit,
16	_md5_hash,
17	BINARY_RETRIES
18	)
19
20
21	@pytest.fixture	5✔
22	def arrange_test():	3✔
23	def _arrange_test():	5✔
24	card = notecard.Notecard()	5✔
25	card.Transaction = MagicMock()	5✔
26	card.lock = MagicMock()	5✔
27	card.unlock = MagicMock()	5✔
28
29	return card	5✔
30
31	yield _arrange_test	5✔
32
33
34	@pytest.fixture	5✔
35	def arrange_rx_test(arrange_test):	3✔
36	def _arrange_rx_test(bad_md5=False):	5✔
37	card = arrange_test()	5✔
38
39	# Set up receive.
40	rx_data = bytearray.fromhex('deadbeef0a')	5✔
41	card.receive = MagicMock(return_value=rx_data)	5✔
42
43	# Set up Transaction.
44	if bad_md5:	5✔
45	rsp = {'status': 'abc'}	5✔
46	else:
47	# This is the MD5 of 0xdeadbeef. Note that 0x0a is omitted -- that's
48	# the newline that terminates the binary payload. It isn't included
49	# in the MD5 calculation.
50	rsp = {'status': '2f249230a8e7c2bf6005ccd2679259ec'}	5✔
51	card.Transaction.return_value = rsp	5✔
52
53	# Don't actually do any COBS decoding. Just return the received data.
54	# data minus the newline.
55	notecard.binary_helpers.cobs_decode = MagicMock(	5✔
56	return_value=rx_data[:-1])
57
58	return card	5✔
59
60	yield _arrange_rx_test	5✔
61
62
63	@pytest.fixture	5✔
64	def arrange_tx_test(arrange_test):	3✔
65	def _arrange_tx_test(transmit_exception=None, **kwargs):	5✔
66	card = arrange_test()	5✔
67	card.transmit = MagicMock()	5✔
68
69	CARD_BINARY_PRE_TRANSMIT = 0	5✔
70	CARD_BINARY_PUT = 1	5✔
71	CARD_BINARY_POST_TRANSMIT = 2	5✔
72	DONE = 3	5✔
73
74	class TransactionSideEffect:	5✔
75	'''Iterable for the Transaction method that uses a state machine for
76	various binary_store_transmit scenarios exercised in these unit
77	tests.'''
78	def __init__(self, pre_transmit_err=None, maximum=1024, length=0,	5✔
79	card_binary_put_err=None,
80	card_binary_put_exception=None, post_transmit_err=None,
81	retry_transmit_forever=False):
82	# 'err' field value in the pre-transmission card.binary
83	# response.
84	self.pre_transmit_err = pre_transmit_err	5✔
85	# 'max' field value in the pre-transmission card.binary
86	# response.
87	self.maximum = maximum	5✔
88	# 'length' field value in the pre-transmission card.binary
89	# response.
90	self.length = length	5✔
91	# 'err' field value in the card.binary.put response.
92	self.card_binary_put_err = card_binary_put_err	5✔
93	# Raise an exception on the card.binary.put request, using the
94	# string `card_binary_put_exception`.
95	self.card_binary_put_exception = card_binary_put_exception	5✔
96	# 'err' field value in the post-transmission card.binary
97	# response.
98	self.post_transmit_err = post_transmit_err	5✔
99	# If this is true, post_transmit_err will not be cleared on a
100	# transmission retry. This results in continuous retrying.
101	self.retry_transmit_forever = retry_transmit_forever	5✔
102
103	def __iter__(self):	5✔
104	self.state = CARD_BINARY_PRE_TRANSMIT	5✔
105
106	def __next__(self):	5✔
107	# This is the aforementioned state machine. The first
108	# Transaction call in binary_store_transmit is a card.binary
109	# request.
110	if self.state == CARD_BINARY_PRE_TRANSMIT:	5✔
111	rsp = {'length': self.length}	5✔
112	if self.maximum:	5✔
113	rsp['max'] = self.maximum	5✔
114	if self.pre_transmit_err:	5✔
115	rsp['err'] = self.pre_transmit_err	5✔
116	self.state = CARD_BINARY_PUT	5✔
117
118	# The second call is card.binary.put.
119	elif self.state == CARD_BINARY_PUT:	5✔
120	if self.card_binary_put_exception:	5✔
121	rsp = Exception(self.card_binary_put_exception)	5✔
122	elif self.card_binary_put_err:	5✔
123	rsp = {'err': self.card_binary_put_err}	5✔
124	else:
125	rsp = {}	5✔
126	self.state = CARD_BINARY_POST_TRANSMIT	5✔
127
128	# The third call is card.binary again.
129	elif self.state == CARD_BINARY_POST_TRANSMIT:	5✔
130	if self.post_transmit_err:	5✔
131	rsp = {'err': self.post_transmit_err}	5✔
132	# Unless we intend to error out on the post transmission
133	# card.binary continuously, clear the error here so that
134	# we don't hit it again on the retry.
135	if not self.retry_transmit_forever:	5✔
136	self.post_transmit_err = None	5✔
137
138	self.state = CARD_BINARY_PUT	5✔
139	else:
140	rsp = {}	5✔
141	self.state = DONE	5✔
142
143	elif self.state == DONE:	×
144	raise StopIteration	×
145
146	return rsp	5✔
147
148	card.Transaction.side_effect = TransactionSideEffect(**kwargs)	5✔
149
150	if transmit_exception:	5✔
151	card.transmit.side_effect = Exception(transmit_exception)	5✔
152
153	return card	5✔
154
155	with patch('notecard.binary_helpers.cobs_encode', side_effect=cobs_encode):	5✔
156	yield _arrange_tx_test	5✔
157
158
159	@pytest.fixture	5✔
160	def tx_data():	3✔
161	return bytearray.fromhex(('82f9a19b7e02c95bd87b38a8ce479608830ee68ffe4d1763'	5✔
162	'ecc205681ed537a3'))
163
164
165	def get_card_binary_put_call(card):	5✔
166	'''Find the card.binary.put request by searching through the calls to
167	card.Transaction.'''
168	card_binary_put_call = None	5✔
169	for call in card.Transaction.call_args_list:	5✔
170	req = call[0][0]	5✔
171	if req['req'] == 'card.binary.put':	5✔
172	card_binary_put_call = call	5✔
173	break	5✔
174
175	return card_binary_put_call	5✔
176
177
178	class TestBinaryStoreDecodedLength:	5✔
179	# binary_store_decoded_length tests.
180	def test_makes_card_binary_request(self, arrange_test):	5✔
181	card = arrange_test()	5✔
182
183	binary_store_decoded_length(card)	5✔
184
185	card.Transaction.assert_called_once_with({'req': 'card.binary'})	5✔
186
187	def test_returns_zero_if_no_length_field(self, arrange_test):	5✔
188	card = arrange_test()	5✔
189	card.Transaction.return_value = {}	5✔
190
191	assert binary_store_decoded_length(card) == 0	5✔
192
193	def test_returns_length_length_field_present(self, arrange_test):	5✔
194	card = arrange_test()	5✔
195	length = 99	5✔
196	card.Transaction.return_value = {'length': length}	5✔
197
198	assert binary_store_decoded_length(card) == length	5✔
199
200	def test_ignores_bad_bin_errors(self, arrange_test):	5✔
201	card = arrange_test()	5✔
202	card.Transaction.return_value = {'err': '{bad-bin}'}	5✔
203
204	binary_store_decoded_length(card)	5✔
205
206	def test_raises_exception_on_non_bad_bin_error(self, arrange_test):	5✔
207	card = arrange_test()	5✔
208	err_msg = 'some error'	5✔
209	card.Transaction.return_value = {'err': err_msg}	5✔
210	exception_msg = f'Error in response to card.binary request: {err_msg}.'	5✔
211
212	with pytest.raises(Exception, match=exception_msg):	5✔
213	binary_store_decoded_length(card)	5✔
214
215
216	class TestBinaryStoreReset:	5✔
217	def test_makes_card_binary_request_with_delete_true(	5✔
218	self, arrange_test):
219	card = arrange_test()	5✔
220
221	binary_store_reset(card)	5✔
222
223	card.Transaction.assert_called_once_with(	5✔
224	{'req': 'card.binary', 'delete': True})
225
226	def test_raises_exception_on_error_in_response(self, arrange_test):	5✔
227	card = arrange_test()	5✔
228	err_msg = 'some error'	5✔
229	card.Transaction.return_value = {'err': err_msg}	5✔
230	exception_msg = ('Error in response to card.binary delete request: '	5✔
231	f'{err_msg}.')
232
233	with pytest.raises(Exception, match=exception_msg):	5✔
234	binary_store_reset(card)	5✔
235
236
237	class TestBinaryStoreReceive:	5✔
238	def test_maps_card_binary_get_params_correctly(self, arrange_rx_test):	5✔
239	card = arrange_rx_test()	5✔
240	offset = 11	5✔
241	length = 99	5✔
242	expected_req = {	5✔
243	'req': 'card.binary.get',
244	'offset': offset,
245	'length': length
246	}
247
248	binary_store_receive(card, offset, length)	5✔
249
250	card.Transaction.assert_called_once_with(expected_req, lock=False)	5✔
251
252	def test_locks_and_unlocks(self, arrange_rx_test):	5✔
253	card = arrange_rx_test()	5✔
254	offset = 11	5✔
255	length = 99	5✔
256
257	binary_store_receive(card, offset, length)	5✔
258
259	card.lock.assert_called_once()	5✔
260	card.unlock.assert_called_once()	5✔
261
262	def test_unlocks_after_card_binary_get_exception(self, arrange_rx_test):	5✔
263	card = arrange_rx_test()	5✔
264	offset = 11	5✔
265	length = 99	5✔
266	exception_msg = 'Transaction failed.'	5✔
267	card.Transaction.side_effect = Exception(exception_msg)	5✔
268
269	with pytest.raises(Exception, match=exception_msg):	5✔
270	binary_store_receive(card, offset, length)	5✔
271
272	card.lock.assert_called_once()	5✔
273	card.unlock.assert_called_once()	5✔
274
275	def test_unlocks_after_receive_exception(self, arrange_rx_test):	5✔
276	card = arrange_rx_test()	5✔
277	offset = 11	5✔
278	length = 99	5✔
279	exception_msg = 'receive failed.'	5✔
280	card.receive.side_effect = Exception(exception_msg)	5✔
281
282	with pytest.raises(Exception, match=exception_msg):	5✔
283	binary_store_receive(card, offset, length)	5✔
284
285	card.lock.assert_called_once()	5✔
286	card.unlock.assert_called_once()	5✔
287
288	def test_raises_exception_if_error_in_card_binary_get_response(	5✔
289	self, arrange_rx_test):
290	card = arrange_rx_test()	5✔
291	offset = 11	5✔
292	length = 99	5✔
293	err_msg = 'some error'	5✔
294	card.Transaction.return_value['err'] = err_msg	5✔
295
296	with pytest.raises(Exception, match=err_msg):	5✔
297	binary_store_receive(card, offset, length)	5✔
298
299	def test_unlocks_if_error_in_card_binary_get_response(	5✔
300	self, arrange_rx_test):
301	card = arrange_rx_test()	5✔
302	offset = 11	5✔
303	length = 99	5✔
304	err_msg = 'some error'	5✔
305	card.Transaction.return_value['err'] = err_msg	5✔
306
307	with pytest.raises(Exception, match=err_msg):	5✔
308	binary_store_receive(card, offset, length)	5✔
309
310	card.lock.assert_called_once()	5✔
311	card.unlock.assert_called_once()	5✔
312
313	def test_queues_reset_after_receive_exception(self, arrange_rx_test):	5✔
314	card = arrange_rx_test()	5✔
315	card._reset_required = False	5✔
316	offset = 11	5✔
317	length = 99	5✔
318	exception_msg = 'receive failed.'	5✔
319	card.receive.side_effect = Exception(exception_msg)	5✔
320
321	with pytest.raises(Exception, match=exception_msg):	5✔
322	binary_store_receive(card, offset, length)	5✔
323
324	assert card._reset_required	5✔
325
326	def test_raises_exception_on_bad_md5(self, arrange_rx_test):	5✔
327	card = arrange_rx_test(bad_md5=True)	5✔
328	offset = 11	5✔
329	length = 99	5✔
330	exception_msg = 'Computed MD5 does not match received MD5.'	5✔
331
332	with pytest.raises(Exception, match=exception_msg):	5✔
333	binary_store_receive(card, offset, length)	5✔
334
335	def test_calls_cobs_decode_on_rx_data_without_newline(	5✔
336	self, arrange_rx_test):
337	card = arrange_rx_test()	5✔
338	offset = 11	5✔
339	length = 99	5✔
340
341	binary_store_receive(card, offset, length)	5✔
342
343	# First, verify what was received ends in a newline.
344	assert card.receive.return_value[-1] == ord('\n')	5✔
345	# Then, check that cobs_decode was called with everything but that
346	# newline.
347	assert card.receive.return_value[:-1] == \	5✔
348	notecard.binary_helpers.cobs_decode.call_args[0][0]
349
350	def test_calls_cobs_decode_with_newline_as_eop(self, arrange_rx_test):	5✔
351	card = arrange_rx_test()	5✔
352	offset = 11	5✔
353	length = 99	5✔
354
355	binary_store_receive(card, offset, length)	5✔
356
357	assert notecard.binary_helpers.cobs_decode.call_args[0][1] == ord('\n')	5✔
358
359	def test_returns_cobs_decoded_data(self, arrange_rx_test):	5✔
360	card = arrange_rx_test()	5✔
361	offset = 11	5✔
362	length = 99	5✔
363
364	data = binary_store_receive(card, offset, length)	5✔
365
366	assert notecard.binary_helpers.cobs_decode.return_value == data	5✔
367
368	def test_computes_md5_over_cobs_decoded_data(self, arrange_rx_test):	5✔
369	card = arrange_rx_test()	5✔
370	offset = 11	5✔
371	length = 99	5✔
372	md5_fn = notecard.binary_helpers._md5_hash	5✔
373
374	with patch('notecard.binary_helpers._md5_hash',	5✔
375	side_effect=md5_fn) as md5_mock:
376	binary_store_receive(card, offset, length)	5✔
377
378	cobs_decoded_data = \	5✔
379	notecard.binary_helpers.cobs_decode.call_args[0][0]
380	assert md5_mock.call_args[0][0] == cobs_decoded_data	5✔
381
382
383	class TestBinaryStoreTransmit:	5✔
384	def test_ignores_bad_bin_err_on_initial_card_binary_request(	5✔
385	self, tx_data, arrange_tx_test):
386	offset = 0	5✔
387	card = arrange_tx_test(pre_transmit_err='{bad-bin}')	5✔
388
389	binary_store_transmit(card, tx_data, offset)	5✔
390
391	def test_fails_on_non_bad_bin_err_on_initial_card_binary_request(	5✔
392	self, tx_data, arrange_tx_test):
393	offset = 0	5✔
394	err_msg = 'some error'	5✔
395	card = arrange_tx_test(pre_transmit_err=err_msg)	5✔
396
397	with pytest.raises(Exception, match=err_msg):	5✔
398	binary_store_transmit(card, tx_data, offset)	5✔
399
400	@pytest.mark.parametrize('maximum', [None, 0])	5✔
401	def test_fails_on_invalid_max_on_initial_card_binary_request(	3✔
402	self, tx_data, arrange_tx_test, maximum):
403	offset = 0	5✔
404	card = arrange_tx_test(maximum=maximum)	5✔
405	exception_msg = ('Unexpected card.binary response: max is zero or not '	5✔
406	'present.')
407
408	with pytest.raises(Exception, match=exception_msg):	5✔
409	binary_store_transmit(card, tx_data, offset)	5✔
410
411	def test_fails_if_offset_not_equal_to_length_reported_by_notecard(	5✔
412	self, tx_data, arrange_tx_test):
413	offset = 1	5✔
414	card = arrange_tx_test()	5✔
415	exception_msg = 'Notecard data length is misaligned with offset.'	5✔
416
417	with pytest.raises(Exception, match=exception_msg):	5✔
418	binary_store_transmit(card, tx_data, offset)	5✔
419
420	def test_fails_if_data_length_exceeds_available_space_on_notecard(	5✔
421	self, tx_data, arrange_tx_test):
422	offset = 5	5✔
423	card = arrange_tx_test(maximum=offset, length=offset)	5✔
424	exception_msg = ('Data to transmit won\'t fit in the Notecard\'s binary'	5✔
425	' store.')
426
427	with pytest.raises(Exception, match=exception_msg):	5✔
428	binary_store_transmit(card, tx_data, offset)	5✔
429
430	def test_calls_cobs_encode_on_tx_data(self, tx_data, arrange_tx_test):	5✔
431	offset = 0	5✔
432	card = arrange_tx_test()	5✔
433
434	binary_store_transmit(card, tx_data, offset)	5✔
435
436	assert notecard.binary_helpers.cobs_encode.call_args[0][0] == tx_data	5✔
437
438	def test_calls_cobs_encode_with_newline_as_eop(	5✔
439	self, tx_data, arrange_tx_test):
440	offset = 0	5✔
441	card = arrange_tx_test()	5✔
442
443	binary_store_transmit(card, tx_data, offset)	5✔
444
445	assert notecard.binary_helpers.cobs_encode.call_args[0][1] == ord('\n')	5✔
446
447	def test_maps_card_binary_put_params_correctly(	5✔
448	self, tx_data, arrange_tx_test):
449	offset = 1	5✔
450	card = arrange_tx_test(length=offset)	5✔
451
452	binary_store_transmit(card, tx_data, offset)	5✔
453
454	card_binary_put_call = get_card_binary_put_call(card)	5✔
455	# If we didn't find a card.binary.put request, something's wrong.
456	assert card_binary_put_call is not None	5✔
457
458	req = card_binary_put_call[0][0]	5✔
459	assert req['cobs'] == len(cobs_encode(tx_data, ord('\n')))	5✔
460	assert req['status'] == _md5_hash(tx_data)	5✔
461	assert req['offset'] == offset	5✔
462
463	def test_sends_card_binary_put_with_lock_false(	5✔
464	self, tx_data, arrange_tx_test):
465	offset = 0	5✔
466	card = arrange_tx_test()	5✔
467
468	binary_store_transmit(card, tx_data, offset)	5✔
469
470	card_binary_put_call = get_card_binary_put_call(card)	5✔
471	# If we didn't find a card.binary.put request, something's wrong.
472	assert card_binary_put_call is not None	5✔
473
474	assert not card_binary_put_call[1]['lock']	5✔
475
476	def test_calls_transmit_with_delay_false(self, tx_data, arrange_tx_test):	5✔
477	offset = 0	5✔
478	card = arrange_tx_test()	5✔
479
480	binary_store_transmit(card, tx_data, offset)	5✔
481
482	assert not card.transmit.call_args[1]['delay']	5✔
483
484	def test_calls_transmit_with_cobs_encoded_data_plus_newline(	5✔
485	self, tx_data, arrange_tx_test):
486	expected_data = cobs_encode(bytearray(tx_data), ord('\n')) + b'\n'	5✔
487	offset = 0	5✔
488	card = arrange_tx_test()	5✔
489
490	binary_store_transmit(card, tx_data, offset)	5✔
491
492	assert card.transmit.call_args[0][0] == expected_data	5✔
493
494	def test_raises_exception_on_card_binary_put_error(	5✔
495	self, tx_data, arrange_tx_test):
496	offset = 0	5✔
497	err_msg = 'some error'	5✔
498	card = arrange_tx_test(card_binary_put_err=err_msg)	5✔
499
500	with pytest.raises(Exception, match=err_msg):	5✔
501	binary_store_transmit(card, tx_data, offset)	5✔
502
503	def test_locks_and_unlocks(self, tx_data, arrange_tx_test):	5✔
504	offset = 0	5✔
505	card = arrange_tx_test()	5✔
506
507	binary_store_transmit(card, tx_data, offset)	5✔
508
509	card.lock.assert_called_once()	5✔
510	card.unlock.assert_called_once()	5✔
511
512	def test_unlocks_after_card_binary_put_exception(	5✔
513	self, tx_data, arrange_tx_test):
514	offset = 0	5✔
515	exception_msg = 'card.binary.put failed'	5✔
516	card = arrange_tx_test(	5✔
517	card_binary_put_exception=exception_msg)
518
519	with pytest.raises(Exception, match=exception_msg):	5✔
520	binary_store_transmit(card, tx_data, offset)	5✔
521
522	card.lock.assert_called_once()	5✔
523	card.unlock.assert_called_once()	5✔
524
525	def test_unlocks_after_card_binary_put_error(	5✔
526	self, tx_data, arrange_tx_test):
527	offset = 0	5✔
528	err_msg = 'some error'	5✔
529	card = arrange_tx_test(card_binary_put_err=err_msg)	5✔
530
531	with pytest.raises(Exception, match=err_msg):	5✔
532	binary_store_transmit(card, tx_data, offset)	5✔
533
534	card.lock.assert_called_once()	5✔
535	card.unlock.assert_called_once()	5✔
536
537	def test_unlocks_after_transmit_exception(self, tx_data, arrange_tx_test):	5✔
538	offset = 0	5✔
539	exception_msg = 'transmit failed'	5✔
540	card = arrange_tx_test(	5✔
541	transmit_exception=exception_msg)
542
543	with pytest.raises(Exception, match=exception_msg):	5✔
544	binary_store_transmit(card, tx_data, offset)	5✔
545
546	card.lock.assert_called_once()	5✔
547	card.unlock.assert_called_once()	5✔
548
549	def test_retries_on_bad_bin_error_after_transmit(	5✔
550	self, tx_data, arrange_tx_test):
551	offset = 0	5✔
552	card = arrange_tx_test(post_transmit_err='{bad-bin}')	5✔
553
554	binary_store_transmit(card, tx_data, offset)	5✔
555
556	# transmit should've been called once and then retried once due to the
557	# {bad-bin} error.
558	assert card.transmit.call_count == 2	5✔
559
560	def test_fails_on_card_binary_error_after_transmission(	5✔
561	self, tx_data, arrange_tx_test):
562	offset = 0	5✔
563	err_msg = 'some error'	5✔
564	card = arrange_tx_test(post_transmit_err=err_msg)	5✔
565
566	with pytest.raises(Exception, match=err_msg):	5✔
567	binary_store_transmit(card, tx_data, offset)	5✔
568
569	def test_fails_after_running_out_of_retries(	5✔
570	self, tx_data, arrange_tx_test):
571	offset = 0	5✔
572	exception_msg = 'Failed to transmit binary data.'	5✔
573	card = arrange_tx_test(post_transmit_err='{bad-bin}',	5✔
574	retry_transmit_forever=True)
575
576	with pytest.raises(Exception, match=exception_msg):	5✔
577	binary_store_transmit(card, tx_data, offset)	5✔
578
579	# transmit should've been called once and then retried BINARY_RETRIES
580	# times.
581	assert card.transmit.call_count == BINARY_RETRIES + 1	5✔

blues / note-python / 6738061857

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous