Put together first pass of a self contained core service API client.

Author: albu-diku

mig/lib/coreapi/__init__.py

@@ -0,0 +1,89 @@
+import codecs
+import json
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError
+
+from mig.lib.coresvc.payloads import PAYLOAD_POST_USER
+
+
+def attempt_to_decode_response_data(data, response_encoding=None):
+    if data is None:
+        return None
+    elif response_encoding == 'textual':
+        data = codecs.decode(data, 'utf8')
+
+        try:
+            return json.loads(data)
+        except Exception as e:
+            return data
+    elif response_encoding == 'binary':
+        return data
+    else:
+        raise AssertionError(
+            'issue_POST: unknown response_encoding "%s"' % (response_encoding,))
+
+
+class CoreApiClient:
+    def __init__(self, base_url):
+        self._base_url = base_url
+
+    def _issue_GET(self, request_path, query_dict=None, response_encoding='textual'):
+        request_url = ''.join((self._base_url, request_path))
+
+        if query_dict is not None:
+            query_string = urlencode(query_dict)
+            request_url = ''.join((request_url, '?', query_string))
+
+        status = 0
+        data = None
+
+        try:
+            response = urlopen(request_url, None, timeout=2000)
+
+            status = response.getcode()
+            data = response.read()
+        except HTTPError as httpexc:
+            status = httpexc.code
+            data = None
+
+        content = attempt_to_decode_response_data(data, response_encoding)
+        return (status, content)
+
+    def _issue_POST(self, request_path, request_data=None, request_json=None, response_encoding='textual'):
+        request_url = ''.join((self._base_url, request_path))
+
+        if request_data and request_json:
+            raise ValueError(
+                "only one of data or json request data may be specified")
+
+        status = 0
+        data = None
+
+        try:
+            if request_json is not None:
+                request_data = codecs.encode(json.dumps(request_json), 'utf8')
+                request_headers = {
+                    'Content-Type': 'application/json'
+                }
+                request = Request(request_url, request_data,
+                                  headers=request_headers)
+            elif request_data is not None:
+                request = Request(request_url, request_data)
+            else:
+                request = Request(request_url)
+
+            response = urlopen(request, timeout=2000)
+
+            status = response.getcode()
+            data = response.read()
+        except HTTPError as httpexc:
+            status = httpexc.code
+            data = httpexc.file.read()
+
+        content = attempt_to_decode_response_data(data, response_encoding)
+        return (status, content)
+
+    def createUser(self, user_dict):
+        payload = PAYLOAD_POST_USER.ensure(user_dict)
+
+        return self._issue_POST('/user', request_json=dict(payload))

tests/support/serversupp.py

@@ -41,7 +41,7 @@ class ServerWithinThreadExecutor:
 
     def __init__(self, ServerClass, *args, **kwargs):
         self._serverclass = ServerClass
-        self._serverclass_on_instance = kwargs.pop('on_instance')
+        self._serverclass_on_instance = kwargs.pop('on_instance', None)
         self._arguments = (args, kwargs)
         self._started = ThreadEvent()
         self._thread = None

tests/test_mig_lib_coreapi.py

@@ -0,0 +1,49 @@
+from http.server import HTTPServer, BaseHTTPRequestHandler
+
+from tests.support import MigTestCase, testmain
+from tests.support.serversupp import make_wrapped_server
+
+from mig.lib.coreapi import CoreApiClient
+
+
+class TestRequestHandler(BaseHTTPRequestHandler):
+    def do_POST(self):
+        self.send_response(418)
+        self.end_headers()
+        self.wfile.write(b'')
+
+
+class TestHTTPServer(HTTPServer):
+    def __init__(self, addr, **kwargs):
+        self._on_start = kwargs.pop('on_start', lambda _: None)
+
+        HTTPServer.__init__(self, addr, TestRequestHandler, **kwargs)
+
+    def server_activate(self):
+        HTTPServer.server_activate(self)
+        self._on_start(self)
+
+
+class TestBooleans(MigTestCase):
+    def test_true(self):
+        server_addr = ('localhost', 4567)
+        server = make_wrapped_server(TestHTTPServer, server_addr)
+        server.start_wait_until_ready()
+
+        instance = CoreApiClient("http://%s:%s/" % server_addr)
+
+        status, content = instance.createUser({
+            'full_name': "Test User",
+            'organization': "Test Org",
+            'state': "NA",
+            'country': "DK",
+            'email': "user@example.com",
+            'comment': "This is the create comment",
+            'password': "password",
+        })
+
+        self.assertEqual(status, 418)
+
+
+if __name__ == '__main__':
+    testmain()