From 37a8f29af4dd40850446673b8c1a855ded2c0316 Mon Sep 17 00:00:00 2001 From: Rossen Georgiev Date: Wed, 27 May 2015 23:22:23 +0100 Subject: [PATCH] refined code using bytes type --- aprslib/IS.py | 10 ++-------- aprslib/parsing.py | 7 ++----- tests/test_IS.py | 42 ++++++++++++++++++------------------------ tests/test_parse.py | 24 ++++++++++++------------ 4 files changed, 34 insertions(+), 49 deletions(-) diff --git a/aprslib/IS.py b/aprslib/IS.py index b2f016b..7efa709 100644 --- a/aprslib/IS.py +++ b/aprslib/IS.py @@ -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: @@ -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() diff --git a/aprslib/parsing.py b/aprslib/parsing.py index f4380dd..83d8407 100644 --- a/aprslib/parsing.py +++ b/aprslib/parsing.py @@ -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']) diff --git a/tests/test_IS.py b/tests/test_IS.py index ee258ce..acc2f7d 100644 --- a/tests/test_IS.py +++ b/tests/test_IS.py @@ -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): @@ -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") @@ -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): @@ -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) @@ -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) @@ -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) @@ -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() @@ -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() @@ -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() @@ -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) @@ -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) diff --git a/tests/test_parse.py b/tests/test_parse.py index da2a5f5..c0b7a4c 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -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):