Clean up and strip down netstrings module.

libmproxy/netstring.py

@@ -10,13 +10,6 @@ def header(data):
     return str(len(data))+":"
     
     
-def encode(data):
-    if not isinstance(data, str):
-        raise ValueError("data should be of type 'str'")
-    
-    return "%i:%s," % (len(data), data)
-
-
 class FileEncoder(object):
     def __init__(self, file_out):
         """"
@@ -39,43 +32,16 @@ class FileEncoder(object):
         return self
         
     
-def netstrings_to_file(file_out, data_container):
-    """
-        Writes a container of netstrings to a file.
-        
-        file_out -- A writeable file-like object
-        data_container -- An iterable of strings
-    """
-    write = file_out.write    
-    for s in data_container:
-        if not isinstance(s, str):
-            raise ValueError("data should be of type 'str'")
-        write(header(s))
-        write(s)
-        write(',')
-            
-    
-def encode_netstrings(data_container):
-    """
-        Encodes a number of strings as sequence of netstrings.
-        
-        data_container -- An iterable of strings to be encoded
-    """    
-    return "".join(encode(s) for s in data_container)
-    
-
 class DecoderError(Exception):
-    (
-        PRECEDING_ZERO_IN_SIZE,
-        MAX_SIZE_REACHED,
-        ILLEGAL_DIGIT_IN_SIZE,
-        ILLEGAL_DIGIT
-    ) = range(4)
+    PRECEDING_ZERO_IN_SIZE = 0
+    MAX_SIZE_REACHED = 1
+    ILLEGAL_DIGIT_IN_SIZE = 2
+    ILLEGAL_DIGIT = 3
     error_text = {    
-        PRECEDING_ZERO_IN_SIZE:"PRECEDING_ZERO_IN_SIZE",
-        MAX_SIZE_REACHED:"MAX_SIZE_REACHED",
-        ILLEGAL_DIGIT_IN_SIZE:"ILLEGAL_DIGIT_IN_SIZE",
-        ILLEGAL_DIGIT:"ILLEGAL_DIGIT"
+        PRECEDING_ZERO_IN_SIZE: "PRECEDING_ZERO_IN_SIZE",
+        MAX_SIZE_REACHED: "MAX_SIZE_REACHED",
+        ILLEGAL_DIGIT_IN_SIZE: "ILLEGAL_DIGIT_IN_SIZE",
+        ILLEGAL_DIGIT: "ILLEGAL_DIGIT"
     }
     def __init__(self, code, text):
         Exception.__init__(self)
@@ -109,34 +75,6 @@ class Decoder(object):
         self.data_out = StringIO()
         self.yield_data = ""
         
-    def __str__(self):
-        if self.data_size is None:
-            bytes = len(self.size_string)
-        else:
-            bytes = self.data_out.tell()
-        return "<netstring decoder, %i bytes in buffer>"%bytes
-        
-    def peek_buffer(self):
-        """
-            Returns any bytes not used by decoder.
-        """
-        return self.data_out.getvalue()            
-            
-    def reset(self):
-        """
-            Resets decoder to initial state, and discards any cached stream data.
-        """
-        self.data_pos = 0
-        self.string_start = 0          
-        self.expecting_terminator = False
-        self.size_string = ""
-        self.data_size = None
-        self.remaining_bytes = 0                
-        self.yield_data = ""
-        
-        self.data_out.reset()
-        self.data_out.truncate()
-        
     def feed(self, data):
         """
             A generator that yields 0 or more strings from the given data.
@@ -200,19 +138,6 @@ class Decoder(object):
                     self.expecting_terminator = True                            
         
         
-def decode(data):
-    """
-        Decodes netstrings and returns a tuple containing a
-        list of strings, and any remaining data.
-        
-        data -- A string containing netstring data
-    """
-    decoder = Decoder()
-    netstrings = list(decoder.feed(data))
-    remaining = data[decoder.string_start:]
-    return netstrings, remaining
-        
-        
 def decode_file(file_in, buffer_size=1024):
     """
         Generates 0 or more strings from a netstring file.

libmproxy/utils.py

@@ -342,8 +342,10 @@ def dummy_ca(path):
         stdout=subprocess.PIPE,
         stdin=subprocess.PIPE
     )
+    # begin nocover
     if ret:
         return False
+    # end nocover
     else:
         return True
 
@@ -382,8 +384,7 @@ def dummy_cert(certdir, ca, commonname):
             stdout=subprocess.PIPE,
             stdin=subprocess.PIPE
         )
-        if ret:
-            return None
+        if ret: return None
         cmd = [
             "openssl",
             "x509",
@@ -402,8 +403,7 @@ def dummy_cert(certdir, ca, commonname):
             stdout=subprocess.PIPE,
             stdin=subprocess.PIPE
         )
-        if ret:
-            return None
+        if ret: return None
     else:
         # Create a new selfsigned certificate + key
         cmd = [
@@ -424,8 +424,7 @@ def dummy_cert(certdir, ca, commonname):
             stdout=subprocess.PIPE,
             stdin=subprocess.PIPE
         )
-        if ret:
-            return None
+        if ret: return None
     return certpath
 
 
@@ -437,5 +436,3 @@ def mkdir_p(path):
             pass
         else:
             raise
-
-

test/test_netstring.py

@@ -16,13 +16,6 @@ class uNetstring(libpry.AutoTree):
         for test, result in tests:
             assert netstring.header(test) == result 
             
-    def test_encode(self):
-        tests = [ ("netstring", "9:netstring,"),
-                  ("Will McGugan", "12:Will McGugan,"),
-                  ("", "0:,") ]
-        for test, result in tests:                
-            assert netstring.encode(test) == result 
-            
     def test_file_encoder(self):
         file_out = StringIO()
         data = self.test_data.split()