Implement finding users by email address.

Author: albu-diku

mig/services/coreapi/server.py

@@ -74,7 +74,7 @@
 from mig.shared.tlsserver import hardened_ssl_context
 from mig.shared.url import urlparse, urlencode, parse_qsl
 from mig.shared.useradm import get_any_oid_user_dn, check_password_scramble, \
-    check_hash
+    check_hash, search_users as useradm_search_users
 from mig.shared.userdb import default_db_path
 from mig.shared.validstring import possible_user_id, is_valid_email_address
 from mig.server.createuser import _main as createuser
@@ -203,6 +203,13 @@ def _is_string_and_non_empty(value):
 ))
 
 
+def search_users(configuration, search_filter):
+    conf_path = configuration.config_file
+    db_path = default_db_path(configuration)
+    _, hits = useradm_search_users(search_filter, conf_path, db_path)
+    return list((obj for _, obj in hits))
+
+
 def validate_payload(definition, payload):
     args = definition(*[payload.get(field, None) for field in definition._fields])
 
@@ -228,6 +235,19 @@ def GET_user():
     def GET_user_username(username):
         return 'FOOBAR'
 
+    @app.get('/user/find')
+    def GET_user_find():
+        query_params = request.args
+
+        objects = search_users(configuration, {
+            'email': query_params['email']
+        })
+
+        if len(objects) != 1:
+            raise http_error_from_status_code(404, None)
+
+        return dict(objects=objects)
+
     @app.post('/user')
     def POST_user():
         payload = request.get_json()

tests/test_mig_services_coreapi.py

@@ -15,7 +15,9 @@
 
 from mig.services.coreapi import ThreadedApiHttpServer, \
     _create_and_expose_server
+from mig.server.createuser import _main as _createuser
 from mig.shared.conf import get_configuration_object
+from mig.shared.url import urlencode
 from mig.shared.useradm import _USERADM_CONFIG_DIR_KEYS
 
 _PYTHON_MAJOR = '2' if PY2 else '3'
@@ -248,5 +250,111 @@ def _on_instance(server):
         return server_thread
 
 
+class MigServerGrid_openid__existing_user(MigTestCase, HtmlAssertMixin):
+    def before_each(self):
+        self.server_addr = None
+        self.server_thread = None
+
+        for config_key in _USERADM_CONFIG_DIR_KEYS:
+            dir_path = getattr(self.configuration, config_key)[0:-1]
+            try:
+                shutil.rmtree(dir_path)
+            except OSError as exc:
+                if exc.errno != errno.ENOENT:  # FileNotFoundError
+                    pass
+
+        _createuser(self.configuration, [
+            "Test User",
+            "Test Org",
+            "NA",
+            "DK",
+            "user@example.com",
+            "This is the create comment",
+            "password"
+        ], default_renew=True)
+        pass
+
+    def _provide_configuration(self):
+        return 'testconfig'
+
+    def after_each(self):
+        if self.server_thread:
+            self.server_thread.stop()
+
+    def issue_request(self, request_path):
+        return self.issue_GET(request_path)
+
+    def issue_GET(self, request_path, query_dict=None, response_encoding='textual'):
+        assert isinstance(request_path, str) and request_path.startswith('/'), "require http path starting with /"
+        request_url =  ''.join(('http://', self.server_addr[0], ':', str(self.server_addr[1]), 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)
+
+    @unittest.skipIf(PY2, "Python 3 only")
+    def test_GET_openid_user_find(self):
+        flask_app = None
+
+        self.server_addr = ('localhost', 4567)
+        self.server_thread = self._make_server(self.configuration, self.logger, self.server_addr)
+        self.server_thread.start_wait_until_ready()
+
+        status, content = self.issue_GET('/user/find', {
+            'email': 'user@example.com'
+        })
+
+        self.assertEqual(status, 200)
+
+        self.assertIsInstance(content, dict)
+        self.assertIn('objects', content)
+        self.assertIsInstance(content['objects'], list)
+
+        user = content['objects'][0]
+        # check we received the correct user
+        self.assertEqual(user['full_name'], 'Test User')
+
+    def _make_configuration(self, test_logger, server_addr, overrides=None):
+        configuration = self.configuration
+        _extend_configuration(
+            configuration,
+            server_addr[0],
+            server_addr[1],
+            logger=test_logger,
+            expandusername=False,
+            host_rsa_key='',
+            nossl=True,
+            show_address=False,
+            show_port=False,
+        )
+        return configuration
+
+    @staticmethod
+    def _make_server(configuration, logger=None, server_address=None):
+        def _on_instance(server):
+            server.server_app = _create_and_expose_server(
+                server, server.configuration)
+
+        (host, port) = server_address
+        server_thread = make_wrapped_server(ThreadedApiHttpServer, \
+            configuration, logger, host, port, on_instance=_on_instance)
+        return server_thread
+
+
 if __name__ == '__main__':
     testmain()