saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Migrate `rest_cherrypy` tests to pytest

Browse source
This commit is contained in:
Pedro Algarvio 2021-10-25 11:42:30 +01:00 • committed by Megan Wilhite
parent 05aa6d5a44
commit 7c905ac7b0
27 changed files with 815 additions and 905 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
salt/netapi/rest_cherrypy/app.py
View file

@ -601,6 +601,7 @@ import salt
import salt.auth
import salt.exceptions
import salt.netapi
import salt.utils.args
import salt.utils.event
import salt.utils.json
import salt.utils.stringutils
@ -976,6 +977,15 @@ def urlencoded_processor(entity):
unserialized_data[key] = val[0]
if len(val) == 0:
unserialized_data[key] = ""
# Parse `arg` and `kwarg` just like we do it on the CLI
if "kwarg" in unserialized_data:
unserialized_data["kwarg"] = salt.utils.args.yamlify_arg(
unserialized_data["kwarg"]
)
if "arg" in unserialized_data:
for idx, value in enumerate(unserialized_data["arg"]):
unserialized_data["arg"][idx] = salt.utils.args.yamlify_arg(value)
cherrypy.serving.request.unserialized_data = unserialized_data

12
tests/filename_map.yml
View file

@ -226,9 +226,17 @@ salt/modules/*_sysctl.py:
- integration.modules.test_sysctl
salt/netapi/rest_cherrypy/*:
- unit.netapi.test_rest_cherrypy
- integration.netapi.rest_cherrypy.test_app_pam
- integration.netapi.test_client
- pytests.functional.netapi.rest_cherrypy.test_auth
- pytests.functional.netapi.rest_cherrypy.test_auth_pam
- pytests.functional.netapi.rest_cherrypy.test_cors
- pytests.functional.netapi.rest_cherrypy.test_in_formats
- pytests.functional.netapi.rest_cherrypy.test_out_formats
- pytests.integration.netapi.rest_cherrypy.test_arg_kwarg
- pytests.integration.netapi.rest_cherrypy.test_auth
- pytests.integration.netapi.rest_cherrypy.test_jobs
- pytests.integration.netapi.rest_cherrypy.test_run
- pytests.integration.netapi.rest_cherrypy.test_webhook_disable_auth
salt/netapi/rest_tornado/*:
- pytests.functional.netapi.rest_tornado.test_auth_handler

374
tests/integration/netapi/rest_cherrypy/test_app.py
View file

@ -1,374 +0,0 @@
import os
import urllib.parse
import pytest
import salt.utils.json
import salt.utils.stringutils
import tests.support.cherrypy_testclasses as cptc
class TestAuth(cptc.BaseRestCherryPyTest):
def test_get_root_noauth(self):
"""
GET requests to the root URL should not require auth
"""
request, response = self.request("/")
self.assertEqual(response.status, "200 OK")
def test_post_root_auth(self):
"""
POST requests to the root URL redirect to login
"""
request, response = self.request("/", method="POST", data={})
self.assertEqual(response.status, "401 Unauthorized")
def test_login_noauth(self):
"""
GET requests to the login URL should not require auth
"""
request, response = self.request("/login")
self.assertEqual(response.status, "200 OK")
def test_webhook_auth(self):
"""
Requests to the webhook URL require auth by default
"""
request, response = self.request("/hook", method="POST", data={})
self.assertEqual(response.status, "401 Unauthorized")
class TestLogin(cptc.BaseRestCherryPyTest):
auth_creds = (("username", "saltdev"), ("password", "saltdev"), ("eauth", "auto"))
def test_good_login(self):
"""
Test logging in
"""
body = urllib.parse.urlencode(self.auth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
return response
def test_leak(self):
"""
Test perms leak array is becoming bigger and bigger after each call
"""
lengthOfPerms = []
run_tests = 2
for x in range(0, run_tests):
body = urllib.parse.urlencode(self.auth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
response = salt.utils.json.loads(response.body[0])
lengthOfPerms.append(len(response["return"][0]["perms"]))
self.assertEqual(lengthOfPerms[0], lengthOfPerms[run_tests - 1])
return response
def test_bad_login(self):
"""
Test logging in
"""
body = urllib.parse.urlencode({"totally": "invalid_creds"})
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "401 Unauthorized")
def test_logout(self):
ret = self.test_good_login()
token = ret.headers["X-Auth-Token"]
body = urllib.parse.urlencode({})
request, response = self.request(
"/logout",
method="POST",
body=body,
headers={
"content-type": "application/x-www-form-urlencoded",
"X-Auth-Token": token,
},
)
self.assertEqual(response.status, "200 OK")
class TestRun(cptc.BaseRestCherryPyTest):
auth_creds = (
("username", "saltdev_auto"),
("password", "saltdev"),
("eauth", "auto"),
)
low = (
("client", "local"),
("tgt", "*"),
("fun", "test.ping"),
)
@pytest.mark.slow_test
def test_run_good_login(self):
"""
Test the run URL with good auth credentials
"""
cmd = dict(self.low, **dict(self.auth_creds))
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
def test_run_bad_login(self):
"""
Test the run URL with bad auth credentials
"""
cmd = dict(self.low, **{"totally": "invalid_creds"})
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "401 Unauthorized")
def test_run_empty_token(self):
"""
Test the run URL with empty token
"""
cmd = dict(self.low, **{"token": ""})
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
assert response.status == "401 Unauthorized"
def test_run_empty_token_upercase(self):
"""
Test the run URL with empty token with upercase characters
"""
cmd = dict(self.low, **{"ToKen": ""})
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
assert response.status == "401 Unauthorized"
def test_run_wrong_token(self):
"""
Test the run URL with incorrect token
"""
cmd = dict(self.low, **{"token": "bad"})
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
assert response.status == "401 Unauthorized"
def test_run_pathname_token(self):
"""
Test the run URL with path that exists in token
"""
cmd = dict(self.low, **{"token": os.path.join("etc", "passwd")})
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
assert response.status == "401 Unauthorized"
def test_run_pathname_not_exists_token(self):
"""
Test the run URL with path that does not exist in token
"""
cmd = dict(self.low, **{"token": os.path.join("tmp", "doesnotexist")})
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
assert response.status == "401 Unauthorized"
@pytest.mark.slow_test
def test_run_extra_parameters(self):
"""
Test the run URL with good auth credentials
"""
cmd = dict(self.low, **dict(self.auth_creds))
cmd["id_"] = "someminionname"
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
class TestWebhookDisableAuth(cptc.BaseRestCherryPyTest):
def __get_opts__(self):
return {
"rest_cherrypy": {
"port": 8000,
"debug": True,
"webhook_disable_auth": True,
},
}
def test_webhook_noauth(self):
"""
Auth can be disabled for requests to the webhook URL
"""
body = urllib.parse.urlencode({"foo": "Foo!"})
request, response = self.request(
"/hook",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
class TestArgKwarg(cptc.BaseRestCherryPyTest):
auth_creds = (("username", "saltdev"), ("password", "saltdev"), ("eauth", "auto"))
low = (
("client", "runner"),
("fun", "test.arg"),
# use singular form for arg and kwarg
("arg", [1234]),
("kwarg", {"ext_source": "redis"}),
)
def _token(self):
"""
Return the token
"""
body = urllib.parse.urlencode(self.auth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
return response.headers["X-Auth-Token"]
@pytest.mark.slow_test
def test_accepts_arg_kwarg_keys(self):
"""
Ensure that (singular) arg and kwarg keys (for passing parameters)
are supported by runners.
"""
cmd = dict(self.low)
body = salt.utils.json.dumps(cmd)
request, response = self.request(
"/",
method="POST",
body=body,
headers={
"content-type": "application/json",
"X-Auth-Token": self._token(),
"Accept": "application/json",
},
)
resp = salt.utils.json.loads(salt.utils.stringutils.to_str(response.body[0]))
self.assertEqual(resp["return"][0]["args"], [1234])
self.assertEqual(resp["return"][0]["kwargs"], {"ext_source": "redis"})
class TestJobs(cptc.BaseRestCherryPyTest):
auth_creds = (
("username", "saltdev_auto"),
("password", "saltdev"),
("eauth", "auto"),
)
low = (
("client", "local"),
("tgt", "*"),
("fun", "test.ping"),
)
def _token(self):
"""
Return the token
"""
body = urllib.parse.urlencode(self.auth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
return response.headers["X-Auth-Token"]
def _add_job(self):
"""
Helper function to add a job to the job cache
"""
cmd = dict(self.low, **dict(self.auth_creds))
body = urllib.parse.urlencode(cmd)
request, response = self.request(
"/run",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
@pytest.mark.flaky(max_runs=4)
@pytest.mark.slow_test
def test_all_jobs(self):
"""
test query to /jobs returns job data
"""
self._add_job()
request, response = self.request(
"/jobs",
method="GET",
headers={"Accept": "application/json", "X-Auth-Token": self._token()},
)
resp = salt.utils.json.loads(salt.utils.stringutils.to_str(response.body[0]))
self.assertIn("test.ping", str(resp["return"]))
self.assertEqual(response.status, "200 OK")

135
tests/integration/netapi/rest_cherrypy/test_app_pam.py
View file

@ -1,135 +0,0 @@
"""
Integration Tests for restcherry salt-api with pam eauth
"""
import urllib.parse
import pytest
import salt.utils.platform
import tests.support.cherrypy_testclasses as cptc
from tests.support.case import ModuleCase
from tests.support.unit import skipIf
if cptc.HAS_CHERRYPY:
import cherrypy
USERA = "saltdev-netapi"
USERA_PWD = "saltdev"
HASHED_USERA_PWD = "$6$SALTsalt$ZZFD90fKFWq8AGmmX0L3uBtS9fXL62SrTk5zcnQ6EkD6zoiM3kB88G1Zvs0xm/gZ7WXJRs5nsTBybUvGSqZkT."
AUTH_CREDS = {"username": USERA, "password": USERA_PWD, "eauth": "pam"}
@skipIf(cptc.HAS_CHERRYPY is False, "CherryPy not installed")
class TestAuthPAM(cptc.BaseRestCherryPyTest, ModuleCase):
"""
Test auth with pam using salt-api
"""
@pytest.mark.destructive_test
@pytest.mark.skip_if_not_root
def setUp(self):
super().setUp()
try:
add_user = self.run_function("user.add", [USERA], createhome=False)
add_pwd = self.run_function(
"shadow.set_password",
[
USERA,
USERA_PWD if salt.utils.platform.is_darwin() else HASHED_USERA_PWD,
],
)
self.assertTrue(add_user)
self.assertTrue(add_pwd)
user_list = self.run_function("user.list_users")
self.assertIn(USERA, str(user_list))
except AssertionError:
self.run_function("user.delete", [USERA], remove=True)
self.skipTest("Could not add user or password, skipping test")
@pytest.mark.slow_test
def test_bad_pwd_pam_chsh_service(self):
"""
Test login while specifying chsh service with bad passwd
This test ensures this PR is working correctly:
https://github.com/saltstack/salt/pull/31826
"""
copyauth_creds = AUTH_CREDS.copy()
copyauth_creds["service"] = "chsh"
copyauth_creds["password"] = "wrong_password"
body = urllib.parse.urlencode(copyauth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "401 Unauthorized")
@pytest.mark.slow_test
def test_bad_pwd_pam_login_service(self):
"""
Test login while specifying login service with bad passwd
This test ensures this PR is working correctly:
https://github.com/saltstack/salt/pull/31826
"""
copyauth_creds = AUTH_CREDS.copy()
copyauth_creds["service"] = "login"
copyauth_creds["password"] = "wrong_password"
body = urllib.parse.urlencode(copyauth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "401 Unauthorized")
@pytest.mark.slow_test
def test_good_pwd_pam_chsh_service(self):
"""
Test login while specifying chsh service with good passwd
This test ensures this PR is working correctly:
https://github.com/saltstack/salt/pull/31826
"""
copyauth_creds = AUTH_CREDS.copy()
copyauth_creds["service"] = "chsh"
body = urllib.parse.urlencode(copyauth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
@pytest.mark.slow_test
def test_good_pwd_pam_login_service(self):
"""
Test login while specifying login service with good passwd
This test ensures this PR is working correctly:
https://github.com/saltstack/salt/pull/31826
"""
copyauth_creds = AUTH_CREDS.copy()
copyauth_creds["service"] = "login"
body = urllib.parse.urlencode(copyauth_creds)
request, response = self.request(
"/login",
method="POST",
body=body,
headers={"content-type": "application/x-www-form-urlencoded"},
)
self.assertEqual(response.status, "200 OK")
@pytest.mark.destructive_test
@pytest.mark.skip_if_not_root
def tearDown(self):
"""
Clean up after tests. Delete user
"""
super().tearDown()
user_list = self.run_function("user.list_users")
# Remove saltdev user
if USERA in user_list:
self.run_function("user.delete", [USERA], remove=True)
# need to exit cherypy engine
cherrypy.engine.exit()

6
tests/pytests/functional/netapi/conftest.py
View file

@ -3,6 +3,12 @@ import pathlib
import pytest
import salt.config
import tests.support.netapi as netapi
from saltfactories.utils.ports import get_unused_localhost_port
@pytest.fixture
def netapi_port():
return get_unused_localhost_port()
@pytest.fixture

0
tests/integration/netapi/rest_cherrypy/__init__.py → tests/pytests/functional/netapi/rest_cherrypy/__init__.py
View file

39
tests/pytests/functional/netapi/rest_cherrypy/conftest.py Normal file
View file

@ -0,0 +1,39 @@
import pytest
import salt.ext.tornado.wsgi
import salt.netapi.rest_cherrypy.app
import tests.support.netapi as netapi
from tests.support.mock import patch
cherrypy = pytest.importorskip("cherrypy")
@pytest.fixture
def client_config(client_config, netapi_port):
client_config["rest_cherrypy"] = {"port": netapi_port, "debug": True}
return client_config
@pytest.fixture
def app(client_config, load_auth):
app, _, cherry_opts = salt.netapi.rest_cherrypy.app.get_app(client_config)
# These patches are here to allow running tests without a master running
with patch("salt.netapi.NetapiClient._is_master_running", return_value=True), patch(
"salt.auth.Resolver.mk_token", load_auth.mk_token
):
yield salt.ext.tornado.wsgi.WSGIContainer(
cherrypy.Application(app, "/", config=cherry_opts)
)
@pytest.fixture
def http_server(io_loop, app, netapi_port):
with netapi.TestsTornadoHttpServer(
io_loop=io_loop, app=app, port=netapi_port
) as server:
yield server
@pytest.fixture
def http_client(http_server):
return http_server.client

97
tests/pytests/functional/netapi/rest_cherrypy/test_auth.py Normal file
View file

@ -0,0 +1,97 @@
import urllib.parse
import pytest
import salt.utils.json
from salt.ext.tornado.httpclient import HTTPError
async def test_get_root_noauth(http_client):
"""
GET requests to the root URL should not require auth
"""
response = await http_client.fetch("/")
assert response.code == 200
async def test_post_root_auth(http_client):
"""
POST requests to the root URL redirect to login
"""
with pytest.raises(HTTPError) as exc:
await http_client.fetch("/", method="POST", body=salt.utils.json.dumps({}))
assert exc.value.code == 401
async def test_login_noauth(http_client):
"""
GET requests to the login URL should not require auth
"""
response = await http_client.fetch("/login")
assert response.code == 200
async def test_webhook_auth(http_client):
"""
Requests to the webhook URL require auth by default
"""
with pytest.raises(HTTPError) as exc:
await http_client.fetch("/hook", method="POST", body=salt.utils.json.dumps({}))
assert exc.value.code == 401
async def test_good_login(http_client, auth_creds, content_type_map, client_config):
"""
Test logging in
"""
response = await http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
assert response.code == 200
cookies = response.headers["Set-Cookie"]
response_obj = salt.utils.json.loads(response.body)["return"][0]
token = response_obj["token"]
assert "session_id={}".format(token) in cookies
perms = response_obj["perms"]
perms_config = client_config["external_auth"]["auto"][auth_creds["username"]]
assert set(perms) == set(perms_config)
assert "token" in response_obj # TODO: verify that its valid?
assert response_obj["user"] == auth_creds["username"]
assert response_obj["eauth"] == auth_creds["eauth"]
async def test_bad_login(http_client, content_type_map):
"""
Test logging in
"""
with pytest.raises(HTTPError) as exc:
body = urllib.parse.urlencode({"totally": "invalid_creds"})
await http_client.fetch(
"/login",
method="POST",
body=body,
headers={"Content-Type": content_type_map["form"]},
)
assert exc.value.code == 401
async def test_logout(http_client, auth_creds, content_type_map):
response = await http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
assert response.code == 200
token = response.headers["X-Auth-Token"]
body = urllib.parse.urlencode({})
response = await http_client.fetch(
"/logout",
method="POST",
body=body,
headers={"content-type": content_type_map["form"], "X-Auth-Token": token},
)
assert response.code == 200

66
tests/pytests/functional/netapi/rest_cherrypy/test_auth_pam.py Normal file
View file

@ -0,0 +1,66 @@
import urllib.parse
import pytest
from salt.ext.tornado.httpclient import HTTPError
pytestmark = [
pytest.mark.destructive_test,
pytest.mark.skip_if_not_root,
]
@pytest.fixture(scope="module")
def netapi_account():
with pytest.helpers.create_account(
username="saltdev-netapi", password="saltdev"
) as account:
yield account
@pytest.fixture
def auth_creds(netapi_account):
return {
"username": netapi_account.username,
"password": netapi_account.password,
"eauth": "pam",
}
@pytest.mark.parametrize("service", ["chsh", "login"])
async def test_bad_pwd_pam_chsh_service(
http_client, auth_creds, content_type_map, service
):
"""
Test login while specifying `chsh` or `login` service with bad passwd
This test ensures this PR is working correctly:
https://github.com/saltstack/salt/pull/31826
"""
auth_creds["service"] = service
auth_creds["password"] = "wrong_password"
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
assert exc.value.code == 401
@pytest.mark.parametrize("service", ["chsh", "login"])
async def test_good_pwd_pam_chsh_service(
http_client, auth_creds, content_type_map, service
):
"""
Test login while specifying `chsh` and `login` service with good passwd
This test ensures this PR is working correctly:
https://github.com/saltstack/salt/pull/31826
"""
auth_creds["service"] = service
response = await http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
assert response.code == 200

19
tests/pytests/functional/netapi/rest_cherrypy/test_cors.py Normal file
View file

@ -0,0 +1,19 @@
import pytest
from tests.support.mock import MagicMock, patch
@pytest.fixture
def app(app):
app.wsgi_application.config["global"]["tools.cors_tool.on"] = True
return app
async def test_option_request(http_client):
with patch(
"salt.netapi.rest_cherrypy.app.cherrypy.session", MagicMock(), create=True
):
response = await http_client.fetch(
"/", method="OPTIONS", headers={"Origin": "https://domain.com"}
)
assert response.code == 200
assert response.headers["Access-Control-Allow-Origin"] == "https://domain.com"

106
tests/pytests/functional/netapi/rest_cherrypy/test_in_formats.py Normal file
View file

@ -0,0 +1,106 @@
import urllib.parse
import pytest
import salt.utils.json
import salt.utils.yaml
from tests.support.mock import patch
@pytest.fixture
def app(app):
app.wsgi_application.config["global"]["tools.hypermedia_in.on"] = True
return app
@pytest.fixture
def token(http_client, auth_creds, content_type_map, io_loop):
response = io_loop.run_sync(
lambda: http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
)
assert response.code == 200
return response.headers["X-Auth-Token"]
@pytest.fixture
def client_headers(token, content_type_map):
return {
"Accept": content_type_map["json"],
"X-Auth-Token": token,
"Content-Type": content_type_map["form"],
}
async def test_urlencoded_ctype(http_client, client_headers, content_type_map):
low = {"client": "local", "fun": "test.ping", "tgt": "jerry"}
body = urllib.parse.urlencode(low)
client_headers["Content-Type"] = content_type_map["form"]
with patch(
"salt.client.LocalClient.run_job",
return_value={"jid": "20131219215650131543", "minions": ["jerry"]},
):
# We don't really want to run the job, hence the patch
response = await http_client.fetch(
"/", method="POST", body=body, headers=client_headers
)
assert response.code == 200
assert response.body == '{"return": [{"jerry": false}]}'
async def test_json_ctype(http_client, client_headers, content_type_map):
low = {"client": "local", "fun": "test.ping", "tgt": "jerry"}
body = salt.utils.json.dumps(low)
client_headers["Content-Type"] = content_type_map["json"]
with patch(
"salt.client.LocalClient.run_job",
return_value={"jid": "20131219215650131543", "minions": ["jerry"]},
):
# We don't really want to run the job, hence the patch
response = await http_client.fetch(
"/", method="POST", body=body, headers=client_headers
)
assert response.code == 200
assert response.body == '{"return": [{"jerry": false}]}'
async def test_json_as_text_out(http_client, client_headers):
"""
Some service send JSON as text/plain for compatibility purposes
"""
low = {"client": "local", "fun": "test.ping", "tgt": "jerry"}
body = salt.utils.json.dumps(low)
client_headers["Content-Type"] = "text/plain"
with patch(
"salt.client.LocalClient.run_job",
return_value={"jid": "20131219215650131543", "minions": ["jerry"]},
):
# We don't really want to run the job, hence the patch
response = await http_client.fetch(
"/", method="POST", body=body, headers=client_headers
)
assert response.code == 200
assert response.body == '{"return": [{"jerry": false}]}'
async def test_yaml_ctype(http_client, client_headers, content_type_map):
low = {"client": "local", "fun": "test.ping", "tgt": "jerry"}
body = salt.utils.yaml.safe_dump(low)
client_headers["Content-Type"] = content_type_map["yaml"]
with patch(
"salt.client.LocalClient.run_job",
return_value={"jid": "20131219215650131543", "minions": ["jerry"]},
):
# We don't really want to run the job, hence the patch
response = await http_client.fetch(
"/", method="POST", body=body, headers=client_headers
)
assert response.code == 200
assert response.body == '{"return": [{"jerry": false}]}'

35
tests/pytests/functional/netapi/rest_cherrypy/test_out_formats.py Normal file
View file

@ -0,0 +1,35 @@
import pytest
from salt.ext.tornado.httpclient import HTTPError
@pytest.fixture
def app(app):
app.wsgi_application.config["global"]["tools.hypermedia_out.on"] = True
return app
async def test_default_accept(http_client, content_type_map):
response = await http_client.fetch("/", method="GET")
assert response.headers["Content-Type"] == content_type_map["json"]
async def test_unsupported_accept(http_client):
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/", method="GET", headers={"Accept": "application/ms-word"}
)
assert exc.value.code == 406
async def test_json_out(http_client, content_type_map):
response = await http_client.fetch(
"/", method="GET", headers={"Accept": content_type_map["json"]}
)
assert response.headers["Content-Type"] == content_type_map["json"]
async def test_yaml_out(http_client, content_type_map):
response = await http_client.fetch(
"/", method="GET", headers={"Accept": content_type_map["yaml"]}
)
assert response.headers["Content-Type"] == content_type_map["yaml"]

12
tests/pytests/functional/netapi/rest_tornado/conftest.py
View file

@ -2,14 +2,22 @@ import pytest
import tests.support.netapi as netapi
@pytest.fixture
def client_config(client_config, netapi_port):
client_config["rest_tornado"] = {"port": netapi_port}
return client_config
@pytest.fixture
def app(app_urls, load_auth, client_config, minion_config):
return netapi.build_tornado_app(app_urls, load_auth, client_config, minion_config)
@pytest.fixture
def http_server(io_loop, app):
with netapi.TestsTornadoHttpServer(io_loop=io_loop, app=app) as server:
def http_server(io_loop, app, netapi_port):
with netapi.TestsTornadoHttpServer(
io_loop=io_loop, app=app, port=netapi_port
) as server:
yield server

6
tests/pytests/integration/netapi/conftest.py
View file

@ -1,6 +1,12 @@
import pytest
import salt.config
import tests.support.netapi as netapi
from saltfactories.utils.ports import get_unused_localhost_port
@pytest.fixture
def netapi_port():
return get_unused_localhost_port()
@pytest.fixture

0
tests/unit/netapi/__init__.py → tests/pytests/integration/netapi/rest_cherrypy/__init__.py
View file

41
tests/pytests/integration/netapi/rest_cherrypy/conftest.py Normal file
View file

@ -0,0 +1,41 @@
import pytest
import salt.ext.tornado.wsgi
import salt.netapi.rest_cherrypy.app
import tests.support.netapi as netapi
cherrypy = pytest.importorskip("cherrypy")
@pytest.fixture
def client_config(client_config, netapi_port):
client_config["rest_cherrypy"] = {"port": netapi_port, "debug": True}
return client_config
@pytest.fixture
def app(client_config, load_auth, salt_minion):
app, _, cherry_opts = salt.netapi.rest_cherrypy.app.get_app(client_config)
return salt.ext.tornado.wsgi.WSGIContainer(
cherrypy.Application(app, "/", config=cherry_opts)
)
@pytest.fixture
def client_headers(auth_token, content_type_map):
return {
"Content-Type": content_type_map["form"],
}
@pytest.fixture
def http_server(io_loop, app, netapi_port, client_headers):
with netapi.TestsTornadoHttpServer(
io_loop=io_loop, app=app, port=netapi_port, client_headers=client_headers
) as server:
yield server
@pytest.fixture
def http_client(http_server):
return http_server.client

58
tests/pytests/integration/netapi/rest_cherrypy/test_arg_kwarg.py Normal file
View file

@ -0,0 +1,58 @@
import urllib.parse
import pytest
import salt.utils.json
@pytest.mark.slow_test
async def test_accepts_arg_kwarg_keys(
http_client, auth_creds, content_type_map, subtests
):
"""
Ensure that (singular) arg and kwarg keys (for passing parameters)
are supported by runners.
"""
# Login to get the token
body = salt.utils.json.dumps(auth_creds)
response = await http_client.fetch(
"/login",
method="POST",
body=body,
headers={
"Accept": content_type_map["json"],
"Content-Type": content_type_map["json"],
},
)
assert response.code == 200
token = response.headers["X-Auth-Token"]
low = {
"client": "runner",
"fun": "test.arg",
"arg": [1234, 5678],
"kwarg": {"ext_source": "redis"},
}
for content_type in ("json", "form"):
with subtests.test(content_type=content_type):
if content_type == "json":
body = salt.utils.json.dumps(low)
else:
_low = low.copy()
arg = _low.pop("arg")
body = urllib.parse.urlencode(_low)
for _arg in arg:
body += "&arg={}".format(_arg)
response = await http_client.fetch(
"/",
method="POST",
body=body,
headers={
"Accept": content_type_map["json"],
"Content-Type": content_type_map[content_type],
"X-Auth-Token": token,
},
)
assert response.code == 200
body = salt.utils.json.loads(response.body)
ret = body["return"][0]
assert ret["args"] == low["arg"]
assert ret["kwargs"] == low["kwarg"]

97
tests/pytests/integration/netapi/rest_cherrypy/test_auth.py Normal file
View file

@ -0,0 +1,97 @@
import urllib.parse
import pytest
import salt.utils.json
from salt.ext.tornado.httpclient import HTTPError
async def test_get_root_noauth(http_client):
"""
GET requests to the root URL should not require auth
"""
response = await http_client.fetch("/")
assert response.code == 200
async def test_post_root_auth(http_client):
"""
POST requests to the root URL redirect to login
"""
with pytest.raises(HTTPError) as exc:
await http_client.fetch("/", method="POST", body=salt.utils.json.dumps({}))
assert exc.value.code == 401
async def test_login_noauth(http_client):
"""
GET requests to the login URL should not require auth
"""
response = await http_client.fetch("/login")
assert response.code == 200
async def test_webhook_auth(http_client):
"""
Requests to the webhook URL require auth by default
"""
with pytest.raises(HTTPError) as exc:
await http_client.fetch("/hook", method="POST", body=salt.utils.json.dumps({}))
assert exc.value.code == 401
async def test_good_login(http_client, auth_creds, content_type_map, client_config):
"""
Test logging in
"""
response = await http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
assert response.code == 200
cookies = response.headers["Set-Cookie"]
response_obj = salt.utils.json.loads(response.body)["return"][0]
token = response_obj["token"]
assert "session_id={}".format(token) in cookies
perms = response_obj["perms"]
perms_config = client_config["external_auth"]["auto"][auth_creds["username"]]
assert set(perms) == set(perms_config)
assert "token" in response_obj # TODO: verify that its valid?
assert response_obj["user"] == auth_creds["username"]
assert response_obj["eauth"] == auth_creds["eauth"]
async def test_bad_login(http_client, content_type_map):
"""
Test logging in
"""
with pytest.raises(HTTPError) as exc:
body = urllib.parse.urlencode({"totally": "invalid_creds"})
await http_client.fetch(
"/login",
method="POST",
body=body,
headers={"Content-Type": content_type_map["form"]},
)
assert exc.value.code == 401
async def test_logout(http_client, auth_creds, content_type_map):
response = await http_client.fetch(
"/login",
method="POST",
body=urllib.parse.urlencode(auth_creds),
headers={"Content-Type": content_type_map["form"]},
)
assert response.code == 200
token = response.headers["X-Auth-Token"]
body = urllib.parse.urlencode({})
response = await http_client.fetch(
"/logout",
method="POST",
body=body,
headers={"content-type": content_type_map["form"], "X-Auth-Token": token},
)
assert response.code == 200

52
tests/pytests/integration/netapi/rest_cherrypy/test_jobs.py Normal file
View file

@ -0,0 +1,52 @@
import pytest
import salt.utils.json
@pytest.mark.slow_test
async def test_all_jobs(http_client, auth_creds, content_type_map):
"""
test query to /jobs returns job data
"""
# Login to get the token
body = salt.utils.json.dumps(auth_creds)
response = await http_client.fetch(
"/login",
method="POST",
body=body,
headers={
"Accept": content_type_map["json"],
"Content-Type": content_type_map["json"],
},
)
assert response.code == 200
token = response.headers["X-Auth-Token"]
low = {"client": "local", "tgt": "*", "fun": "test.ping", **auth_creds}
body = salt.utils.json.dumps(low)
# Add a job
response = await http_client.fetch(
"/run",
method="POST",
body=body,
headers={
"Accept": content_type_map["json"],
"Content-Type": content_type_map["json"],
},
)
assert response.code == 200
body = salt.utils.json.loads(response.body)
# Get Jobs
response = await http_client.fetch(
"/jobs",
method="GET",
headers={"Accept": content_type_map["json"], "X-Auth-Token": token},
)
assert response.code == 200
body = salt.utils.json.loads(response.body)
for ret in body["return"][0].values():
assert "Function" in ret
if ret["Function"] == "test.ping":
break
else:
pytest.fail("Failed to get the 'test.ping' job")

133
tests/pytests/integration/netapi/rest_cherrypy/test_run.py Normal file
View file

@ -0,0 +1,133 @@
import urllib.parse
import pytest
from salt.ext.tornado.httpclient import HTTPError
async def test_run_good_login(http_client, auth_creds):
"""
Test the run URL with good auth credentials
"""
low = {"client": "local", "tgt": "*", "fun": "test.ping", **auth_creds}
body = urllib.parse.urlencode(low)
response = await http_client.fetch("/run", method="POST", body=body)
assert response.code == 200
async def test_run_bad_login(http_client):
"""
Test the run URL with bad auth credentials
"""
low = {
"client": "local",
"tgt": "*",
"fun": "test.ping",
**{"totally": "invalid_creds"},
}
body = urllib.parse.urlencode(low)
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/run",
method="POST",
body=body,
)
assert exc.value.code == 401
async def test_run_empty_token(http_client):
"""
Test the run URL with empty token
"""
low = {"client": "local", "tgt": "*", "fun": "test.ping", **{"token": ""}}
body = urllib.parse.urlencode(low)
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/run",
method="POST",
body=body,
)
assert exc.value.code == 401
async def test_run_empty_token_upercase(http_client):
"""
Test the run URL with empty token with upercase characters
"""
low = {"client": "local", "tgt": "*", "fun": "test.ping", **{"ToKen": ""}}
body = urllib.parse.urlencode(low)
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/run",
method="POST",
body=body,
)
assert exc.value.code == 401
async def test_run_wrong_token(http_client):
"""
Test the run URL with incorrect token
"""
low = {"client": "local", "tgt": "*", "fun": "test.ping", **{"token": "bad"}}
body = urllib.parse.urlencode(low)
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/run",
method="POST",
body=body,
)
assert exc.value.code == 401
async def test_run_pathname_token(http_client):
"""
Test the run URL with path that exists in token
"""
low = {
"client": "local",
"tgt": "*",
"fun": "test.ping",
**{"token": "/etc/passwd"},
}
body = urllib.parse.urlencode(low)
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/run",
method="POST",
body=body,
)
assert exc.value.code == 401
async def test_run_pathname_not_exists_token(http_client):
"""
Test the run URL with path that does not exist in token
"""
low = {
"client": "local",
"tgt": "*",
"fun": "test.ping",
**{"token": "/tmp/does-not-exist"},
}
body = urllib.parse.urlencode(low)
with pytest.raises(HTTPError) as exc:
await http_client.fetch(
"/run",
method="POST",
body=body,
)
assert exc.value.code == 401
@pytest.mark.slow_test
async def test_run_extra_parameters(http_client, auth_creds):
"""
Test the run URL with good auth credentials
"""
low = {"client": "local", "tgt": "*", "fun": "test.ping", **auth_creds}
low["id_"] = "some-minion-name"
body = urllib.parse.urlencode(low)
response = await http_client.fetch("/run", method="POST", body=body)
assert response.code == 200

20
tests/pytests/integration/netapi/rest_cherrypy/test_webhook_disable_auth.py Normal file
View file

@ -0,0 +1,20 @@
import urllib.parse
import pytest
@pytest.fixture
def client_config(client_config):
client_config["rest_cherrypy"]["webhook_disable_auth"] = True
return client_config
async def test_webhook_noauth(http_client):
"""
Auth can be disabled for requests to the webhook URL
See the above ``client_config`` fixture where we disable it
"""
body = urllib.parse.urlencode({"foo": "Foo!"})
response = await http_client.fetch("/hook", method="POST", body=body)
assert response.code == 200

10
tests/pytests/integration/netapi/rest_tornado/conftest.py
View file

@ -3,6 +3,12 @@ import tests.support.netapi as netapi
from salt.netapi.rest_tornado import saltnado
@pytest.fixture
def client_config(client_config, netapi_port):
client_config["rest_tornado"] = {"port": netapi_port}
return client_config
@pytest.fixture
def app(app_urls, load_auth, client_config, minion_config, salt_sub_minion):
return netapi.build_tornado_app(
@ -19,9 +25,9 @@ def client_headers(auth_token, content_type_map):
@pytest.fixture
def http_server(io_loop, app, client_headers):
def http_server(io_loop, app, client_headers, netapi_port):
with netapi.TestsTornadoHttpServer(
io_loop=io_loop, app=app, client_headers=client_headers
io_loop=io_loop, app=app, port=netapi_port, client_headers=client_headers
) as server:
yield server

135
tests/support/cherrypy_testclasses.py
View file

@ -1,135 +0,0 @@
import os
import salt.config
from tests.support.mock import patch
from tests.support.runtests import RUNTIME_VARS
try:
import cherrypy
HAS_CHERRYPY = True
except ImportError:
HAS_CHERRYPY = False
if HAS_CHERRYPY:
from tests.support.cptestcase import BaseCherryPyTestCase
from salt.netapi.rest_cherrypy import app
else:
from tests.support.unit import TestCase, skipIf
@skipIf(HAS_CHERRYPY is False, "The CherryPy python package needs to be installed")
class BaseCherryPyTestCase(TestCase):
pass
class BaseToolsTest(BaseCherryPyTestCase):
pass
class BaseRestCherryPyTest(BaseCherryPyTestCase):
"""
A base TestCase subclass for the rest_cherrypy module
This mocks all interactions with Salt-core and sets up a dummy
(unsubscribed) CherryPy web server.
"""
def __get_opts__(self):
return None
@classmethod
def setUpClass(cls):
master_conf = os.path.join(RUNTIME_VARS.TMP_CONF_DIR, "master")
cls.config = salt.config.client_config(master_conf)
cls.base_opts = {}
cls.base_opts.update(cls.config)
@classmethod
def tearDownClass(cls):
del cls.config
del cls.base_opts
def setUp(self):
# Make a local reference to the CherryPy app so we can mock attributes.
self.app = app
self.addCleanup(delattr, self, "app")
master_conf = os.path.join(RUNTIME_VARS.TMP_CONF_DIR, "master")
client_config = salt.config.client_config(master_conf)
base_opts = {}
base_opts.update(client_config)
base_opts.update(
self.__get_opts__()
or {
"external_auth": {
"auto": {"saltdev": ["@wheel", "@runner", ".*"], "*": "cmd.*"},
"pam": {"saltdev": ["@wheel", "@runner", ".*"]},
},
"rest_cherrypy": {"port": 8000, "debug": True},
}
)
root, apiopts, conf = app.get_app(base_opts)
cherrypy.tree.mount(root, "/", conf)
cherrypy.server.unsubscribe()
cherrypy.engine.start()
# Make sure cherrypy does not memleak on its bus since it keeps
# adding handlers without cleaning the old ones each time we setup
# a new application
for value in cherrypy.engine.listeners.values():
value.clear()
cherrypy.engine._priorities.clear()
self.addCleanup(cherrypy.engine.exit)
class Root:
"""
The simplest CherryPy app needed to test individual tools
"""
exposed = True
_cp_config = {}
def GET(self):
return {"return": ["Hello world."]}
def POST(self, *args, **kwargs):
return {"return": [{"args": args}, {"kwargs": kwargs}]}
if HAS_CHERRYPY:
class BaseToolsTest(BaseCherryPyTestCase): # pylint: disable=E0102
"""
A base class so tests can selectively turn individual tools on for testing
"""
def __get_conf__(self):
return {
"/": {"request.dispatch": cherrypy.dispatch.MethodDispatcher()},
}
def __get_cp_config__(self):
return {}
def setUp(self):
root = Root()
patcher = patch.object(root, "_cp_config", self.__get_cp_config__())
patcher.start()
self.addCleanup(patcher.stop)
# Make sure cherrypy does not memleak on its bus since it keeps
# adding handlers without cleaning the old ones each time we setup
# a new application
for value in cherrypy.engine.listeners.values():
value.clear()
cherrypy.engine._priorities.clear()
app = cherrypy.tree.mount(root, "/", self.__get_conf__())
cherrypy.server.unsubscribe()
cherrypy.engine.start()
self.addCleanup(cherrypy.engine.exit)

135
tests/support/cptestcase.py
View file

@ -1,135 +0,0 @@
# Copyright (c) 2011-2012, Sylvain Hellegouarch
# All rights reserved.
# Redistribution and use in source and binary forms, with or without modification,
# are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
# * Neither the name of Sylvain Hellegouarch nor the names of his contributors
# may be used to endorse or promote products derived from this software
# without specific prior written permission.
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Modified from the original. See the Git history of this file for details.
# https://bitbucket.org/Lawouach/cherrypy-recipes/src/50aff88dc4e24206518ec32e1c32af043f2729da/testing/unit/serverless/cptestcase.py
# pylint: disable=import-error
import io
import cherrypy # pylint: disable=3rd-party-module-not-gated
import salt.utils.stringutils
from tests.support.case import TestCase
# pylint: enable=import-error
# Not strictly speaking mandatory but just makes sense
cherrypy.config.update({"environment": "test_suite"})
# This is mandatory so that the HTTP server isn't started
# if you need to actually start (why would you?), simply
# subscribe it back.
cherrypy.server.unsubscribe()
# simulate fake socket address... they are irrelevant in our context
local = cherrypy.lib.httputil.Host("127.0.0.1", 50000, "")
remote = cherrypy.lib.httputil.Host("127.0.0.1", 50001, "")
__all__ = ["BaseCherryPyTestCase"]
class BaseCherryPyTestCase(TestCase):
def request(
self,
path="/",
method="GET",
app_path="",
scheme="http",
proto="HTTP/1.1",
body=None,
qs=None,
headers=None,
**kwargs
):
"""
CherryPy does not have a facility for serverless unit testing.
However this recipe demonstrates a way of doing it by
calling its internal API to simulate an incoming request.
This will exercise the whole stack from there.
Remember a couple of things:
* CherryPy is multithreaded. The response you will get
from this method is a thread-data object attached to
the current thread. Unless you use many threads from
within a unit test, you can mostly forget
about the thread data aspect of the response.
* Responses are dispatched to a mounted application's
page handler, if found. This is the reason why you
must indicate which app you are targeting with
this request by specifying its mount point.
You can simulate various request settings by setting
the `headers` parameter to a dictionary of headers,
the request's `scheme` or `protocol`.
.. seealso: http://docs.cherrypy.org/stable/refman/_cprequest.html#cherrypy._cprequest.Response
"""
# This is a required header when running HTTP/1.1
h = {"Host": "127.0.0.1"}
# if we had some data passed as the request entity
# let's make sure we have the content-length set
fd = None
if body is not None:
h["content-length"] = "{}".format(len(body))
fd = io.BytesIO(salt.utils.stringutils.to_bytes(body))
if headers is not None:
h.update(headers)
# Get our application and run the request against it
app = cherrypy.tree.apps.get(app_path)
if not app:
# XXX: perhaps not the best exception to raise?
raise AssertionError("No application mounted at '{}'".format(app_path))
# Cleanup any previous returned response
# between calls to this method
app.release_serving()
# Let's fake the local and remote addresses
request, response = app.get_serving(local, remote, scheme, proto)
try:
h = [(k, v) for k, v in h.items()]
response = request.run(method, path, qs, proto, h, fd)
finally:
if fd:
fd.close()
fd = None
if response.output_status.startswith(b"500"):
response_body = response.collapse_body()
response_body = response_body.decode(__salt_system_encoding__)
print(response_body)
raise AssertionError("Unexpected error")
# collapse the response into a bytestring
response.collapse_body()
return request, response

4
tests/support/netapi.py
View file

@ -53,10 +53,10 @@ class TestsHttpClient:
class TestsTornadoHttpServer:
io_loop = attr.ib(repr=False)
app = attr.ib()
port = attr.ib(repr=False)
protocol = attr.ib(default="http", repr=False)
http_server_options = attr.ib(default=attr.Factory(dict))
sock = attr.ib(init=False, repr=False)
port = attr.ib(init=False, repr=False)
address = attr.ib(init=False)
server = attr.ib(init=False)
client_headers = attr.ib(default=None)
@ -65,7 +65,7 @@ class TestsTornadoHttpServer:
@sock.default
def _sock_default(self):
return netutil.bind_sockets(
None, "127.0.0.1", family=socket.AF_INET, reuse_port=False
self.port, "127.0.0.1", family=socket.AF_INET, reuse_port=False
)[0]
@port.default

117
tests/unit/netapi/test_rest_cherrypy.py
View file

@ -1,117 +0,0 @@
from urllib.parse import urlencode
import salt.utils.json
import salt.utils.yaml
from tests.support.cherrypy_testclasses import BaseToolsTest
class TestOutFormats(BaseToolsTest):
def __get_cp_config__(self):
return {
"tools.hypermedia_out.on": True,
}
def test_default_accept(self):
request, response = self.request("/")
self.assertEqual(response.headers["Content-type"], "application/json")
def test_unsupported_accept(self):
request, response = self.request(
"/", headers=(("Accept", "application/ms-word"),)
)
self.assertEqual(response.status, "406 Not Acceptable")
def test_json_out(self):
request, response = self.request("/", headers=(("Accept", "application/json"),))
self.assertEqual(response.headers["Content-type"], "application/json")
def test_yaml_out(self):
request, response = self.request(
"/", headers=(("Accept", "application/x-yaml"),)
)
self.assertEqual(response.headers["Content-type"], "application/x-yaml")
class TestInFormats(BaseToolsTest):
def __get_cp_config__(self):
return {
"tools.hypermedia_in.on": True,
}
def test_urlencoded_ctype(self):
data = {"valid": "stuff"}
raw = "valid=stuff"
request, response = self.request(
"/",
method="POST",
body=urlencode(data),
headers=(("Content-type", "application/x-www-form-urlencoded"),),
)
self.assertEqual(response.status, "200 OK")
self.assertEqual(request.raw_body, raw)
self.assertDictEqual(request.unserialized_data, data)
def test_urlencoded_multi_args(self):
multi_args = "arg=arg1&arg=arg2"
expected = {"arg": ["arg1", "arg2"]}
request, response = self.request(
"/",
method="POST",
body=multi_args,
headers=(("Content-type", "application/x-www-form-urlencoded"),),
)
self.assertEqual(response.status, "200 OK")
self.assertEqual(request.raw_body, multi_args)
self.assertDictEqual(request.unserialized_data, expected)
def test_json_ctype(self):
data = {"valid": "stuff"}
request, response = self.request(
"/",
method="POST",
body=salt.utils.json.dumps(data),
headers=(("Content-type", "application/json"),),
)
self.assertEqual(response.status, "200 OK")
self.assertDictEqual(request.unserialized_data, data)
def test_json_as_text_out(self):
"""
Some service send JSON as text/plain for compatibility purposes
"""
data = {"valid": "stuff"}
request, response = self.request(
"/",
method="POST",
body=salt.utils.json.dumps(data),
headers=(("Content-type", "text/plain"),),
)
self.assertEqual(response.status, "200 OK")
self.assertDictEqual(request.unserialized_data, data)
def test_yaml_ctype(self):
data = {"valid": "stuff"}
request, response = self.request(
"/",
method="POST",
body=salt.utils.yaml.safe_dump(data),
headers=(("Content-type", "application/x-yaml"),),
)
self.assertEqual(response.status, "200 OK")
self.assertDictEqual(request.unserialized_data, data)
class TestCors(BaseToolsTest):
def __get_cp_config__(self):
return {
"tools.cors_tool.on": True,
}
def test_option_request(self):
request, response = self.request(
"/", method="OPTIONS", headers=(("Origin", "https://domain.com"),)
)
self.assertEqual(response.status, "200 OK")
self.assertEqual(
response.headers.get("Access-Control-Allow-Origin"), "https://domain.com"
)

1
tests/unit/test_module_names.py
View file

@ -136,7 +136,6 @@ class BadTestModuleNamesTestCase(TestCase):
"integration.modules.test_service",
"integration.modules.test_state_jinja_filters",
"integration.modules.test_sysctl",
"integration.netapi.rest_cherrypy.test_app_pam",
"integration.netapi.rest_tornado.test_app",
"integration.netapi.test_client",
"integration.output.test_output",