103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
import pytest
|
|
import requests
|
|
import time
|
|
import importlib
|
|
import logging
|
|
import logging.config
|
|
from pytest_httpserver import HTTPServer
|
|
|
|
from pyexecplugins.pyexecplugins import PyExecPlugin
|
|
|
|
logging.config.fileConfig('tests/logging-test.conf')
|
|
pluginstotest = ("Http", "Shell", "Eval", "Mail")
|
|
for plg in pluginstotest:
|
|
path = "pyexecplugins." + plg
|
|
importlib.import_module(path)
|
|
|
|
logger = logging.getLogger("test")
|
|
logger.info("All available plugins are: %s", PyExecPlugin.getPluginNames())
|
|
|
|
def sleeping5(request):
|
|
time.sleep(5)
|
|
|
|
def sleeping15(request):
|
|
time.sleep(15)
|
|
|
|
def test_data(httpserver):
|
|
uri = "/postdata"
|
|
data = {
|
|
"method" : "post",
|
|
"url" : httpserver.url_for(uri),
|
|
"headers" : {
|
|
"Authorization" : "Bearer 9392j9fn9392",
|
|
"Content-Type" : "application/json"
|
|
},
|
|
"body" : {"foo": "bar"},
|
|
"params" : "gino=stilla",
|
|
"expect" : 201,
|
|
"fail" : None,
|
|
"connection-timeout" : 1,
|
|
"read-timeout" : 1,
|
|
"no-timeout" : "false",
|
|
}
|
|
httpserver.expect_request(uri=uri, headers=data["headers"], json=data["body"]).respond_with_data("Created", status=data["expect"])
|
|
http_plugin = PyExecPlugin.get("Http")(data, None)
|
|
out = http_plugin.execute()
|
|
assert out["status"] == 201
|
|
assert out["body"] == "Created"
|
|
|
|
def test_params(httpserver):
|
|
uri = "/params"
|
|
data = {
|
|
"method" : "get",
|
|
"url" : httpserver.url_for(uri),
|
|
"headers" : {
|
|
"Authorization" : "Bearer 9392j9fn9392"
|
|
},
|
|
"params" : "gino=stilla",
|
|
"expect" : 200,
|
|
"fail" : None,
|
|
"connection-timeout" : 1,
|
|
"read-timeout" : 1,
|
|
"no-timeout" : "false",
|
|
}
|
|
httpserver.expect_request(uri=uri, headers=data["headers"], query_string=data["params"]).respond_with_data("OK")
|
|
http_plugin = PyExecPlugin.get("Http")(data, None)
|
|
out = http_plugin.execute()
|
|
assert out["status"] == 200
|
|
assert out["body"] == "OK"
|
|
|
|
def test_timeout(httpserver):
|
|
uri = "/timeout"
|
|
data = {
|
|
"method" : "get",
|
|
"url" : httpserver.url_for(uri),
|
|
"read-timeout" : 3
|
|
}
|
|
httpserver.expect_request(uri).respond_with_handler(sleeping5)
|
|
http_plugin = PyExecPlugin.get("Http")(data, None)
|
|
with pytest.raises(requests.exceptions.ReadTimeout):
|
|
assert http_plugin.execute()
|
|
|
|
# no timeout is tested with a server wait time of 15 seconds that simulates an indefinite wait, that is higher than defaults
|
|
def test_no_timeout(httpserver):
|
|
uri = "/notimeout"
|
|
data = {
|
|
"method" : "get",
|
|
"url" : httpserver.url_for(uri),
|
|
"no-timeout" : True
|
|
}
|
|
httpserver.expect_request(uri).respond_with_handler(sleeping15)
|
|
http_plugin = PyExecPlugin.get("Http")(data, None)
|
|
http_plugin.execute()
|
|
|
|
def test_connection_refused(httpserver):
|
|
httpserver.stop()
|
|
data = {
|
|
"method" : "get",
|
|
"url" : httpserver.url_for("/")
|
|
}
|
|
http_plugin = PyExecPlugin.get("Http")(data, None)
|
|
with pytest.raises(requests.exceptions.ConnectionError):
|
|
http_plugin.execute()
|