formatted code
diff --git a/casbin_sqlobject_adapter/adapter.py b/casbin_sqlobject_adapter/adapter.py index 1001474..fc33f35 100644 --- a/casbin_sqlobject_adapter/adapter.py +++ b/casbin_sqlobject_adapter/adapter.py
@@ -1,13 +1,12 @@ from casbin import persist -from sqlobject import * +from sqlobject import SQLObject, StringCol, sqlhub, connectionForURI class CasbinRule(SQLObject): - class sqlmeta: - table = 'casbin_rule' - + table = "casbin_rule" + ptype = StringCol(length=255) v0 = StringCol(length=255, default=None) v1 = StringCol(length=255, default=None) @@ -22,13 +21,10 @@ if v is None: break arr.append(v) - return ', '.join(arr) - + return ", ".join(arr) + def __repr__(self): return '<CasbinRule {}: "{}">'.format(self.id, str(self)) - - - class Adapter(persist.Adapter): @@ -37,20 +33,18 @@ def __init__(self, connection_string): self._conhandler = connectionForURI(connection_string) sqlhub.processConnection = self._conhandler - def load_policy(self, model): """loads all policy rules from the storage.""" count = CasbinRule.select().count() - for i in range(1,1+count): + for i in range(1, 1 + count): line = CasbinRule.get(i) persist.load_policy_line(str(line), model) def _save_policy_line(self, ptype, rule): line = CasbinRule.selectBy(ptype=ptype) for i, v in enumerate(rule): - setattr(line, 'v{}'.format(i), v) - + setattr(line, "v{}".format(i), v) def save_policy(self, model): """saves all policy rules to the storage.""" @@ -60,13 +54,12 @@ for ptype, ast in model.model[sec].items(): for rule in ast.policy: self._save_policy_line(ptype, rule) - + return True def add_policy(self, sec, ptype, rule): """adds a policy rule to the storage.""" self._save_policy_line(ptype, rule) - def remove_policy(self, sec, ptype, rule): """removes a policy rule from the storage.""" @@ -77,6 +70,3 @@ This is part of the Auto-Save feature. """ pass - - -
diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 70538bb..56f89a9 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py
@@ -13,60 +13,54 @@ def get_enforcer(): - con_string = 'sqlite:/:memory:' + con_string = "sqlite:/:memory:" adapter = Adapter(con_string) CasbinRule.createTable(ifNotExists=True) - CasbinRule(ptype='p', v0='alice', v1='data1', v2='read') - CasbinRule(ptype='p', v0='bob', v1='data2', v2='write') - CasbinRule(ptype='p', v0='data2_admin', v1='data2', v2='read') - CasbinRule(ptype='p', v0='data2_admin', v1='data2', v2='write') - CasbinRule(ptype='g', v0='alice', v1='data2_admin') + CasbinRule(ptype="p", v0="alice", v1="data1", v2="read") + CasbinRule(ptype="p", v0="bob", v1="data2", v2="write") + CasbinRule(ptype="p", v0="data2_admin", v1="data2", v2="read") + CasbinRule(ptype="p", v0="data2_admin", v1="data2", v2="write") + CasbinRule(ptype="g", v0="alice", v1="data2_admin") - return casbin.Enforcer(get_fixture('rbac_model.conf'), adapter) + return casbin.Enforcer(get_fixture("rbac_model.conf"), adapter) class TestConfig(unittest.TestCase): - def test_enforcer_basic(self): e = get_enforcer() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('bob', 'data1', 'read')) - self.assertTrue(e.enforce('bob', 'data2', 'write')) - self.assertTrue(e.enforce('alice', 'data2', 'read')) - self.assertTrue(e.enforce('alice', 'data2', 'write')) + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("bob", "data1", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertTrue(e.enforce("alice", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) def test_add_policy(self): e = get_enforcer() - self.assertFalse(e.enforce('eve', 'data3', 'read')) - res = e.add_permission_for_user('eve', 'data3', 'read') + self.assertFalse(e.enforce("eve", "data3", "read")) + res = e.add_permission_for_user("eve", "data3", "read") self.assertTrue(res) - self.assertTrue(e.enforce('eve', 'data3', 'read')) + self.assertTrue(e.enforce("eve", "data3", "read")) def test_save_policy(self): e = get_enforcer() - self.assertFalse(e.enforce('alice', 'data4', 'read')) + self.assertFalse(e.enforce("alice", "data4", "read")) model = e.get_model() model.clear_policy() - model.add_policy('p', 'p', ['alice', 'data4', 'read']) + model.add_policy("p", "p", ["alice", "data4", "read"]) adapter = e.get_adapter() adapter.save_policy(model) - self.assertTrue(e.enforce('alice', 'data4', 'read')) + self.assertTrue(e.enforce("alice", "data4", "read")) def test_str(self): - rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read') - self.assertEqual(str(rule), 'p, alice, data1, read') - + rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read") + self.assertEqual(str(rule), "p, alice, data1, read") def test_repr(self): - rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read') + rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read") self.assertRegex(repr(rule), r'<CasbinRule \d+: "p, alice, data1, read">') - -if(__name__=='__main__'): - unittest.main() -