diff --git a/.gitignore b/.gitignore index ba04aac..b8fd992 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,6 @@ /xmodem.egg-info /.coverage /.tox +/.cache +/.idea *.py.swp diff --git a/test/unit/test_xmodem.py b/test/unit/test_xmodem.py index 3b724d5..0e0c96c 100644 --- a/test/unit/test_xmodem.py +++ b/test/unit/test_xmodem.py @@ -10,7 +10,8 @@ import StringIO.StringIO as BytesIO # local -import xmodem +from xmodem import XMODEM +from xmodem import NAK, CRC, ACK # 3rd-party import pytest @@ -24,24 +25,122 @@ def dummy_putc(data, timeout=1): return 0 +def test_xmodem_bad_mode(): + # given, + mode = 'XXX' + modem = XMODEM(getc=dummy_getc, putc=dummy_putc, mode=mode) + # exercise + with pytest.raises(ValueError): + status = modem.send(BytesIO(b'dummy-stream')) + + @pytest.mark.parametrize('mode', ['xmodem', 'xmodem1k']) def test_xmodem_dummy_fails_send(mode): # given, - modem = xmodem.XMODEM(getc=dummy_getc, - putc=dummy_putc, - mode=mode) + modem = XMODEM(getc=dummy_getc, putc=dummy_putc, mode=mode) # exercise status = modem.send(BytesIO(b'dummy-stream')) # verify assert not status, ("Expected value of status `False'") -def test_xmodem_bad_mode(): +@pytest.mark.parametrize('mode', ['xmodem', 'xmodem1k']) +def test_xmodem_send_exceed_maximum_number_of_resend(mode): # given, - mode = 'XXX' - modem = xmodem.XMODEM(getc=dummy_getc, - putc=dummy_putc, - mode=mode) + max_resend = 16 + + def generator(): + if mode == 'xmodem': + yield NAK + else: + yield CRC + yield ACK + yield ACK + + for i in range(max_resend + 1): + yield None + + mock = generator() + + def getc(size, timeout=1): + x = mock.next() + print("mock_getc => ", x) + return x + + def putc(data, timeout=1): + return 0 + + xmodem = XMODEM(getc=getc, putc=putc, mode=mode) # exercise - with pytest.raises(ValueError): - status = modem.send(BytesIO(b'dummy-stream')) + assert xmodem.send(stream = BytesIO(b'dummy-stream'), retry=max_resend) == None + + +@pytest.mark.parametrize('mode', ['xmodem', 'xmodem1k']) +def test_xmodem_send_fails_once_each_packet_short_ByteIO(mode): + # given, + max_resend = 16 + + def generator(): + if mode == 'xmodem': + yield NAK + else: + yield CRC + + while True: + yield None + yield ACK + + mock = generator() + + def getc(size, timeout=1): + x = mock.next() + print("mock_getc => ", x) + return x + + def putc(data, timeout=1): + return 0 + + xmodem = XMODEM(getc=getc, putc=putc, mode=mode) + # exercise + assert xmodem.send(stream=BytesIO(b'dummy-stream'), retry=max_resend) + + +@pytest.mark.parametrize('mode', ['xmodem', 'xmodem1k']) +def test_xmodem_send_fails_once_each_packet_long_ByteIO(mode): + # given, + max_resend = 16 + + def generator(): + if mode == 'xmodem': + yield NAK + else: + yield CRC + + yield ACK + for i in range(10): + yield None + yield ACK + + yield ACK + yield ACK + + for i in range(10): + yield None + yield ACK + + while True: + ACK + + mock = generator() + + def getc(size, timeout=1): + x = mock.next() + print("[]mock_getc => ", x) + return x + + def putc(data, timeout=1): + return 0 + + xmodem = XMODEM(getc=getc, putc=putc, mode=mode) + # exercise + assert xmodem.send(stream=BytesIO(b'dummy-stream'*30), retry=max_resend)