aboutsummaryrefslogtreecommitdiffstats
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/mitmproxy/test_dump.py7
-rw-r--r--test/mitmproxy/test_flow.py6
-rw-r--r--test/mitmproxy/test_flow_export.py218
-rw-r--r--test/mitmproxy/test_server.py6
-rw-r--r--test/netlib/http/http1/test_assemble.py6
5 files changed, 228 insertions, 15 deletions
diff --git a/test/mitmproxy/test_dump.py b/test/mitmproxy/test_dump.py
index 2228a732..7e772881 100644
--- a/test/mitmproxy/test_dump.py
+++ b/test/mitmproxy/test_dump.py
@@ -4,7 +4,6 @@ from mitmproxy.exceptions import ContentViewException
from mitmproxy.models import HTTPResponse
import netlib.tutils
-from netlib.http import CONTENT_MISSING
from mitmproxy import dump, flow
from mitmproxy.proxy import Log
@@ -38,7 +37,7 @@ def test_strfuncs():
flow.request.stickycookie = True
flow.client_conn = mock.MagicMock()
flow.client_conn.address.host = "foo"
- flow.response = netlib.tutils.tresp(content=CONTENT_MISSING)
+ flow.response = netlib.tutils.tresp(content=None)
flow.response.is_replay = True
flow.response.status_code = 300
m.echo_flow(flow)
@@ -104,10 +103,10 @@ class TestDumpMaster:
o = dump.Options(flow_detail=3)
m = dump.DumpMaster(None, o, outfile=cs)
f = tutils.tflow()
- f.request.content = CONTENT_MISSING
+ f.request.content = None
m.handle_request(f)
f.response = HTTPResponse.wrap(netlib.tutils.tresp())
- f.response.content = CONTENT_MISSING
+ f.response.content = None
m.handle_response(f)
assert "content missing" in cs.getvalue()
diff --git a/test/mitmproxy/test_flow.py b/test/mitmproxy/test_flow.py
index 33d2b8f9..2353935b 100644
--- a/test/mitmproxy/test_flow.py
+++ b/test/mitmproxy/test_flow.py
@@ -8,7 +8,7 @@ import mock
import netlib.utils
from netlib import odict
-from netlib.http import CONTENT_MISSING, Headers
+from netlib.http import Headers
from mitmproxy import filt, controller, tnetstring, flow
from mitmproxy.models import Error
from mitmproxy.models import Flow
@@ -465,7 +465,7 @@ class TestFlow(object):
def test_replace_no_content(self):
f = tutils.tflow()
- f.request.content = CONTENT_MISSING
+ f.request.content = None
assert f.replace("foo", "bar") == 0
def test_replace(self):
@@ -751,7 +751,7 @@ class TestFlowMaster:
s = flow.State()
fm = flow.FlowMaster(None, s)
f = tutils.tflow(resp=True)
- f.request.content = CONTENT_MISSING
+ f.request.content = None
assert "missing" in fm.replay_request(f)
f.intercepted = True
diff --git a/test/mitmproxy/test_flow_export.py b/test/mitmproxy/test_flow_export.py
index 62161d5d..75a8090f 100644
--- a/test/mitmproxy/test_flow_export.py
+++ b/test/mitmproxy/test_flow_export.py
@@ -1,3 +1,4 @@
+import json
from textwrap import dedent
import netlib.tutils
@@ -142,7 +143,7 @@ class TestExportPythonCode():
assert flow_export.python_code(flow) == result
-def TestRawRequest():
+class TestRawRequest():
def test_get(self):
flow = tutils.tflow(req=req_get)
@@ -159,9 +160,10 @@ def TestRawRequest():
flow = tutils.tflow(req=req_post)
result = dedent("""
POST /path HTTP/1.1\r
+ content-type: application/json\r
host: address:22\r
\r
- content
+ {"name": "example", "email": "example@example.com"}
""").strip()
assert flow_export.raw_request(flow) == result
@@ -176,3 +178,215 @@ def TestRawRequest():
content
""").strip()
assert flow_export.raw_request(flow) == result
+
+class TestExportLocustCode():
+
+ def test_get(self):
+ flow = tutils.tflow(req=req_get)
+ result = """
+from locust import HttpLocust, TaskSet, task
+
+class UserBehavior(TaskSet):
+ def on_start(self):
+ ''' on_start is called when a Locust start before any task is scheduled '''
+ self.path()
+
+ @task()
+ def path(self):
+ url = self.locust.host + '/path'
+
+ headers = {
+ 'header': 'qvalue',
+ 'content-length': '7',
+ }
+
+ self.response = self.client.request(
+ method='GET',
+ url=url,
+ headers=headers,
+ )
+
+ ### Additional tasks can go here ###
+
+
+class WebsiteUser(HttpLocust):
+ task_set = UserBehavior
+ min_wait = 1000
+ max_wait = 3000
+ """.strip()
+
+ assert flow_export.locust_code(flow) == result
+
+ def test_post(self):
+ req_post.content = '''content'''
+ req_post.headers = ''
+ flow = tutils.tflow(req=req_post)
+ result = """
+from locust import HttpLocust, TaskSet, task
+
+class UserBehavior(TaskSet):
+ def on_start(self):
+ ''' on_start is called when a Locust start before any task is scheduled '''
+ self.path()
+
+ @task()
+ def path(self):
+ url = self.locust.host + '/path'
+
+ data = '''content'''
+
+ self.response = self.client.request(
+ method='POST',
+ url=url,
+ data=data,
+ )
+
+ ### Additional tasks can go here ###
+
+
+class WebsiteUser(HttpLocust):
+ task_set = UserBehavior
+ min_wait = 1000
+ max_wait = 3000
+
+ """.strip()
+
+ assert flow_export.locust_code(flow) == result
+
+
+ def test_patch(self):
+ flow = tutils.tflow(req=req_patch)
+ result = """
+from locust import HttpLocust, TaskSet, task
+
+class UserBehavior(TaskSet):
+ def on_start(self):
+ ''' on_start is called when a Locust start before any task is scheduled '''
+ self.path()
+
+ @task()
+ def path(self):
+ url = self.locust.host + '/path'
+
+ headers = {
+ 'header': 'qvalue',
+ 'content-length': '7',
+ }
+
+ params = {
+ 'query': 'param',
+ }
+
+ data = '''content'''
+
+ self.response = self.client.request(
+ method='PATCH',
+ url=url,
+ headers=headers,
+ params=params,
+ data=data,
+ )
+
+ ### Additional tasks can go here ###
+
+
+class WebsiteUser(HttpLocust):
+ task_set = UserBehavior
+ min_wait = 1000
+ max_wait = 3000
+
+ """.strip()
+
+ assert flow_export.locust_code(flow) == result
+
+
+class TestExportLocustTask():
+
+ def test_get(self):
+ flow = tutils.tflow(req=req_get)
+ result = ' ' + """
+ @task()
+ def path(self):
+ url = self.locust.host + '/path'
+
+ headers = {
+ 'header': 'qvalue',
+ 'content-length': '7',
+ }
+
+ self.response = self.client.request(
+ method='GET',
+ url=url,
+ headers=headers,
+ )
+ """.strip() + '\n'
+
+ assert flow_export.locust_task(flow) == result
+
+ def test_post(self):
+ flow = tutils.tflow(req=req_post)
+ result = ' ' + """
+ @task()
+ def path(self):
+ url = self.locust.host + '/path'
+
+ data = '''content'''
+
+ self.response = self.client.request(
+ method='POST',
+ url=url,
+ data=data,
+ )
+ """.strip() + '\n'
+
+ assert flow_export.locust_task(flow) == result
+
+
+ def test_patch(self):
+ flow = tutils.tflow(req=req_patch)
+ result = ' ' + """
+ @task()
+ def path(self):
+ url = self.locust.host + '/path'
+
+ headers = {
+ 'header': 'qvalue',
+ 'content-length': '7',
+ }
+
+ params = {
+ 'query': 'param',
+ }
+
+ data = '''content'''
+
+ self.response = self.client.request(
+ method='PATCH',
+ url=url,
+ headers=headers,
+ params=params,
+ data=data,
+ )
+ """.strip() + '\n'
+
+ assert flow_export.locust_task(flow) == result
+
+
+class TestIsJson():
+
+ def test_empty(self):
+ assert flow_export.is_json(None, None) == False
+
+ def test_json_type(self):
+ headers = Headers(content_type="application/json")
+ assert flow_export.is_json(headers, "foobar") == False
+
+ def test_valid(self):
+ headers = Headers(content_type="application/foobar")
+ j = flow_export.is_json(headers, '{"name": "example", "email": "example@example.com"}')
+ assert j == False
+
+ def test_valid(self):
+ headers = Headers(content_type="application/json")
+ j = flow_export.is_json(headers, '{"name": "example", "email": "example@example.com"}')
+ assert isinstance(j, dict)
diff --git a/test/mitmproxy/test_server.py b/test/mitmproxy/test_server.py
index 26e53e8a..dc72f032 100644
--- a/test/mitmproxy/test_server.py
+++ b/test/mitmproxy/test_server.py
@@ -8,7 +8,7 @@ from netlib.tcp import Address
import netlib.tutils
from netlib import tcp, http, socks
from netlib.certutils import SSLCert
-from netlib.http import authentication, CONTENT_MISSING, http1
+from netlib.http import authentication, http1
from netlib.tutils import raises
from pathod import pathoc, pathod
@@ -281,7 +281,7 @@ class TestHTTP(tservers.HTTPProxyTest, CommonMixin, AppMixin):
self.pathod("200:b@3k")
assert self.master.state.view[-1].response.stream
- assert self.master.state.view[-1].response.content == CONTENT_MISSING
+ assert self.master.state.view[-1].response.content is None
self.master.set_stream_large_bodies(None)
def test_stream_modify(self):
@@ -816,7 +816,7 @@ class MasterIncomplete(tservers.TestMaster):
def handle_request(self, f):
resp = HTTPResponse.wrap(netlib.tutils.tresp())
- resp.content = CONTENT_MISSING
+ resp.content = None
f.reply(resp)
diff --git a/test/netlib/http/http1/test_assemble.py b/test/netlib/http/http1/test_assemble.py
index 31a62438..8dcbae8e 100644
--- a/test/netlib/http/http1/test_assemble.py
+++ b/test/netlib/http/http1/test_assemble.py
@@ -1,6 +1,6 @@
from __future__ import absolute_import, print_function, division
from netlib.exceptions import HttpException
-from netlib.http import CONTENT_MISSING, Headers
+from netlib.http import Headers
from netlib.http.http1.assemble import (
assemble_request, assemble_request_head, assemble_response,
assemble_response_head, _assemble_request_line, _assemble_request_headers,
@@ -20,7 +20,7 @@ def test_assemble_request():
)
with raises(HttpException):
- assemble_request(treq(content=CONTENT_MISSING))
+ assemble_request(treq(content=None))
def test_assemble_request_head():
@@ -41,7 +41,7 @@ def test_assemble_response():
)
with raises(HttpException):
- assemble_response(tresp(content=CONTENT_MISSING))
+ assemble_response(tresp(content=None))
def test_assemble_response_head():
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-27 16:13:34 +0000