Merge "Use function registration for policy checks"

1 files changed, 113 insertions, 56 deletions

diff --git a/tests/unit/test_policy.py b/tests/unit/test_policy.py
index 449c8a9..e7f391f 100644
--- a/tests/unit/test_policy.py
+++ b/tests/unit/test_policy.py
@@ -157,12 +157,15 @@ class BrainTestCase(unittest.TestCase):
     def test_check_with_specific(self):
         self.spam_called = False
 
+        def check_spam(brain, kind, match, target_dict, cred_dict):
+            self.assertEqual(kind, "spam")
+            self.assertEqual(match, "check")
+            self.assertEqual(target_dict, "target")
+            self.assertEqual(cred_dict, "credentials")
+            self.spam_called = True
+
         class TestBrain(policy.Brain):
-            def _check_spam(inst, match, target_dict, cred_dict):
-                self.assertEqual(match, "check")
-                self.assertEqual(target_dict, "target")
-                self.assertEqual(cred_dict, "credentials")
-                self.spam_called = True
+            _checks = dict(spam=check_spam)
 
         brain = TestBrain()
         result = brain._check("spam:check", "target", "credentials")
@@ -172,18 +175,45 @@ class BrainTestCase(unittest.TestCase):
     def test_check_with_generic(self):
         self.generic_called = False
 
+        def check_generic(brain, kind, match, target_dict, cred_dict):
+            self.assertEqual(kind, "spam")
+            self.assertEqual(match, "check")
+            self.assertEqual(target_dict, "target")
+            self.assertEqual(cred_dict, "credentials")
+            self.generic_called = True
+
         class TestBrain(policy.Brain):
-            def _check_generic(inst, match, target_dict, cred_dict):
-                self.assertEqual(match, "spam:check")
-                self.assertEqual(target_dict, "target")
-                self.assertEqual(cred_dict, "credentials")
-                self.generic_called = True
+            _checks = {None: check_generic}
 
         brain = TestBrain()
         result = brain._check("spam:check", "target", "credentials")
 
         self.assertEqual(self.generic_called, True)
 
+    def test_check_with_inheritance(self):
+        self.inherited_called = False
+
+        class TestBrain(policy.Brain):
+            _checks = {}
+
+            def _check_inherited(inst, match, target_dict, cred_dict):
+                self.assertEqual(match, "check")
+                self.assertEqual(target_dict, "target")
+                self.assertEqual(cred_dict, "credentials")
+                self.inherited_called = True
+
+        brain = TestBrain()
+        result = brain._check("inherited:check", "target", "credentials")
+
+        self.assertEqual(self.inherited_called, True)
+
+    def test_check_no_handler(self):
+        class TestBrain(policy.Brain):
+            _checks = {}
+
+        brain = TestBrain()
+        self.assertEqual(brain._check('spam:mer', 'target', 'cred'), False)
+
     def test_check_empty(self):
         class TestBrain(policy.Brain):
             def _check(inst, match, target_dict, cred_dict):
@@ -274,6 +304,52 @@ class BrainTestCase(unittest.TestCase):
         self.assertEqual(self.targets, ["target", "target", "target"])
         self.assertEqual(self.creds, ["creds", "creds", "creds"])
 
+
+class CheckRegisterTestCase(unittest.TestCase):
+    def setUp(self):
+        self.brain_checks = policy.Brain._checks
+        policy.Brain._checks = {}
+
+    def tearDown(self):
+        policy.Brain._checks = self.brain_checks
+
+    def test_class_register(self):
+        policy.Brain._register('spam', 'func')
+        policy.Brain._register('spammer', 'funcer')
+
+        self.assertEqual(policy.Brain._checks,
+                         dict(spam='func', spammer='funcer'))
+
+    def test_register_func(self):
+        policy.register('spam', 'func')
+
+        self.assertEqual(policy.Brain._checks,
+                         dict(spam='func'))
+
+    def test_register_decorator(self):
+        @policy.register('spam')
+        def test_func():
+            pass
+
+        self.assertEqual(policy.Brain._checks,
+                         dict(spam=test_func))
+
+
+class CheckTestCase(unittest.TestCase):
+    def setUp(self):
+        self.urlopen_result = ""
+
+        def fake_urlopen(url, post_data):
+            self.url = url
+            self.post_data = post_data
+            return StringIO.StringIO(self.urlopen_result)
+
+        self.patcher = mock.patch.object(urllib2, "urlopen", fake_urlopen)
+        self.patcher.start()
+
+    def tearDown(self):
+        self.patcher.stop()
+
     def stub__check_rule(self, rules=None, default_rule=None):
         self.check_called = False
 
@@ -288,21 +364,21 @@ class BrainTestCase(unittest.TestCase):
 
     def test_rule_no_rules_no_default(self):
         brain = self.stub__check_rule()
-        result = brain._check_rule("spam", "target", "creds")
+        result = policy._check_rule(brain, "rule", "spam", "target", "creds")
 
         self.assertEqual(result, False)
         self.assertEqual(self.check_called, False)
 
     def test_rule_no_rules_default(self):
         brain = self.stub__check_rule(default_rule="spam")
-        result = brain._check_rule("spam", "target", "creds")
+        result = policy._check_rule(brain, "rule", "spam", "target", "creds")
 
         self.assertEqual(result, False)
         self.assertEqual(self.check_called, False)
 
     def test_rule_no_rules_non_default(self):
         brain = self.stub__check_rule(default_rule="spam")
-        result = brain._check_rule("python", "target", "creds")
+        result = policy._check_rule(brain, "rule", "python", "target", "creds")
 
         self.assertEqual(self.check_called, True)
         self.assertEqual(result, ("rule:spam",))
@@ -311,7 +387,7 @@ class BrainTestCase(unittest.TestCase):
 
     def test_rule_with_rules(self):
         brain = self.stub__check_rule(rules=dict(spam=["hiho:ni"]))
-        result = brain._check_rule("spam", "target", "creds")
+        result = policy._check_rule(brain, "rule", "spam", "target", "creds")
 
         self.assertEqual(self.check_called, True)
         self.assertEqual(result, ["hiho:ni"])
@@ -319,57 +395,38 @@ class BrainTestCase(unittest.TestCase):
         self.assertEqual(self.cred, "creds")
 
     def test_role_no_match(self):
-        brain = policy.Brain()
-        result = brain._check_role("SpAm", {}, dict(roles=["a", "b", "c"]))
+        result = policy._check_role(None, "role", "SpAm", {},
+                                    dict(roles=["a", "b", "c"]))
 
         self.assertEqual(result, False)
 
     def test_role_with_match(self):
-        brain = policy.Brain()
-        result = brain._check_role("SpAm", {}, dict(roles=["a", "b", "sPaM"]))
+        result = policy._check_role(None, "role", "SpAm", {},
+                                    dict(roles=["a", "b", "sPaM"]))
 
         self.assertEqual(result, True)
 
     def test_generic_no_key(self):
-        brain = policy.Brain()
-        result = brain._check_generic("tenant:%(tenant_id)s",
-                                      dict(tenant_id="spam"),
-                                      {})
+        result = policy._check_generic(None, "tenant", "%(tenant_id)s",
+                                       dict(tenant_id="spam"),
+                                       {})
 
         self.assertEqual(result, False)
 
     def test_generic_with_key_mismatch(self):
-        brain = policy.Brain()
-        result = brain._check_generic("tenant:%(tenant_id)s",
-                                      dict(tenant_id="spam"),
-                                      dict(tenant="nospam"))
+        result = policy._check_generic(None, "tenant", "%(tenant_id)s",
+                                       dict(tenant_id="spam"),
+                                       dict(tenant="nospam"))
 
         self.assertEqual(result, False)
 
     def test_generic_with_key_match(self):
-        brain = policy.Brain()
-        result = brain._check_generic("tenant:%(tenant_id)s",
-                                      dict(tenant_id="spam"),
-                                      dict(tenant="spam"))
+        result = policy._check_generic(None, "tenant", "%(tenant_id)s",
+                                       dict(tenant_id="spam"),
+                                       dict(tenant="spam"))
 
         self.assertEqual(result, True)
 
-
-class HttpBrainTestCase(unittest.TestCase):
-    def setUp(self):
-        self.urlopen_result = ""
-
-        def fake_urlopen(url, post_data):
-            self.url = url
-            self.post_data = post_data
-            return StringIO.StringIO(self.urlopen_result)
-
-        self.patcher = mock.patch.object(urllib2, "urlopen", fake_urlopen)
-        self.patcher.start()
-
-    def tearDown(self):
-        self.patcher.stop()
-
     def decode_post_data(self):
         result = {}
         for item in self.post_data.split('&'):
@@ -379,26 +436,26 @@ class HttpBrainTestCase(unittest.TestCase):
         return result
 
     def test_http_false(self):
-        brain = policy.HttpBrain()
-        result = brain._check_http("//spam.example.org/%(tenant)s",
-                                   dict(tenant="spam"),
-                                   dict(roles=["a", "b", "c"]))
+        result = policy._check_http(None, "http",
+                                    "//spam.example.org/%(tenant)s",
+                                    dict(tenant="spam"),
+                                    dict(roles=["a", "b", "c"]))
 
         self.assertEqual(result, False)
-        self.assertEqual(self.url, "//spam.example.org/spam")
+        self.assertEqual(self.url, "http://spam.example.org/spam")
         self.assertEqual(self.decode_post_data(), dict(
                          target=dict(tenant="spam"),
                          credentials=dict(roles=["a", "b", "c"])))
 
     def test_http_true(self):
         self.urlopen_result = "True"
-        brain = policy.HttpBrain()
-        result = brain._check_http("//spam.example.org/%(tenant)s",
-                                   dict(tenant="spam"),
-                                   dict(roles=["a", "b", "c"]))
+        result = policy._check_http(None, "http",
+                                    "//spam.example.org/%(tenant)s",
+                                    dict(tenant="spam"),
+                                    dict(roles=["a", "b", "c"]))
 
         self.assertEqual(result, True)
-        self.assertEqual(self.url, "//spam.example.org/spam")
+        self.assertEqual(self.url, "http://spam.example.org/spam")
         self.assertEqual(self.decode_post_data(), dict(
                          target=dict(tenant="spam"),
                          credentials=dict(roles=["a", "b", "c"])))