Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-26 10:16:27 +00:00
Code Issues Projects Releases Wiki Activity

nuke tutils.tmpdir, use pytest tmpdir

Browse Source
This commit is contained in:
Thomas Kriechbaumer 2017-03-12 22:55:22 +01:00
parent d069ba9da5
commit 1b045d24bc
13 changed files with 252 additions and 324 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
mitmproxy/test/tutils.py
View File

@ -1,9 +1,5 @@
from io import BytesIO
import tempfile
import os
import time
import shutil
from contextlib import contextmanager
from io import BytesIO
from mitmproxy.utils import data
from mitmproxy.net import tcp
@ -13,18 +9,6 @@ from mitmproxy.net import http
test_data = data.Data(__name__).push("../../test/")
@contextmanager
def tmpdir(*args, **kwargs):
orig_workdir = os.getcwd()
temp_workdir = tempfile.mkdtemp(*args, **kwargs)
os.chdir(temp_workdir)
yield temp_workdir
os.chdir(orig_workdir)
shutil.rmtree(temp_workdir)
def treader(bytes):
"""
Construct a tcp.Read object from bytes.

7
test/mitmproxy/addons/test_clientplayback.py
View File

@ -1,9 +1,7 @@
import os
import pytest
from unittest import mock
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy import io
from mitmproxy import exceptions
@ -49,11 +47,10 @@ class TestClientPlayback:
cp.tick()
assert cp.current_thread is None
def test_configure(self):
def test_configure(self, tmpdir):
cp = clientplayback.ClientPlayback()
with taddons.context() as tctx:
with tutils.tmpdir() as td:
path = os.path.join(td, "flows")
path = str(tmpdir.join("flows"))
tdump(path, [tflow.tflow()])
tctx.configure(cp, client_replay=[path])
tctx.configure(cp, client_replay=[])

16
test/mitmproxy/addons/test_replace.py
View File

@ -1,11 +1,9 @@
import os.path
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from .. import tservers
from mitmproxy.addons import replace
from mitmproxy.test import taddons
from mitmproxy.test import tflow
class TestReplace:
@ -71,18 +69,16 @@ class TestUpstreamProxy(tservers.HTTPUpstreamProxyTest):
class TestReplaceFile:
def test_simple(self):
def test_simple(self, tmpdir):
r = replace.ReplaceFile()
with tutils.tmpdir() as td:
rp = os.path.join(td, "replacement")
with open(rp, "w") as f:
f.write("bar")
rp = tmpdir.join("replacement")
rp.write("bar")
with taddons.context() as tctx:
tctx.configure(
r,
replacement_files = [
"/~q/foo/" + rp,
"/~s/foo/" + rp,
"/~q/foo/" + str(rp),
"/~s/foo/" + str(rp),
"/~b nonexistent/nonexistent/nonexistent",
]
)

15
test/mitmproxy/addons/test_script.py
View File

@ -68,13 +68,12 @@ class TestParseCommand:
with pytest.raises(ValueError):
script.parse_command(" ")
def test_no_script_file(self):
def test_no_script_file(self, tmpdir):
with pytest.raises(Exception, match="not found"):
script.parse_command("notfound")
with tutils.tmpdir() as dir:
with pytest.raises(Exception, match="Not a file"):
script.parse_command(dir)
script.parse_command(str(tmpdir))
def test_parse_args(self):
with utils.chdir(tutils.test_data.dirname):
@ -128,15 +127,13 @@ class TestScript:
recf = sc.ns.call_log[0]
assert recf[1] == "request"
def test_reload(self):
def test_reload(self, tmpdir):
with taddons.context() as tctx:
with tutils.tmpdir():
with open("foo.py", "w"):
pass
sc = script.Script("foo.py")
f = tmpdir.join("foo.py")
f.ensure(file=True)
sc = script.Script(str(f))
tctx.configure(sc)
for _ in range(100):
with open("foo.py", "a") as f:
f.write(".")
sc.tick()
time.sleep(0.1)

11
test/mitmproxy/addons/test_serverplayback.py
View File

@ -1,10 +1,8 @@
import os
import urllib
import pytest
from mitmproxy.test import tutils
from mitmproxy.test import tflow
from mitmproxy.test import taddons
from mitmproxy.test import tflow
import mitmproxy.test.tutils
from mitmproxy.addons import serverplayback
@ -19,15 +17,14 @@ def tdump(path, flows):
w.add(i)
def test_config():
def test_config(tmpdir):
s = serverplayback.ServerPlayback()
with tutils.tmpdir() as p:
with taddons.context() as tctx:
fpath = os.path.join(p, "flows")
fpath = str(tmpdir.join("flows"))
tdump(fpath, [tflow.tflow(resp=True)])
tctx.configure(s, server_replay=[fpath])
with pytest.raises(exceptions.OptionsError):
tctx.configure(s, server_replay=[p])
tctx.configure(s, server_replay=[str(tmpdir)])
def test_tick():

22
test/mitmproxy/addons/test_streamfile.py
View File

@ -1,9 +1,7 @@
import os.path
import pytest
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.test import tflow
from mitmproxy import io
from mitmproxy import exceptions
@ -11,15 +9,13 @@ from mitmproxy import options
from mitmproxy.addons import streamfile
def test_configure():
def test_configure(tmpdir):
sa = streamfile.StreamFile()
with taddons.context(options=options.Options()) as tctx:
with tutils.tmpdir() as tdir:
p = os.path.join(tdir, "foo")
with pytest.raises(exceptions.OptionsError):
tctx.configure(sa, streamfile=tdir)
tctx.configure(sa, streamfile=str(tmpdir))
with pytest.raises(Exception, match="Invalid filter"):
tctx.configure(sa, streamfile=p, filtstr="~~")
tctx.configure(sa, streamfile=str(tmpdir.join("foo")), filtstr="~~")
tctx.configure(sa, filtstr="foo")
assert sa.filt
tctx.configure(sa, filtstr=None)
@ -31,11 +27,10 @@ def rd(p):
return list(x.stream())
def test_tcp():
def test_tcp(tmpdir):
sa = streamfile.StreamFile()
with taddons.context() as tctx:
with tutils.tmpdir() as tdir:
p = os.path.join(tdir, "foo")
p = str(tmpdir.join("foo"))
tctx.configure(sa, streamfile=p)
tt = tflow.ttcpflow()
@ -45,11 +40,10 @@ def test_tcp():
assert rd(p)
def test_simple():
def test_simple(tmpdir):
sa = streamfile.StreamFile()
with taddons.context() as tctx:
with tutils.tmpdir() as tdir:
p = os.path.join(tdir, "foo")
p = str(tmpdir.join("foo"))
tctx.configure(sa, streamfile=p)

7
test/mitmproxy/net/test_tcp.py
View File

@ -11,8 +11,8 @@ from OpenSSL import SSL
from mitmproxy import certs
from mitmproxy.net import tcp
from mitmproxy.test import tutils
from mitmproxy import exceptions
from mitmproxy.test import tutils
from . import tservers
from ...conftest import requires_alpn
@ -783,12 +783,11 @@ class TestSSLKeyLogger(tservers.ServerTestBase):
cipher_list="AES256-SHA"
)
def test_log(self):
def test_log(self, tmpdir):
testval = b"echo!\n"
_logfun = tcp.log_ssl_key
with tutils.tmpdir() as d:
logfile = os.path.join(d, "foo", "bar", "logfile")
logfile = str(tmpdir.join("foo", "bar", "logfile"))
tcp.log_ssl_key = tcp.SSLKeyLogger(logfile)
c = tcp.TCPClient(("127.0.0.1", self.port))

60
test/mitmproxy/test_certs.py
View File

@ -34,24 +34,21 @@ from mitmproxy.test import tutils
class TestCertStore:
def test_create_explicit(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_create_explicit(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
assert ca.get_cert(b"foo", [])
ca2 = certs.CertStore.from_store(d, "test")
ca2 = certs.CertStore.from_store(str(tmpdir), "test")
assert ca2.get_cert(b"foo", [])
assert ca.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()
def test_create_no_common_name(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_create_no_common_name(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
assert ca.get_cert(None, [])[0].cn is None
def test_create_tmp(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_create_tmp(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
assert ca.get_cert(b"foo.com", [])
assert ca.get_cert(b"foo.com", [])
assert ca.get_cert(b"*.foo.com", [])
@ -59,25 +56,22 @@ class TestCertStore:
r = ca.get_cert(b"*.foo.com", [])
assert r[1] == ca.default_privatekey
def test_sans(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_sans(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
c1 = ca.get_cert(b"foo.com", [b"*.bar.com"])
ca.get_cert(b"foo.bar.com", [])
# assert c1 == c2
c3 = ca.get_cert(b"bar.com", [])
assert not c1 == c3
def test_sans_change(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_sans_change(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
ca.get_cert(b"foo.com", [b"*.bar.com"])
cert, key, chain_file = ca.get_cert(b"foo.bar.com", [b"*.baz.com"])
assert b"*.baz.com" in cert.altnames
def test_expire(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_expire(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
ca.STORE_CAP = 3
ca.get_cert(b"one.com", [])
ca.get_cert(b"two.com", [])
@ -100,35 +94,29 @@ class TestCertStore:
assert (b"three.com", ()) in ca.certs
assert (b"four.com", ()) in ca.certs
def test_overrides(self):
with tutils.tmpdir() as d:
ca1 = certs.CertStore.from_store(os.path.join(d, "ca1"), "test")
ca2 = certs.CertStore.from_store(os.path.join(d, "ca2"), "test")
assert not ca1.default_ca.get_serial_number(
) == ca2.default_ca.get_serial_number()
def test_overrides(self, tmpdir):
ca1 = certs.CertStore.from_store(str(tmpdir.join("ca1")), "test")
ca2 = certs.CertStore.from_store(str(tmpdir.join("ca2")), "test")
assert not ca1.default_ca.get_serial_number() == ca2.default_ca.get_serial_number()
dc = ca2.get_cert(b"foo.com", [b"sans.example.com"])
dcp = os.path.join(d, "dc")
f = open(dcp, "wb")
f.write(dc[0].to_pem())
f.close()
ca1.add_cert_file(b"foo.com", dcp)
dcp = tmpdir.join("dc")
dcp.write(dc[0].to_pem())
ca1.add_cert_file(b"foo.com", str(dcp))
ret = ca1.get_cert(b"foo.com", [])
assert ret[0].serial == dc[0].serial
def test_create_dhparams(self):
with tutils.tmpdir() as d:
filename = os.path.join(d, "dhparam.pem")
def test_create_dhparams(self, tmpdir):
filename = str(tmpdir.join("dhparam.pem"))
certs.CertStore.load_dhparam(filename)
assert os.path.exists(filename)
class TestDummyCert:
def test_with_ca(self):
with tutils.tmpdir() as d:
ca = certs.CertStore.from_store(d, "test")
def test_with_ca(self, tmpdir):
ca = certs.CertStore.from_store(str(tmpdir), "test")
r = certs.dummy_cert(
ca.default_privatekey,
ca.default_ca,

19
test/mitmproxy/test_examples.py
View File

@ -1,5 +1,4 @@
import json
import os
import shlex
import pytest
@ -142,9 +141,8 @@ class TestHARDump:
with pytest.raises(ScriptError):
tscript("complex/har_dump.py")
def test_simple(self):
with tutils.tmpdir() as tdir:
path = os.path.join(tdir, "somefile")
def test_simple(self, tmpdir):
path = str(tmpdir.join("somefile"))
m, sc = tscript("complex/har_dump.py", shlex.quote(path))
m.addons.invoke(m, "response", self.flow())
@ -152,12 +150,10 @@ class TestHARDump:
with open(path, "r") as inp:
har = json.load(inp)
assert len(har["log"]["entries"]) == 1
def test_base64(self):
with tutils.tmpdir() as tdir:
path = os.path.join(tdir, "somefile")
def test_base64(self, tmpdir):
path = str(tmpdir.join("somefile"))
m, sc = tscript("complex/har_dump.py", shlex.quote(path))
m.addons.invoke(m, "response", self.flow(resp_content=b"foo" + b"\xFF" * 10))
@ -165,7 +161,6 @@ class TestHARDump:
with open(path, "r") as inp:
har = json.load(inp)
assert har["log"]["entries"][0]["response"]["content"]["encoding"] == "base64"
def test_format_cookies(self):
@ -187,7 +182,7 @@ class TestHARDump:
f = format_cookies([("n", "v", CA([("expires", "Mon, 24-Aug-2037 00:00:00 GMT")]))])[0]
assert f['expires']
def test_binary(self):
def test_binary(self, tmpdir):
f = self.flow()
f.request.method = "POST"
@ -196,8 +191,7 @@ class TestHARDump:
f.response.headers["random-junk"] = bytes(range(256))
f.response.content = bytes(range(256))
with tutils.tmpdir() as tdir:
path = os.path.join(tdir, "somefile")
path = str(tmpdir.join("somefile"))
m, sc = tscript("complex/har_dump.py", shlex.quote(path))
m.addons.invoke(m, "response", f)
@ -205,5 +199,4 @@ class TestHARDump:
with open(path, "r") as inp:
har = json.load(inp)
assert len(har["log"]["entries"]) == 1

7
test/mitmproxy/test_optmanager.py
View File

@ -1,5 +1,4 @@
import copy
import os
import pytest
import typing
import argparse
@ -7,7 +6,6 @@ import argparse
from mitmproxy import options
from mitmproxy import optmanager
from mitmproxy import exceptions
from mitmproxy.test import tutils
class TO(optmanager.OptManager):
@ -238,11 +236,10 @@ def test_serialize_defaults():
assert o.serialize(None, defaults=True)
def test_saving():
def test_saving(tmpdir):
o = TD2()
o.three = "set"
with tutils.tmpdir() as tdir:
dst = os.path.join(tdir, "conf")
dst = str(tmpdir.join("conf"))
o.save(dst, defaults=True)
o2 = TD2()

7
test/mitmproxy/tools/test_dump.py
View File

@ -1,4 +1,3 @@
import os
import pytest
from unittest import mock
@ -9,7 +8,6 @@ from mitmproxy import controller
from mitmproxy import options
from mitmproxy.tools import dump
from mitmproxy.test import tutils
from .. import tservers
@ -19,9 +17,8 @@ class TestDumpMaster(tservers.MasterTest):
m = dump.DumpMaster(o, proxy.DummyServer(), with_termlog=False, with_dumper=False)
return m
def test_read(self):
with tutils.tmpdir() as t:
p = os.path.join(t, "read")
def test_read(self, tmpdir):
p = str(tmpdir.join("read"))
self.flowfile(p)
self.dummy_cycle(
self.mkmaster(None, rfile=p),

15
test/pathod/language/test_base.py
View File

@ -1,11 +1,8 @@
import os
import pytest
from pathod import language
from pathod.language import base, exceptions
from mitmproxy.test import tutils
def parse_request(s):
return language.parse_pathoc(s).next()
@ -137,24 +134,22 @@ class TestTokValueFile:
v = base.TokValue.parseString("<path")[0]
assert v.path == "path"
def test_access_control(self):
def test_access_control(self, tmpdir):
v = base.TokValue.parseString("<path")[0]
with tutils.tmpdir() as t:
p = os.path.join(t, "path")
with open(p, "wb") as f:
f = tmpdir.join("path")
f.write(b"x" * 10000)
assert v.get_generator(language.Settings(staticdir=t))
assert v.get_generator(language.Settings(staticdir=str(tmpdir)))
v = base.TokValue.parseString("<path2")[0]
with pytest.raises(exceptions.FileAccessDenied):
v.get_generator(language.Settings(staticdir=t))
v.get_generator(language.Settings(staticdir=str(tmpdir)))
with pytest.raises(Exception, match="access disabled"):
v.get_generator(language.Settings())
v = base.TokValue.parseString("</outside")[0]
with pytest.raises(Exception, match="outside"):
v.get_generator(language.Settings(staticdir=t))
v.get_generator(language.Settings(staticdir=str(tmpdir)))
def test_spec(self):
v = base.TokValue.parseString("<'one two'")[0]

12
test/pathod/language/test_generators.py
View File

@ -1,7 +1,4 @@
import os
from pathod.language import generators
from mitmproxy.test import tutils
def test_randomgenerator():
@ -15,13 +12,10 @@ def test_randomgenerator():
assert len(g[1000:1001]) == 0
def test_filegenerator():
with tutils.tmpdir() as t:
path = os.path.join(t, "foo")
f = open(path, "wb")
def test_filegenerator(tmpdir):
f = tmpdir.join("foo")
f.write(b"x" * 10000)
f.close()
g = generators.FileGenerator(path)
g = generators.FileGenerator(str(f))
assert len(g) == 10000
assert g[0] == b"x"
assert g[-1] == b"x"