89 lines
3.1 KiB
Python
Executable File
89 lines
3.1 KiB
Python
Executable File
#!/usr/bin/python3
|
|
|
|
import composertest
|
|
import requests
|
|
import subprocess
|
|
|
|
class TestApi(composertest.ComposerTestCase):
|
|
"""Test Composer HTTP API"""
|
|
|
|
def setUp(self):
|
|
super().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, "-fNT",
|
|
"-o", "ExitOnForwardFailure=yes",
|
|
"-L", f"localhost:{self.composer_port}:/run/weldr/api.socket"]
|
|
self.forwarder_proc = subprocess.Popen(forwarder_command, stdout=subprocess.PIPE)
|
|
self.forwarder_proc.stdout.read()
|
|
|
|
def tearDown(self):
|
|
self.forwarder_proc.terminate()
|
|
try:
|
|
self.forwarder_proc.wait(timeout=1)
|
|
except TimeoutError:
|
|
self.forwarder_proc.kill()
|
|
super().tearDown()
|
|
|
|
def request(self, method, path, json=None, check=True):
|
|
self.assertEqual(path[0], "/")
|
|
r = requests.request(method, f"http://localhost:{self.composer_port}{path}", json=json, 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" }]
|
|
})
|
|
|
|
#
|
|
# API create blueprint with InvalidChars
|
|
#
|
|
invalid_blueprint = {
|
|
"name": "Name,With,Commas",
|
|
"description": "",
|
|
"version": "0.0.1",
|
|
"modules": [],
|
|
"groups": []
|
|
}
|
|
r = self.request("POST", "/api/v0/blueprints/new", json=invalid_blueprint, check=False)
|
|
self.assertEqual(r.status_code, 400)
|
|
self.assertEqual(r.json(), {
|
|
"status": False,
|
|
"errors": [{ "id": "InvalidChars", "msg": "Invalid characters in API path" }]
|
|
})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
composertest.main()
|