Cleanups, tests, and no-cover directives for code sections we can't test.

netlib/odict.py

@@ -96,16 +96,6 @@ class ODict:
     def items(self):
         return self.lst[:]
 
-    def _get_state(self):
-        return [tuple(i) for i in self.lst]
-
-    def _load_state(self, state):
-        self.list = [list(i) for i in state]
-
-    @classmethod
-    def _from_state(klass, state):
-        return klass([list(i) for i in state])
-
     def copy(self):
         """
             Returns a copy of this object.

netlib/tcp.py

@@ -2,6 +2,8 @@ import select, socket, threading, sys, time, traceback
 from OpenSSL import SSL
 import certutils
 
+EINTR = 4
+
 SSLv2_METHOD = SSL.SSLv2_METHOD
 SSLv3_METHOD = SSL.SSLv3_METHOD
 SSLv23_METHOD = SSL.SSLv23_METHOD
@@ -238,7 +240,7 @@ class SocketCloseMixin(object):
             #Section 4.2.2.13 of RFC 1122 tells us that a close() with any
             # pending readable data could lead to an immediate RST being sent.
             #http://ia600609.us.archive.org/22/items/TheUltimateSo_lingerPageOrWhyIsMyTcpNotReliable/the-ultimate-so_linger-page-or-why-is-my-tcp-not-reliable.html
-            while self.connection.recv(4096):
+            while self.connection.recv(4096): # pragma: no cover
                 pass
             self.connection.close()
         except (socket.error, SSL.Error, IOError):
@@ -420,8 +422,8 @@ class TCPServer(object):
             while not self.__shutdown_request:
                 try:
                     r, w, e = select.select([self.socket], [], [], poll_interval)
-                except select.error, ex:
-                        if ex[0] == 4:
+                except select.error, ex: # pragma: no cover
+                        if ex[0] == EINTR:
                             continue
                         else:
                             raise  

test/test_odict.py

@@ -41,14 +41,6 @@ class TestODict:
         assert h.match_re("two: due")
         assert not h.match_re("nonono")
 
-    def test_getset_state(self):
-        self.od.add("foo", 1)
-        self.od.add("foo", 2)
-        self.od.add("bar", 3)
-        state = self.od._get_state()
-        nd = odict.ODict._from_state(state)
-        assert nd == self.od
-
     def test_in_any(self):
         self.od["one"] = ["atwoa", "athreea"]
         assert self.od.in_any("one", "two")

test/test_tcp.py

@@ -2,6 +2,7 @@ import cStringIO, Queue, time, socket, random
 from netlib import tcp, certutils, test
 import mock
 import tutils
+from OpenSSL import SSL
 
 class SNIHandler(tcp.BaseHandler):
     sni = None
@@ -435,3 +436,22 @@ class TestFileLike:
         s.readline()
         assert s.first_byte_timestamp == expected
 
+    def test_read_ssl_error(self):
+        s = cStringIO.StringIO("foobar\nfoobar")
+        s = mock.MagicMock()
+        s.read = mock.MagicMock(side_effect=SSL.Error())
+        s = tcp.Reader(s)
+        tutils.raises(tcp.NetLibSSLError, s.read, 1)
+
+
+
+class TestAddress:
+    def test_simple(self):
+        a = tcp.Address("localhost", True)
+        assert a.use_ipv6
+        b = tcp.Address("foo.com", True)
+        assert not a == b
+        c = tcp.Address("localhost", True)
+        assert a == c
+
+