mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 08:11:00 +00:00
Split pathod and pathoc daemon test suites out.
This commit is contained in:
parent
a68b4498b8
commit
8f148b6b14
@ -22,6 +22,9 @@ class Pathoc(tcp.TCPClient):
|
|||||||
raise PathocError(v)
|
raise PathocError(v)
|
||||||
|
|
||||||
def request(self, spec):
|
def request(self, spec):
|
||||||
|
"""
|
||||||
|
Return an (httpversion, code, msg, headers, content) tuple.
|
||||||
|
"""
|
||||||
r = rparse.parse_request({}, spec)
|
r = rparse.parse_request({}, spec)
|
||||||
r.serve(self.wfile)
|
r.serve(self.wfile)
|
||||||
self.wfile.flush()
|
self.wfile.flush()
|
||||||
|
27
test/test_pathoc.py
Normal file
27
test/test_pathoc.py
Normal file
@ -0,0 +1,27 @@
|
|||||||
|
import json
|
||||||
|
from libpathod import pathoc, test, version
|
||||||
|
import tutils
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
class TestDaemon:
|
||||||
|
@classmethod
|
||||||
|
def setUpAll(self):
|
||||||
|
self.d = test.Daemon(
|
||||||
|
staticdir=tutils.test_data.path("data"),
|
||||||
|
anchors=[("/anchor/.*", "202")]
|
||||||
|
)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def tearDownAll(self):
|
||||||
|
self.d.shutdown()
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.d.clear_log()
|
||||||
|
|
||||||
|
def test_info(self):
|
||||||
|
c = pathoc.Pathoc("127.0.0.1", self.d.port)
|
||||||
|
c.connect()
|
||||||
|
_, _, _, _, content = c.request("get:/api/info")
|
||||||
|
assert tuple(json.loads(content)["version"]) == version.IVERSION
|
||||||
|
|
@ -1,4 +1,5 @@
|
|||||||
from libpathod import pathod
|
import requests
|
||||||
|
from libpathod import pathod, test, version
|
||||||
import tutils
|
import tutils
|
||||||
|
|
||||||
class _TestApplication:
|
class _TestApplication:
|
||||||
@ -36,3 +37,51 @@ class TestPathod:
|
|||||||
for i in range(p.LOGBUF + 1):
|
for i in range(p.LOGBUF + 1):
|
||||||
p.add_log(dict(s="foo"))
|
p.add_log(dict(s="foo"))
|
||||||
assert len(p.get_log()) <= p.LOGBUF
|
assert len(p.get_log()) <= p.LOGBUF
|
||||||
|
|
||||||
|
|
||||||
|
class TestDaemon:
|
||||||
|
@classmethod
|
||||||
|
def setUpAll(self):
|
||||||
|
self.d = test.Daemon(
|
||||||
|
staticdir=tutils.test_data.path("data"),
|
||||||
|
anchors=[("/anchor/.*", "202")]
|
||||||
|
)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def tearDownAll(self):
|
||||||
|
self.d.shutdown()
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.d.clear_log()
|
||||||
|
|
||||||
|
def getpath(self, path):
|
||||||
|
return requests.get("http://localhost:%s/%s"%(self.d.port, path))
|
||||||
|
|
||||||
|
def get(self, spec):
|
||||||
|
return requests.get("http://localhost:%s/p/%s"%(self.d.port, spec))
|
||||||
|
|
||||||
|
def test_info(self):
|
||||||
|
assert tuple(self.d.info()["version"]) == version.IVERSION
|
||||||
|
|
||||||
|
def test_logs(self):
|
||||||
|
rsp = self.get("202")
|
||||||
|
assert len(self.d.log()) == 1
|
||||||
|
assert self.d.clear_log()
|
||||||
|
assert len(self.d.log()) == 0
|
||||||
|
|
||||||
|
def test_disconnect(self):
|
||||||
|
rsp = self.get("202:b@100k:d200")
|
||||||
|
assert len(rsp.content) < 200
|
||||||
|
|
||||||
|
def test_parserr(self):
|
||||||
|
rsp = self.get("400:msg,b:")
|
||||||
|
assert rsp.status_code == 800
|
||||||
|
|
||||||
|
def test_static(self):
|
||||||
|
rsp = self.get("200:b<file")
|
||||||
|
assert rsp.status_code == 200
|
||||||
|
assert rsp.content.strip() == "testfile"
|
||||||
|
|
||||||
|
def test_anchor(self):
|
||||||
|
rsp = self.getpath("anchor/foo")
|
||||||
|
assert rsp.status_code == 202
|
||||||
|
@ -1,6 +1,6 @@
|
|||||||
import time, logging
|
import time, logging
|
||||||
import requests
|
import requests
|
||||||
from libpathod import test, version, utils
|
from libpathod import test, utils
|
||||||
import tutils
|
import tutils
|
||||||
|
|
||||||
logging.disable(logging.CRITICAL)
|
logging.disable(logging.CRITICAL)
|
||||||
@ -35,49 +35,3 @@ class TestDaemonManual:
|
|||||||
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||||
|
|
||||||
|
|
||||||
class TestDaemon:
|
|
||||||
@classmethod
|
|
||||||
def setUpAll(self):
|
|
||||||
self.d = test.Daemon(
|
|
||||||
staticdir=tutils.test_data.path("data"),
|
|
||||||
anchors=[("/anchor/.*", "202")]
|
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def tearDownAll(self):
|
|
||||||
self.d.shutdown()
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.d.clear_log()
|
|
||||||
|
|
||||||
def getpath(self, path):
|
|
||||||
return requests.get("http://localhost:%s/%s"%(self.d.port, path))
|
|
||||||
|
|
||||||
def get(self, spec):
|
|
||||||
return requests.get("http://localhost:%s/p/%s"%(self.d.port, spec))
|
|
||||||
|
|
||||||
def test_info(self):
|
|
||||||
assert tuple(self.d.info()["version"]) == version.IVERSION
|
|
||||||
|
|
||||||
def test_logs(self):
|
|
||||||
rsp = self.get("202")
|
|
||||||
assert len(self.d.log()) == 1
|
|
||||||
assert self.d.clear_log()
|
|
||||||
assert len(self.d.log()) == 0
|
|
||||||
|
|
||||||
def test_disconnect(self):
|
|
||||||
rsp = self.get("202:b@100k:d200")
|
|
||||||
assert len(rsp.content) < 200
|
|
||||||
|
|
||||||
def test_parserr(self):
|
|
||||||
rsp = self.get("400:msg,b:")
|
|
||||||
assert rsp.status_code == 800
|
|
||||||
|
|
||||||
def test_static(self):
|
|
||||||
rsp = self.get("200:b<file")
|
|
||||||
assert rsp.status_code == 200
|
|
||||||
assert rsp.content.strip() == "testfile"
|
|
||||||
|
|
||||||
def test_anchor(self):
|
|
||||||
rsp = self.getpath("anchor/foo")
|
|
||||||
assert rsp.status_code == 202
|
|
||||||
|
Loading…
Reference in New Issue
Block a user