Skip to content

Commit

Permalink
refined code using bytes type
Browse files Browse the repository at this point in the history
  • Loading branch information
@rossengeorgiev
rossengeorgiev committed May 27, 2015
1 parent 5329bf7 commit 37a8f29
Show file tree
Hide file tree
Showing 4 changed files with 34 additions and 49 deletions.
10 changes: 2 additions & 8 deletions aprslib/IS.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,10 +67,7 @@ def __init__(self, callsign, passwd="-1", host="rotate.aprs.net", port=10152):
self.filter = "" # default filter, everything

self._connected = False
if is_py3:
self.buf = b''
else:
self.buf = ''
self.buf = b''

def _sendall(self, text):
if is_py3:
Expand Down Expand Up @@ -130,10 +127,7 @@ def close(self):
"""

self._connected = False
if is_py3:
self.buf = b''
else:
self.buf = ''
self.buf = b''

if self.sock is not None:
self.sock.close()
Expand Down
7 changes: 2 additions & 5 deletions aprslib/parsing.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,14 +82,11 @@ def parse(packet):
raise TypeError("Epected packet to be str/unicode/bytes, got %s", type(packet))

# attempt to detect encoding
if isinstance(packet, bytes if is_py3 else str):
if isinstance(packet, bytes):
try:
packet = packet.decode('utf-8')
except UnicodeDecodeError:
if is_py3:
res = chardet.detect(packet.split(b':', 1)[-1])
else:
res = chardet.detect(packet.split(':', 1)[-1])
res = chardet.detect(packet.split(b':', 1)[-1])

if res['confidence'] > 0.7:
packet = packet.decode(res['encoding'])
Expand Down
42 changes: 18 additions & 24 deletions tests/test_IS.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,6 @@


# byte shim for testing in both py2 and py3
def _b(text):
if sys.version_info[0] >= 3:
return text.encode('latin-1')
else:
return text


class TC_IS(unittest.TestCase):
def setUp(self):
Expand All @@ -25,7 +19,7 @@ def tearDown(self):

def test_initilization(self):
self.assertFalse(self.ais._connected)
self.assertEqual(self.ais.buf, _b(''))
self.assertEqual(self.ais.buf, b'')
self.assertIsNone(self.ais.sock)
self.assertEqual(self.ais.callsign, "LZ1DEV-99")
self.assertEqual(self.ais.passwd, "testpwd")
Expand All @@ -41,7 +35,7 @@ def test_close(self):

mox.Verify(self.ais.sock)
self.assertFalse(self.ais._connected)
self.assertEqual(self.ais.buf, _b(''))
self.assertEqual(self.ais.buf, b'')

def test_open_socket(self):
with self.assertRaises(socket.error):
Expand All @@ -60,7 +54,7 @@ def test_socket_readlines(self):
# part 2 - conn drop trying to recv
self.ais.sock.setblocking(0)
self.ais.sock.fileno().AndReturn(fdr)
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b(''))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b'')
# part 3 - nothing to read
self.ais.sock.setblocking(0)
self.ais.sock.fileno().AndReturn(fdr)
Expand All @@ -69,16 +63,16 @@ def test_socket_readlines(self):
# part 4 - yield 3 lines (blocking False)
self.ais.sock.setblocking(0)
self.ais.sock.fileno().AndReturn(fdr)
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("a\r\n"*3))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"a\r\n"*3)
self.ais.sock.fileno().AndReturn(fdr)
self.ais.sock.recv(mox.IgnoreArg()).AndRaise(
socket.error("Resource temporarily unavailable"))
# part 5 - yield 3 lines 2 times (blocking True)
self.ais.sock.setblocking(0)
self.ais.sock.fileno().AndReturn(fdr)
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("b\r\n"*3))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"b\r\n"*3)
self.ais.sock.fileno().AndReturn(fdr)
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("b\r\n"*3))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"b\r\n"*3)
self.ais.sock.fileno().AndReturn(fdr)
self.ais.sock.recv(mox.IgnoreArg()).AndRaise(StopIteration)
mox.Replay(self.ais.sock)
Expand All @@ -96,10 +90,10 @@ def test_socket_readlines(self):
getattr(self.ais._socket_readlines(), next_method)()
# part 4
for line in self.ais._socket_readlines():
self.assertEqual(line, _b('a'))
self.assertEqual(line, b'a')
# part 5
for line in self.ais._socket_readlines(blocking=True):
self.assertEqual(line, _b('b'))
self.assertEqual(line, b'b')

mox.Verify(self.ais.sock)

Expand All @@ -110,31 +104,31 @@ def test_send_login(self):
# part 1 - raises
self.ais._sendall(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("invalidreply"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"invalidreply")
self.ais.close()
# part 2 - raises (empty callsign)
self.ais._sendall(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("# logresp verified, xx"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"# logresp verified, xx")
self.ais.close()
# part 3 - raises (callsign doesn't match
self.ais._sendall(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("# logresp NOMATCH verified, xx"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"# logresp NOMATCH verified, xx")
self.ais.close()
# part 4 - raises (unverified, but pass is not -1)
self.ais._sendall(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("# logresp CALL unverified, xx"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"# logresp CALL unverified, xx")
self.ais.close()
# part 5 - normal, receive only
self.ais._sendall(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("# logresp CALL unverified, xx"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"# logresp CALL unverified, xx")
# part 6 - normal, correct pass
self.ais._sendall(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("# logresp CALL verified, xx"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"# logresp CALL verified, xx")
mox.Replay(self.ais.sock)
self.m.ReplayAll()

Expand Down Expand Up @@ -179,7 +173,7 @@ def test_connect(self):
self.ais.sock.setsockopt(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.ais.sock.setsockopt(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.ais.sock.setsockopt(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("junk"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"junk")
self.ais.close()
# part 3 - everything going well
self.ais._open_socket()
Expand All @@ -191,7 +185,7 @@ def test_connect(self):
self.ais.sock.setsockopt(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.ais.sock.setsockopt(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.ais.sock.setsockopt(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(_b("# server banner"))
self.ais.sock.recv(mox.IgnoreArg()).AndReturn(b"# server banner")
mox.Replay(self.ais.sock)
self.m.ReplayAll()

Expand All @@ -215,7 +209,7 @@ def test_filter(self):

self.ais._connected = True
self.ais.sock = mox.MockAnything()
self.ais.sock.sendall(_b("#filter %s\r\n" % testFilter))
self.ais.sock.sendall(b"#filter %s\r\n" % testFilter)
mox.Replay(self.ais.sock)

self.ais.set_filter(testFilter)
Expand Down Expand Up @@ -324,7 +318,7 @@ def test_sendall_passing_to_socket(self):
self.ais.sock = mox.MockAnything()
self.ais.sock.setblocking(mox.IgnoreArg())
self.ais.sock.settimeout(mox.IgnoreArg())
self.ais._sendall(_b("%s\r\n" % str(line).rstrip('\r\n'))).AndReturn(None)
self.ais._sendall(b"%s\r\n" % str(line).rstrip('\r\n')).AndReturn(None)
mox.Replay(self.ais.sock)

self.ais.sendall(line)
Expand Down
24 changes: 12 additions & 12 deletions tests/test_parse.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,39 +9,39 @@
from aprslib.exceptions import ParseError, UnknownFormat


class ParseTestCase(unittest.TestCase):
def test_unicode(self):
def _u(text, c='utf8'):
if sys.version_info[0] >= 3:
return text
else:
return text.decode(c)
def _u(text, c='utf8'):
if sys.version_info[0] >= 3:
return text
else:
return text.decode(c)


class ParseTestCase(unittest.TestCase):
def test_unicode(self):
_unicode = str if sys.version_info[0] >= 3 else unicode

# 7bit ascii
result = parse("A>B:>status")

self.assertIsInstance(result['status'],_unicode)
self.assertIsInstance(result['status'], _unicode)
self.assertEqual(result['status'], _u("status"))

# string with degree sign
result = parse("A>B:>status\xb0")

self.assertIsInstance(result['status'],_unicode)
self.assertEqual(result['status'], _u("status\xb0",'latin-1'))
self.assertIsInstance(result['status'], _unicode)
self.assertEqual(result['status'], _u("status\xb0", 'latin-1'))

# str with utf8
result = parse("A>B:>ัั‚ะฐั‚ัƒั")

self.assertIsInstance(result['status'],_unicode)
self.assertIsInstance(result['status'], _unicode)
self.assertEqual(result['status'], _u("ัั‚ะฐั‚ัƒั"))

# unicode input
result = parse(_u("A>B:>ัั‚ะฐั‚ัƒั"))

self.assertIsInstance(result['status'],_unicode)
self.assertIsInstance(result['status'], _unicode)
self.assertEqual(result['status'], _u("ัั‚ะฐั‚ัƒั"))

def test_empty_packet(self):
Expand Down

0 comments on commit 37a8f29

Please sign in to comment.