76 lines
2.6 KiB
Python
Executable File
76 lines
2.6 KiB
Python
Executable File
#!/usr/bin/python3
|
|
|
|
from time import sleep
|
|
|
|
import composertest
|
|
import requests
|
|
import subprocess
|
|
|
|
|
|
class TestApi(composertest.ComposerTestCase):
|
|
"""Test Composer HTTP API"""
|
|
|
|
def setUp(self):
|
|
super(TestApi, self).setUp()
|
|
|
|
# Forward /run/weldr/api.socket to a port on the host
|
|
# Set ExitOnForwardFailure so that ssh blocks until the forward is set
|
|
# up before going to the background (-f), which it closes stdout. We
|
|
# wait for that by calling read() on it.
|
|
self.composer_port = self.network._lock(8080)
|
|
forwarder_command = self.ssh_command[:]
|
|
forwarder_command.extend(["-fNT",
|
|
"-o", "ExitOnForwardFailure=yes",
|
|
"-L", "localhost:%d:/run/weldr/api.socket" % self.composer_port])
|
|
self.forwarder_proc = subprocess.Popen(forwarder_command, stdout=subprocess.PIPE)
|
|
self.forwarder_proc.stdout.read()
|
|
|
|
def tearDown(self):
|
|
self.forwarder_proc.terminate()
|
|
sleep(1) # wait and check for timeout
|
|
if self.forwarder_proc.poll() is None:
|
|
self.forwarder_proc.kill()
|
|
super(TestApi, self).tearDown()
|
|
|
|
def request(self, method, path, check=True):
|
|
self.assertEqual(path[0], "/")
|
|
url = "http://localhost:%d%s" % (self.composer_port, path)
|
|
r = requests.request(method, url, timeout=30)
|
|
if check:
|
|
r.raise_for_status()
|
|
return r
|
|
|
|
def test_basic(self):
|
|
"""Basic checks for the API"""
|
|
|
|
#
|
|
# API status without depsolve errors
|
|
#
|
|
r = self.request("GET", "/api/status")
|
|
self.assertEqual(r.status_code, 200)
|
|
status = r.json()
|
|
self.assertEqual(status.keys(), { "build", "api", "db_version", "schema_version", "db_supported", "backend", "msgs" })
|
|
self.assertEqual(status["msgs"], [])
|
|
self.assertEqual(r.headers.keys(), { "Content-Type", "Content-Length", "Date" })
|
|
|
|
#
|
|
# HTTP errors should return json responses
|
|
#
|
|
r = self.request("GET", "/marmalade", check=False)
|
|
self.assertEqual(r.status_code, 404)
|
|
self.assertEqual(r.json(), {
|
|
"status": False,
|
|
"errors": [{ "id": "HTTPError", "code": 404, "msg": "Not Found" }]
|
|
})
|
|
|
|
r = self.request("POST", "/api/status", check=False)
|
|
self.assertEqual(r.status_code, 405)
|
|
self.assertEqual(r.json(), {
|
|
"status": False,
|
|
"errors": [{ "id": "HTTPError", "code": 405, "msg": "Method Not Allowed" }]
|
|
})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
composertest.main()
|