Modularize swift-auth CGI script, add unit tests

- Moved most of swift-auth CGI script to kerbauth_utils.py
- Added unit tests for kerbauth_utils.py
- Made MEMCACHE_SERVERS, DEBUG_HEADERS, TOKEN_LIFE as
  configurable parameters

Change-Id: I2e9e9823e8aa99dc2cf41327c55428350c8768dc
Signed-off-by: Prashanth Pai <ppai@redhat.com>
Reviewed-on: http://review.gluster.org/6248
Tested-by: Chetan Risbud <crisbud@redhat.com>
Reviewed-by: Chetan Risbud <crisbud@redhat.com>
Reviewed-by: Luis Pabon <lpabon@redhat.com>
Tested-by: Luis Pabon <lpabon@redhat.com>

3 files changed, 122 insertions, 29 deletions

diff --git a/test/unit/__init__.py b/test/unit/__init__.py
index e69de29..3e26378 100644
--- a/test/unit/__init__.py
+++ b/test/unit/__init__.py
@@ -0,0 +1,44 @@
+# Copyright (c) 2013 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from contextlib import contextmanager
+
+
+class FakeMemcache(object):
+    """A Fake class to emulate memcache."""
+
+    def __init__(self):
+        self.store = {}
+
+    def get(self, key):
+        return self.store.get(key)
+
+    def set(self, key, value, timeout=0):
+        self.store[key] = value
+        return True
+
+    def incr(self, key, time=0):
+        self.store[key] = self.store.setdefault(key, 0) + 1
+        return self.store[key]
+
+    @contextmanager
+    def soft_lock(self, key, timeout=0, retries=5):
+        yield True
+
+    def delete(self, key):
+        try:
+            del self.store[key]
+        except Exception:
+            pass
+        return True
diff --git a/test/unit/test_kerbauth.py b/test/unit/test_kerbauth.py
index 446abb8..1771314 100644
--- a/test/unit/test_kerbauth.py
+++ b/test/unit/test_kerbauth.py
@@ -14,10 +14,10 @@
 # limitations under the License.
 
 import unittest
-from contextlib import contextmanager
 from time import time
 
 from swiftkerbauth import kerbauth as auth
+from test.unit import FakeMemcache
 from swift.common.swob import Request, Response
 
 EXT_AUTHENTICATION_URL = "127.0.0.1"
@@ -46,34 +46,6 @@ def unpatch_filter_factory():
     reload(auth)
 
 
-class FakeMemcache(object):
-
-    def __init__(self):
-        self.store = {}
-
-    def get(self, key):
-        return self.store.get(key)
-
-    def set(self, key, value, time=0):
-        self.store[key] = value
-        return True
-
-    def incr(self, key, time=0):
-        self.store[key] = self.store.setdefault(key, 0) + 1
-        return self.store[key]
-
-    @contextmanager
-    def soft_lock(self, key, timeout=0, retries=5):
-        yield True
-
-    def delete(self, key):
-        try:
-            del self.store[key]
-        except Exception:
-            pass
-        return True
-
-
 class FakeApp(object):
 
     def __init__(self, status_headers_body_iter=None, acl=None, sync_key=None):
diff --git a/test/unit/test_kerbauth_utils.py b/test/unit/test_kerbauth_utils.py
new file mode 100644
index 0000000..abf8ad0
--- /dev/null
+++ b/test/unit/test_kerbauth_utils.py
@@ -0,0 +1,77 @@
+# Copyright (c) 2013 Red Hat, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+import re
+from time import time
+from test.unit import FakeMemcache
+from swiftkerbauth import kerbauth_utils as ku
+
+
+class TestKerbUtils(unittest.TestCase):
+
+    def test_get_remote_user(self):
+        env = {'REMOTE_USER': "auth_admin@EXAMPLE.COM"}
+        result = ku.get_remote_user(env)
+        self.assertEqual(result, "auth_admin")
+
+    def test_get_remote_user_err(self):
+        env = {'REMOTE_USER': "auth_admin"}
+        try:
+            ku.get_remote_user(env)
+        except RuntimeError as err:
+            self.assertTrue(err.args[0].startswith("Malformed REMOTE_USER"))
+        else:
+            self.fail("Expected RuntimeError")
+
+    def test_get_auth_data(self):
+        mc = FakeMemcache()
+        expiry = time() + 100
+        ku.set_auth_data(mc, "root", "AUTH_tk", expiry, "root,admin")
+        (token, expires, groups) = ku.get_auth_data(mc, "root")
+        self.assertEqual(("AUTH_tk", expiry, "root,admin"),
+                         (token, expires, groups))
+
+    def test_get_auth_data_err(self):
+        mc = FakeMemcache()
+        (token, expires, groups) = ku.get_auth_data(mc, "root")
+        self.assertEqual((token, expires, groups), (None, None, None))
+
+        expiry = time() - 1
+        ku.set_auth_data(mc, "root", "AUTH_tk", expiry, "root,admin")
+        (token, expires, groups) = ku.get_auth_data(mc, "root")
+        self.assertEqual((token, expires, groups), (None, None, None))
+
+    def test_set_auth_data(self):
+        mc = FakeMemcache()
+        expiry = time() + 100
+        ku.set_auth_data(mc, "root", "AUTH_tk", expiry, "root,admin")
+
+    def test_generate_token(self):
+        token = ku.generate_token()
+        matches = re.match('AUTH_tk[a-f0-9]{32}', token)
+        self.assertIsNotNone(matches)
+
+    def test_get_groups(self):
+        groups = ku.get_groups("root")
+        self.assertIn("root", groups)
+
+    def test_get_groups_err(self):
+        try:
+            ku.get_groups("Zroot")
+        except RuntimeError as err:
+            self.assertTrue(err.args[0].startswith("Failure running id -G"))
+        else:
+            self.fail("Expected RuntimeError")