py3++: cStringIO

examples/upsidedownternet.py

@@ -1,4 +1,4 @@
-import cStringIO
+from six.moves import cStringIO as StringIO
 from PIL import Image
 from mitmproxy.models import decoded
 
@@ -7,9 +7,9 @@ def response(context, flow):
     if flow.response.headers.get("content-type", "").startswith("image"):
         with decoded(flow.response):  # automatically decode gzipped responses.
             try:
-                s = cStringIO.StringIO(flow.response.content)
+                s = StringIO(flow.response.content)
                 img = Image.open(s).rotate(180)
-                s2 = cStringIO.StringIO()
+                s2 = StringIO()
                 img.save(s2, "png")
                 flow.response.content = s2.getvalue()
                 flow.response.headers["content-type"] = "image/png"

pathod/app.py

@@ -1,6 +1,6 @@
 import logging
 import pprint
-import cStringIO
+from six.moves import cStringIO as StringIO
 import copy
 from flask import Flask, jsonify, render_template, request, abort, make_response
 from . import version, language, utils
@@ -145,7 +145,7 @@ def make_app(noapi, debug):
             args["marked"] = v.marked()
             return render(template, False, **args)
 
-        s = cStringIO.StringIO()
+        s = StringIO()
 
         settings = copy.copy(app.config["pathod"].settings)
         settings.request_host = EXAMPLE_HOST

pathod/test.py

@@ -1,4 +1,4 @@
-import cStringIO
+from six.moves import cStringIO as StringIO
 import threading
 import Queue
 
@@ -14,7 +14,7 @@ class Daemon:
 
     def __init__(self, ssl=None, **daemonargs):
         self.q = Queue.Queue()
-        self.logfp = cStringIO.StringIO()
+        self.logfp = StringIO()
         daemonargs["logfp"] = self.logfp
         self.thread = _PaThread(self.IFACE, self.q, ssl, daemonargs)
         self.thread.start()

test/mitmproxy/test_dump.py

@@ -1,5 +1,5 @@
 import os
-from cStringIO import StringIO
+from six.moves import cStringIO as StringIO
 from mitmproxy.exceptions import ContentViewException
 from mitmproxy.models import HTTPResponse
 

test/mitmproxy/test_filt.py

@@ -1,4 +1,4 @@
-import cStringIO
+from six.moves import cStringIO as StringIO
 from mitmproxy import filt
 from mitmproxy.models import Error
 from mitmproxy.models import http
@@ -9,7 +9,7 @@ from . import tutils
 class TestParsing:
 
     def _dump(self, x):
-        c = cStringIO.StringIO()
+        c = StringIO()
         x.dump(fp=c)
         assert c.getvalue()
 

test/mitmproxy/test_flow.py

@@ -1,7 +1,7 @@
 import Queue
 import time
 import os.path
-from cStringIO import StringIO
+from six.moves import cStringIO as StringIO
 import email.utils
 
 import mock

test/pathod/test_language_actions.py

@@ -1,4 +1,4 @@
-import cStringIO
+from six.moves import cStringIO as StringIO
 
 from pathod.language import actions
 from pathod import language
@@ -61,7 +61,7 @@ class TestInject:
         assert v.offset == "r"
 
     def test_serve(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = language.parse_pathod("400:i0,'foo'").next()
         assert language.serve(r, s, {})
 

test/pathod/test_language_http.py

@@ -1,5 +1,4 @@
-import cStringIO
-
+from six.moves import cStringIO as StringIO
 from pathod import language
 from pathod.language import http, base
 import tutils
@@ -10,7 +9,7 @@ def parse_request(s):
 
 
 def test_make_error_response():
-    d = cStringIO.StringIO()
+    d = StringIO()
     s = http.make_error_response("foo")
     language.serve(s, d, {})
 
@@ -76,7 +75,7 @@ class TestRequest:
         assert r[0].values({})
 
     def test_render(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_request("GET:'/foo'")
         assert language.serve(
             r,
@@ -163,7 +162,7 @@ class TestResponse:
         assert "OK" in [i[:] for i in r.preamble({})]
 
     def test_render(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = language.parse_pathod("400:m'msg'").next()
         assert language.serve(r, s, {})
 
@@ -173,13 +172,13 @@ class TestResponse:
         assert "p0" not in s.spec()
 
     def test_raw(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = language.parse_pathod("400:b'foo'").next()
         language.serve(r, s, {})
         v = s.getvalue()
         assert "Content-Length" in v
 
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = language.parse_pathod("400:b'foo':r").next()
         language.serve(r, s, {})
         v = s.getvalue()
@@ -187,7 +186,7 @@ class TestResponse:
 
     def test_length(self):
         def testlen(x):
-            s = cStringIO.StringIO()
+            s = StringIO()
             x = x.next()
             language.serve(x, s, language.Settings())
             assert x.length(language.Settings()) == len(s.getvalue())
@@ -198,7 +197,7 @@ class TestResponse:
     def test_maximum_length(self):
         def testlen(x):
             x = x.next()
-            s = cStringIO.StringIO()
+            s = StringIO()
             m = x.maximum_length({})
             language.serve(x, s, {})
             assert m >= len(s.getvalue())

test/pathod/test_language_http2.py

@@ -1,4 +1,4 @@
-import cStringIO
+from six.moves import cStringIO as StringIO
 
 import netlib
 from netlib import tcp
@@ -25,7 +25,7 @@ def default_settings():
 
 
 def test_make_error_response():
-    d = cStringIO.StringIO()
+    d = StringIO()
     s = http2.make_error_response("foo", "bar")
     language.serve(s, d, default_settings())
 
@@ -84,7 +84,7 @@ class TestRequest:
         assert r[1].method.string() == "GET"
 
     def test_render_simple(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_request("GET:'/foo'")
         assert language.serve(
             r,
@@ -126,7 +126,7 @@ class TestRequest:
         assert r.headers[0].values(default_settings()) == ("user-agent", user_agents.get_by_shortcut('a')[2])
 
     def test_render_with_headers(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_request('GET:/foo:h"foo"="bar"')
         assert language.serve(
             r,
@@ -143,7 +143,7 @@ class TestRequest:
 
 
     def test_render_with_body(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_request("GET:'/foo':bfoobar")
         assert language.serve(
             r,
@@ -200,7 +200,7 @@ class TestResponse:
         assert r.body.string() == "foobar"
 
     def test_render_simple(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_response('200')
         assert language.serve(
             r,
@@ -209,7 +209,7 @@ class TestResponse:
         )
 
     def test_render_with_headers(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_response('200:h"foo"="bar"')
         assert language.serve(
             r,
@@ -218,7 +218,7 @@ class TestResponse:
         )
 
     def test_render_with_body(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         r = parse_response('200:bfoobar')
         assert language.serve(
             r,

test/pathod/test_language_writer.py

@@ -1,5 +1,4 @@
-import cStringIO
-
+from six.moves import cStringIO as StringIO
 from pathod import language
 from pathod.language import writer
 
@@ -7,12 +6,12 @@ from pathod.language import writer
 def test_send_chunk():
     v = "foobarfoobar"
     for bs in range(1, len(v) + 2):
-        s = cStringIO.StringIO()
+        s = StringIO()
         writer.send_chunk(s, v, bs, 0, len(v))
         assert s.getvalue() == v
         for start in range(len(v)):
             for end in range(len(v)):
-                s = cStringIO.StringIO()
+                s = StringIO()
                 writer.send_chunk(s, v, bs, start, end)
                 assert s.getvalue() == v[start:end]
 
@@ -20,21 +19,21 @@ def test_send_chunk():
 def test_write_values_inject():
     tst = "foo"
 
-    s = cStringIO.StringIO()
+    s = StringIO()
     writer.write_values(s, [tst], [(0, "inject", "aaa")], blocksize=5)
     assert s.getvalue() == "aaafoo"
 
-    s = cStringIO.StringIO()
+    s = StringIO()
     writer.write_values(s, [tst], [(1, "inject", "aaa")], blocksize=5)
     assert s.getvalue() == "faaaoo"
 
-    s = cStringIO.StringIO()
+    s = StringIO()
     writer.write_values(s, [tst], [(1, "inject", "aaa")], blocksize=5)
     assert s.getvalue() == "faaaoo"
 
 
 def test_write_values_disconnects():
-    s = cStringIO.StringIO()
+    s = StringIO()
     tst = "foo" * 100
     writer.write_values(s, [tst], [(0, "disconnect")], blocksize=5)
     assert not s.getvalue()
@@ -42,13 +41,13 @@ def test_write_values_disconnects():
 
 def test_write_values():
     tst = "foobarvoing"
-    s = cStringIO.StringIO()
+    s = StringIO()
     writer.write_values(s, [tst], [])
     assert s.getvalue() == tst
 
     for bs in range(1, len(tst) + 2):
         for off in range(len(tst)):
-            s = cStringIO.StringIO()
+            s = StringIO()
             writer.write_values(
                 s, [tst], [(off, "disconnect")], blocksize=bs
             )
@@ -58,34 +57,34 @@ def test_write_values():
 def test_write_values_pauses():
     tst = "".join(str(i) for i in range(10))
     for i in range(2, 10):
-        s = cStringIO.StringIO()
+        s = StringIO()
         writer.write_values(
             s, [tst], [(2, "pause", 0), (1, "pause", 0)], blocksize=i
         )
         assert s.getvalue() == tst
 
     for i in range(2, 10):
-        s = cStringIO.StringIO()
+        s = StringIO()
         writer.write_values(s, [tst], [(1, "pause", 0)], blocksize=i)
         assert s.getvalue() == tst
 
     tst = ["".join(str(i) for i in range(10))] * 5
     for i in range(2, 10):
-        s = cStringIO.StringIO()
+        s = StringIO()
         writer.write_values(s, tst[:], [(1, "pause", 0)], blocksize=i)
         assert s.getvalue() == "".join(tst)
 
 
 def test_write_values_after():
-    s = cStringIO.StringIO()
+    s = StringIO()
     r = language.parse_pathod("400:da").next()
     language.serve(r, s, {})
 
-    s = cStringIO.StringIO()
+    s = StringIO()
     r = language.parse_pathod("400:pa,0").next()
     language.serve(r, s, {})
 
-    s = cStringIO.StringIO()
+    s = StringIO()
     r = language.parse_pathod("400:ia,'xx'").next()
     language.serve(r, s, {})
     assert s.getvalue().endswith('xx')

test/pathod/test_pathoc.py

@@ -1,5 +1,5 @@
 import json
-import cStringIO
+from six.moves import cStringIO as StringIO
 import re
 import OpenSSL
 import pytest
@@ -64,7 +64,7 @@ class _TestDaemon:
         ignoretimeout=None,
         showsummary=True
     ):
-        s = cStringIO.StringIO()
+        s = StringIO()
         c = pathoc.Pathoc(
             ("127.0.0.1", self.d.port),
             ssl=self.ssl,
@@ -129,7 +129,7 @@ class TestDaemonSSL(_TestDaemon):
         assert d["log"][0]["request"]["clientcert"]["keyinfo"]
 
     def test_http2_without_ssl(self):
-        fp = cStringIO.StringIO()
+        fp = StringIO()
         c = pathoc.Pathoc(
             ("127.0.0.1", self.d.port),
             use_http2=True,
@@ -227,15 +227,15 @@ class TestDaemon(_TestDaemon):
     def test_connect_fail(self):
         to = ("foobar", 80)
         c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)
-        c.rfile, c.wfile = cStringIO.StringIO(), cStringIO.StringIO()
+        c.rfile, c.wfile = StringIO(), StringIO()
         with raises("connect failed"):
             c.http_connect(to)
-        c.rfile = cStringIO.StringIO(
+        c.rfile = StringIO(
             "HTTP/1.1 500 OK\r\n"
         )
         with raises("connect failed"):
             c.http_connect(to)
-        c.rfile = cStringIO.StringIO(
+        c.rfile = StringIO(
             "HTTP/1.1 200 OK\r\n"
         )
         c.http_connect(to)
@@ -243,7 +243,7 @@ class TestDaemon(_TestDaemon):
     def test_socks_connect(self):
         to = ("foobar", 80)
         c = pathoc.Pathoc(("127.0.0.1", self.d.port), fp=None)
-        c.rfile, c.wfile = tutils.treader(""), cStringIO.StringIO()
+        c.rfile, c.wfile = tutils.treader(""), StringIO()
         tutils.raises(pathoc.PathocError, c.socks_connect, to)
 
         c.rfile = tutils.treader(

test/pathod/test_pathoc_cmdline.py

@@ -1,13 +1,13 @@
 from pathod import pathoc_cmdline as cmdline
 import tutils
-import cStringIO
+from six.moves import cStringIO as StringIO
 import mock
 
 
 @mock.patch("argparse.ArgumentParser.error")
 def test_pathoc(perror):
     assert cmdline.args_pathoc(["pathoc", "foo.com", "get:/"])
-    s = cStringIO.StringIO()
+    s = StringIO()
     with tutils.raises(SystemExit):
         cmdline.args_pathoc(["pathoc", "--show-uas"], s, s)
 

test/pathod/test_pathod.py

@@ -1,6 +1,4 @@
-import sys
-import cStringIO
-import OpenSSL
+from six.moves import cStringIO as StringIO
 import pytest
 
 from pathod import pathod, version
@@ -12,7 +10,7 @@ import tutils
 class TestPathod(object):
 
     def test_logging(self):
-        s = cStringIO.StringIO()
+        s = StringIO()
         p = pathod.Pathod(("127.0.0.1", 0), logfp=s)
         assert len(p.get_log()) == 0
         id = p.add_log(dict(s="foo"))

test/pathod/test_pathod_cmdline.py

@@ -1,6 +1,5 @@
 from pathod import pathod_cmdline as cmdline
 import tutils
-import cStringIO
 import mock
 
 

test/pathod/tutils.py

@@ -1,9 +1,7 @@
 import tempfile
-import os
 import re
 import shutil
-import cStringIO
-from contextlib import contextmanager
+from six.moves import cStringIO as StringIO
 
 import netlib
 from pathod import utils, test, pathoc, pathod, language
@@ -14,7 +12,7 @@ def treader(bytes):
     """
         Construct a tcp.Read object from bytes.
     """
-    fp = cStringIO.StringIO(bytes)
+    fp = StringIO(bytes)
     return tcp.Reader(fp)
 
 
@@ -93,7 +91,7 @@ class DaemonTests(object):
         """
         if ssl is None:
             ssl = self.ssl
-        logfp = cStringIO.StringIO()
+        logfp = StringIO()
         c = pathoc.Pathoc(
             ("localhost", self.d.port),
             ssl=ssl,
@@ -123,6 +121,6 @@ test_data = utils.Data(__name__)
 
 def render(r, settings=language.Settings()):
     r = r.resolve(settings)
-    s = cStringIO.StringIO()
+    s = StringIO()
     assert language.serve(r, s, settings)
     return s.getvalue()