Changeset 10478

Timestamp:

12 Aug 2013, 08:52:26 (11 years ago)

Author:

uli

Message:

Kofa authenticator now really contacts Kofa instances for authentication.

Location:

main/waeup.cas/trunk/waeup/cas

Files:

• authenticators.py (modified) (3 diffs)
• tests/test_authenticators.py (modified) (4 diffs)

main/waeup.cas/trunk/waeup/cas/authenticators.py

@@ -3 +3 @@
 import re
 from pkg_resources import iter_entry_points
+try:
+    import xmlrpclib                     # Python 2.x
+except ImportError:                      # pragma: no cover
+    import xmlrpc.client as xmlrpclib    # Python 3.x
 
 
@@ -112 +116 @@
 
 
+#: Regular expression matching a starting university marker like
+#: the string 'MA-' in 'MA-M121212' or 'MA-' in 'MA-APP-13123'
+RE_SCHOOL_MARKER = re.compile('^[^\-]+-')
+
+
 class KofaAuthenticator(Authenticator):
     """Authenticate against a running Kofa instance.
@@ -148 +157 @@
         """Do the real check.
         """
-        return False, 'Not implemented.'
+        for backend_name, backend in self.backends.items():
+            if not re.match(backend['marker'], username):
+                continue
+            # remove school marker
+            username = RE_SCHOOL_MARKER.sub('', username)
+            proxy = xmlrpclib.ServerProxy(
+                backend['url'], allow_none=True)
+            valid = proxy.check_credentials(username, password)
+            if valid is not None:
+                return (True, '')
+        return (False, 'Invalid username or password.')

main/waeup.cas/trunk/waeup/cas/tests/test_authenticators.py

@@ +1 @@
+# Tests for waeup.cas.authentictors
 import os
+import threading
 import unittest
 from paste.deploy import loadapp
+try:
+    from SimpleXMLRPCServer import SimpleXMLRPCServer  # Python 2.x
+except ImportError:
+    from xmlrpc.server import SimpleXMLRPCServer       # Python 3.x
+try:
+    import xmlrpclib                     # Python 2.x
+except ImportError:
+    import xmlrpc.client as xmlrpclib    # Python 3.x
 from waeup.cas.authenticators import (
     get_all_authenticators, get_authenticator, filter_auth_opts,
@@ -59 +69 @@
 
 
-BACKENDS = dict(
+BACKENDS1 = dict(
     inst1=dict(
         url='http://localhost:6666/app',
-        marker='^M[0-9]+$',
+        marker='^MA-',
+        )
+    )
+
+BACKENDS2 = dict(
+    inst1=dict(
+        url='http://localhost:6666/',
+        marker='^SCHOOL1-',
         )
     )
@@ -81 +98 @@
     def test_options(self):
         # we can pass options
-        auth = KofaAuthenticator(auth_backends=str(BACKENDS))
-        assert auth.backends == BACKENDS
+        auth = KofaAuthenticator(auth_backends=str(BACKENDS1))
+        assert auth.backends == BACKENDS1
         auth = KofaAuthenticator(auth_backends='{"foo": {"url": "bar"}}')
         assert auth.backends['foo']['marker'] == '.+'
@@ -108 +125 @@
         assert isinstance(app.auth, KofaAuthenticator)
 
-    def DIStest_check_credentials(self):
+
+class FakeKofaServer(SimpleXMLRPCServer):
+    # A fake Kofa server that provides only XMLRPC methods
+
+    allow_reuse_address = True
+
+    def __init__(self, *args, **kw):
+        kw.update(allow_none=True)
+        SimpleXMLRPCServer.__init__(self, *args, **kw)
+        self.register_function(self._check_credentials, 'check_credentials')
+
+    def _check_credentials(self, username, password):
+        # fake waeup.kofa check_credentials method.
+        #
+        # This method is supposed to mimic the behaviour of an
+        # original waeup.kofa check_credentials method. It returns a
+        # positive result for the credentials `bird`/`bebop`.
+        if username == 'bird' and password == 'bebop':
+            return {'id': 'bird', 'email': 'bird@gods.net',
+                    'description': 'Mr. Charles Parker'}
+        return None
+
+
+class XMLRPCFakeKofaTestCase(unittest.TestCase):
+    # A test case where a fake Kofa server is started before tests (and
+    # shut down afterwards).
+
+    server = None
+    th = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.server = FakeKofaServer(('localhost', 6666))
+        cls.th = threading.Thread(target=cls.server.serve_forever)
+        cls.th.daemon = True
+        cls.th.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.server.shutdown()
+        cls.server.server_close()
+
+
+class MyCase(XMLRPCFakeKofaTestCase):
+
+    def test_fake_kofa_works(self):
+        # make sure the local fake kofa works
+        proxy = xmlrpclib.ServerProxy("http://localhost:6666", allow_none=True)
+        result = proxy.check_credentials('bird', 'bebop')
+        assert result == {
+            'description': 'Mr. Charles Parker',
+            'email': 'bird@gods.net',
+            'id': 'bird'}
+        return
+
+    def test_check_credentials(self):
         # we get real responses when querying Kofa instances
-        auth = KofaAuthenticator()
-        result1 = auth.check_credentials('bird', 'bebop')
+        auth = KofaAuthenticator(auth_backends=str(BACKENDS2))
+        result1 = auth.check_credentials('SCHOOL1-bird', 'bebop')
         assert result1 == (True, '')
-        result2 = auth.check_credentials('foo', 'bar')
+        result2 = auth.check_credentials('SCHOOL1-foo', 'bar')
         assert result2 == (False, 'Invalid username or password.')
+        result3 = auth.check_credentials('SCHOOL2-bar', 'baz')
+        assert result3 == (False, 'Invalid username or password.')