diff options
author | Lars Wirzenius <liw@liw.fi> | 2018-03-17 16:59:16 +0200 |
---|---|---|
committer | Lars Wirzenius <liw@liw.fi> | 2018-03-30 10:11:32 +0300 |
commit | 87001d6a0c5d9d8b48e716ec733e5ba6c0c4afd4 (patch) | |
tree | cebdf8da53237de0740fd0911f8bca80b8134ba4 /ick2/client_tests.py | |
parent | b211b8918ba5dcbdfad04a7b84b2c7ba27b4070f (diff) | |
download | ick2-87001d6a0c5d9d8b48e716ec733e5ba6c0c4afd4.tar.gz |
Add: HttpAPI class for using REST-ful HTTP API
Diffstat (limited to 'ick2/client_tests.py')
-rw-r--r-- | ick2/client_tests.py | 146 |
1 files changed, 146 insertions, 0 deletions
diff --git a/ick2/client_tests.py b/ick2/client_tests.py new file mode 100644 index 0000000..1a0bcc5 --- /dev/null +++ b/ick2/client_tests.py @@ -0,0 +1,146 @@ +# Copyright (C) 2018 Lars Wirzenius +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import json +import unittest + + +import ick2 + + +json_type = 'application/json' + + +class HttpAPITests(unittest.TestCase): + + def setUp(self): + self.session = FakeHttpSession() + self.session.token = 'SECRET-TOKEN' + self.client = ick2.HttpAPI() + self.client.set_session(self.session) + self.client.set_token(self.session.token) + + def test_get_dict_raises_exception_on_error(self): + self.session.response = FakeResponse(400) + with self.assertRaises(ick2.HttpError): + self.client.get_dict('http://controller/version') + + def test_get_dict_raises_exception_on_not_json(self): + self.session.response = FakeResponse(200, body=json.dumps('{}')) + with self.assertRaises(ick2.HttpError): + self.client.get_dict('http://controller/version') + + def test_get_dict_raises_exception_on_malformed_json(self): + self.session.response = FakeResponse( + 200, body='this is not really JSON', content_type=json_type) + with self.assertRaises(ick2.HttpError): + self.client.get_dict('http://controller/version') + + def test_get_dict_returns_response(self): + version = { + 'version': '1.0', + } + + self.session.response = FakeResponse( + 200, body=json.dumps(version), content_type=json_type) + obj = self.client.get_dict('http://controller/version') + self.assertEqual(obj, version) + + def test_get_blob_raises_exception_on_error(self): + self.session.response = FakeResponse(404) + with self.assertRaises(ick2.HttpError): + self.client.get_blob('http://blobs/blob/foo') + + def test_get_blob_returns_response(self): + blob = b'hello, world\n' + self.session.response = FakeResponse(200, body=blob) + obj = self.client.get_blob('http://blobs/blob/foo') + self.assertEqual(obj, blob) + + def test_post_raises_exception_on_error(self): + self.session.response = FakeResponse(400) + with self.assertRaises(ick2.HttpError): + self.client.post('http://controller/work', body='') + + def test_post_succeeds(self): + work = { + 'project': 'foo', + 'stdout': 'hello, world\n', + } + + self.session.response = FakeResponse( + 201, body=json.dumps(work), content_type=json_type) + obj = self.client.post('http://controller/work', body=work) + self.assertEqual(obj, None) + + def test_put_raises_exception_on_error(self): + blob = b'fooblob' + self.session.response = FakeResponse(400) + with self.assertRaises(ick2.HttpError): + self.client.put('http://blobs/blob/foo', body=blob) + + def test_put_succeeds(self): + blob = b'fooblob' + self.session.response = FakeResponse(201, body=None) + obj = self.client.put('http://controller/work', body=blob) + self.assertEqual(obj, None) + + +class FakeHttpSession: + + def __init__(self): + self.response = None + self.token = None + + def get(self, url, headers=None, verify=None): + assert self.response is not None + assert self.is_authorized(headers) + return self.response + + def post(self, url, headers=None, data=None, verify=None): + assert self.response is not None + assert self.is_authorized(headers) + return self.response + + def put(self, url, headers=None, data=None, verify=None): + assert self.response is not None + assert self.is_authorized(headers) + return self.response + + def is_authorized(self, headers): + assert self.token is not None + assert 'Authorization' in headers + v = headers['Authorization'] + return v == 'Bearer {}'.format(self.token) + + +class FakeResponse: + + def __init__(self, status_code, body=None, content_type=None): + if content_type is None: + content_type = 'application/octet-stream' + + self.status_code = status_code + self.headers = { + 'Content-Type': content_type, + } + self.content = body + + @property + def ok(self): + return self.status_code in (200, 201) + + def json(self): + return json.loads(self.content) |