From d3cb83a629c69f320a32edccbf78e5089086fdbf Mon Sep 17 00:00:00 2001 From: ffyuanda <46557895+ffyuanda@users.noreply.github.com> Date: Wed, 28 Jul 2021 23:19:37 +0800 Subject: [PATCH] ci: code reformatted and linter tests added (#11) Signed-off-by: ffyuanda <46557895+ffyuanda@users.noreply.github.com> --- .github/workflows/main.yml | 14 ++ casbin_pymongo_adapter/__init__.py | 2 +- casbin_pymongo_adapter/adapter.py | 64 ++++---- setup.py | 13 +- tests/test_adapter.py | 238 +++++++++++++++-------------- 5 files changed, 187 insertions(+), 144 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index c458ae1..bda382b 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -46,6 +46,20 @@ jobs: COVERALLS_FLAG_NAME: ${{ matrix.os }} - ${{ matrix.python-version }} COVERALLS_PARALLEL: true + lint: + name: Run Linters + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v2 + + - name: Super-Linter + uses: github/super-linter@v4.2.2 + env: + VALIDATE_PYTHON_BLACK: true + DEFAULT_BRANCH: master + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + coveralls: name: Indicate completion to coveralls.io needs: test diff --git a/casbin_pymongo_adapter/__init__.py b/casbin_pymongo_adapter/__init__.py index 5f6eba2..a7dde9b 100644 --- a/casbin_pymongo_adapter/__init__.py +++ b/casbin_pymongo_adapter/__init__.py @@ -1 +1 @@ -from .adapter import Adapter \ No newline at end of file +from .adapter import Adapter diff --git a/casbin_pymongo_adapter/adapter.py b/casbin_pymongo_adapter/adapter.py index 7cea986..2ee8ae3 100644 --- a/casbin_pymongo_adapter/adapter.py +++ b/casbin_pymongo_adapter/adapter.py @@ -1,12 +1,15 @@ from casbin import persist from pymongo import MongoClient + class CasbinRule: - ''' + """ CasbinRule model - ''' + """ - def __init__(self, ptype = None, v0 = None, v1 = None, v2 = None, v3 = None, v4 = None, v5 = None): + def __init__( + self, ptype=None, v0=None, v1=None, v2=None, v3=None, v4=None, v5=None + ): self.ptype = ptype self.v0 = v0 self.v1 = v1 @@ -16,20 +19,25 @@ def __init__(self, ptype = None, v0 = None, v1 = None, v2 = None, v3 = None, v4 self.v5 = v5 def dict(self): - d = {'ptype': self.ptype} + d = {"ptype": self.ptype} for value in dir(self): - if getattr(self, value) is not None and value.startswith('v') and value[1:].isnumeric(): + if ( + getattr(self, value) is not None + and value.startswith("v") + and value[1:].isnumeric() + ): d[value] = getattr(self, value) return d def __str__(self): - return ', '.join(self.dict().values()) + return ", ".join(self.dict().values()) def __repr__(self): return ''.format(str(self)) + class Adapter(persist.Adapter): """the interface for Casbin adapters.""" @@ -37,7 +45,7 @@ def __init__(self, uri, dbname, collection="casbin_rule"): """Create an adapter for Mongodb Args: - uri (str): This should be the same requiement as pymongo Client's 'uri' parameter. + uri (str): This should be the same requiement as pymongo Client's 'uri' parameter. See https://pymongo.readthedocs.io/en/stable/api/pymongo/mongo_client.html#pymongo.mongo_client.MongoClient. dbname (str): Database to store policy. collection (str, optional): Collection of the choosen database. Defaults to "casbin_rule". @@ -54,9 +62,9 @@ def load_policy(self, model): """ for line in self._collection.find(): - if 'ptype' not in line: + if "ptype" not in line: continue - rule = CasbinRule(line['ptype']) + rule = CasbinRule(line["ptype"]) for key, value in line.items(): setattr(rule, key, value) @@ -65,20 +73,20 @@ def load_policy(self, model): def _save_policy_line(self, ptype, rule): line = CasbinRule(ptype=ptype) for index, value in enumerate(rule): - setattr(line, f'v{index}', value) + setattr(line, f"v{index}", value) self._collection.insert_one(line.dict()) - + def _find_policy_lines(self, ptype, rule): line = CasbinRule(ptype=ptype) for index, value in enumerate(rule): - setattr(line, f'v{index}', value) + setattr(line, f"v{index}", value) return self._collection.find(line.dict()) - + def _delete_policy_lines(self, ptype, rule): line = CasbinRule(ptype=ptype) for index, value in enumerate(rule): - setattr(line, f'v{index}', value) - + setattr(line, f"v{index}", value) + # if rule is empty, do nothing # else find all given rules and delete them if len(line.dict()) == 0: @@ -87,16 +95,19 @@ def _delete_policy_lines(self, ptype, rule): line_dict = line.dict() line_dict_keys_len = len(line_dict) results = self._collection.find(line_dict) - to_delete = [result['_id'] for result in results - if line_dict_keys_len == len(result.keys()) - 1] - results = self._collection.delete_many({'_id' : {'$in': to_delete}}) + to_delete = [ + result["_id"] + for result in results + if line_dict_keys_len == len(result.keys()) - 1 + ] + results = self._collection.delete_many({"_id": {"$in": to_delete}}) return results.deleted_count - + def save_policy(self, model) -> bool: """Implement add Interface for casbin. Save the policy in mongodb Args: - model (Class Model): Casbin Model which loads from .conf file usually. + model (Class Model): Casbin Model which loads from .conf file usually. Returns: bool: True if succeed @@ -129,7 +140,7 @@ def remove_policy(self, sec, ptype, rule): Args: ptype (str): Policy type, 'g', 'g2', 'p', etc. rule (CasbinRule): Casbin rule if it is exactly same as will be removed. - + Returns: Number: Number of policies be removed """ @@ -144,17 +155,18 @@ def remove_filtered_policy(self, sec, ptype, field_index, *field_values): ptype (str): Policy type, 'g', 'g2', 'p', etc. rule (CasbinRule): Casbin rule will be removed field_index (int): The policy index at which the filed_values begins filtering. Its range is [0, 5] - field_values(List[str]): A list of rules to filter policy which starts from + field_values(List[str]): A list of rules to filter policy which starts from Returns: bool: True if succeed else False """ - if not (0 <= field_index <=5): + if not (0 <= field_index <= 5): return False if not (1 <= field_index + len(field_values) <= 6): return False - query = {f'v{index + field_index}' : value - for index, value in enumerate(field_values)} - query['ptype'] = ptype + query = { + f"v{index + field_index}": value for index, value in enumerate(field_values) + } + query["ptype"] = ptype results = self._collection.delete_many(query) return results.deleted_count > 0 diff --git a/setup.py b/setup.py index 08555df..dccba64 100644 --- a/setup.py +++ b/setup.py @@ -24,7 +24,16 @@ long_description=long_description, long_description_content_type="text/markdown", url="https://github.com/pycasbin/pymongo-adapter", - keywords=["casbin", "pymongo", "casbin-adapter", "rbac", "access control", "abac", "acl", "permission"], + keywords=[ + "casbin", + "pymongo", + "casbin-adapter", + "rbac", + "access control", + "abac", + "acl", + "permission", + ], packages=find_packages(), install_requires=install_requires, python_requires=">=3.6", @@ -37,4 +46,4 @@ "Operating System :: OS Independent", ], data_files=[desc_file], -) \ No newline at end of file +) diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 44af567..19771cd 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -7,220 +7,228 @@ def get_fixture(path): - ''' + """ get model path - ''' + """ dir_path = os.path.split(os.path.realpath(__file__))[0] + "/" return os.path.abspath(dir_path + path) def get_enforcer(): - adapter = Adapter('mongodb://localhost:27017', 'casbin_test') - e = casbin.Enforcer(get_fixture('rbac_model.conf'), adapter) + adapter = Adapter("mongodb://localhost:27017", "casbin_test") + e = casbin.Enforcer(get_fixture("rbac_model.conf"), adapter) model = e.get_model() model.clear_policy() - model.add_policy('p', 'p', ['alice', 'data1', 'read']) + model.add_policy("p", "p", ["alice", "data1", "read"]) adapter.save_policy(model) model.clear_policy() - model.add_policy('p', 'p', ['bob', 'data2', 'write']) + model.add_policy("p", "p", ["bob", "data2", "write"]) adapter.save_policy(model) model.clear_policy() - model.add_policy('p', 'p', ['data2_admin', 'data2', 'read']) + model.add_policy("p", "p", ["data2_admin", "data2", "read"]) adapter.save_policy(model) model.clear_policy() - model.add_policy('p', 'p', ['data2_admin', 'data2', 'write']) + model.add_policy("p", "p", ["data2_admin", "data2", "write"]) adapter.save_policy(model) model.clear_policy() - model.add_policy('g', 'g', ['alice', 'data2_admin']) + model.add_policy("g", "g", ["alice", "data2_admin"]) adapter.save_policy(model) - return casbin.Enforcer(get_fixture('rbac_model.conf'), adapter) + return casbin.Enforcer(get_fixture("rbac_model.conf"), adapter) + def clear_db(dbname): - client = MongoClient('mongodb://localhost:27017') + client = MongoClient("mongodb://localhost:27017") client.drop_database(dbname) class TestConfig(TestCase): - ''' + """ unittest - ''' + """ + def setUp(self): - clear_db('casbin_test') + clear_db("casbin_test") def tearDown(self): - clear_db('casbin_test') + clear_db("casbin_test") def test_enforcer_basic(self): - ''' + """ test policy - ''' + """ e = get_enforcer() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('alice', 'data1', 'write')) - self.assertFalse(e.enforce('bob', 'data2', '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("alice", "data1", "write")) + self.assertFalse(e.enforce("bob", "data2", "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): - ''' + """ test add_policy - ''' + """ e = get_enforcer() adapter = e.get_adapter() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('alice', 'data1', 'write')) - self.assertFalse(e.enforce('bob', 'data2', '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("alice", "data1", "write")) + self.assertFalse(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertTrue(e.enforce("alice", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) + # test add_policy after insert 2 rules - adapter.add_policy(sec='p', ptype='p', rule=('alice', 'data1', 'write')) - adapter.add_policy(sec='p', ptype='p', rule=('bob', 'data2', 'read')) + adapter.add_policy(sec="p", ptype="p", rule=("alice", "data1", "write")) + adapter.add_policy(sec="p", ptype="p", rule=("bob", "data2", "read")) # reload policies from database e.load_policy() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertTrue(e.enforce('alice', 'data1', 'write')) - self.assertTrue(e.enforce('bob', 'data2', '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.assertTrue(e.enforce("alice", "data1", "write")) + self.assertTrue(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertTrue(e.enforce("alice", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) + def test_remove_policy(self): - ''' + """ test remove_policy - ''' + """ e = get_enforcer() adapter = e.get_adapter() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('alice', 'data1', 'write')) - self.assertFalse(e.enforce('bob', 'data2', '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("alice", "data1", "write")) + self.assertFalse(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertTrue(e.enforce("alice", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) + # test remove_policy after delete a role definition - result = adapter.remove_policy(sec='g', ptype='g', rule=('alice', 'data2_admin')) + result = adapter.remove_policy( + sec="g", ptype="g", rule=("alice", "data2_admin") + ) # reload policies from database e.load_policy() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('alice', 'data1', 'write')) - self.assertFalse(e.enforce('bob', 'data2', 'read')) - self.assertTrue(e.enforce('bob', 'data2', 'write')) - self.assertFalse(e.enforce('alice', 'data2', 'read')) - self.assertFalse(e.enforce('alice', 'data2', 'write')) + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("alice", "data1", "write")) + self.assertFalse(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertFalse(e.enforce("alice", "data2", "read")) + self.assertFalse(e.enforce("alice", "data2", "write")) self.assertTrue(result) def test_remove_policy_no_remove_when_rule_is_incomplete(self): - adapter = Adapter('mongodb://localhost:27017', 'casbin_test') - e = casbin.Enforcer(get_fixture('rbac_with_resources_roles.conf'), adapter) - - adapter.add_policy(sec='p', ptype='p', rule=('alice', 'data1', 'write')) - adapter.add_policy(sec='p', ptype='p', rule=('alice', 'data1', 'read')) - adapter.add_policy(sec='p', ptype='p', rule=('bob', 'data2', 'read')) - adapter.add_policy(sec='p', ptype='p', rule=('data_group_admin', 'data_group', 'write')) - adapter.add_policy(sec='g', ptype='g', rule=('alice', 'data_group_admin')) - adapter.add_policy(sec='g', ptype='g2', rule=('data2', 'data_group')) - + adapter = Adapter("mongodb://localhost:27017", "casbin_test") + e = casbin.Enforcer(get_fixture("rbac_with_resources_roles.conf"), adapter) + + adapter.add_policy(sec="p", ptype="p", rule=("alice", "data1", "write")) + adapter.add_policy(sec="p", ptype="p", rule=("alice", "data1", "read")) + adapter.add_policy(sec="p", ptype="p", rule=("bob", "data2", "read")) + adapter.add_policy( + sec="p", ptype="p", rule=("data_group_admin", "data_group", "write") + ) + adapter.add_policy(sec="g", ptype="g", rule=("alice", "data_group_admin")) + adapter.add_policy(sec="g", ptype="g2", rule=("data2", "data_group")) + e.load_policy() - - self.assertTrue(e.enforce('alice', 'data1', 'write')) - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertTrue(e.enforce('bob', 'data2', 'read')) - self.assertTrue(e.enforce('alice', 'data2', 'write')) + + self.assertTrue(e.enforce("alice", "data1", "write")) + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertTrue(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) # test remove_policy doesn't remove when given an incomplete policy - result = adapter.remove_policy(sec='p', ptype='p', rule=('alice', 'data1')) + result = adapter.remove_policy(sec="p", ptype="p", rule=("alice", "data1")) e.load_policy() - self.assertTrue(e.enforce('alice', 'data1', 'write')) - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertTrue(e.enforce('bob', 'data2', 'read')) - self.assertTrue(e.enforce('alice', 'data2', 'write')) + self.assertTrue(e.enforce("alice", "data1", "write")) + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertTrue(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) self.assertFalse(result) def test_save_policy(self): - ''' + """ test save_policy - ''' + """ 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_remove_filtered_policy(self): - ''' + """ test remove_filtered_policy - ''' + """ e = get_enforcer() adapter = e.get_adapter() - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('alice', 'data1', 'write')) - self.assertFalse(e.enforce('bob', 'data2', 'read')) - self.assertTrue(e.enforce('bob', 'data2', 'write')) - self.assertTrue(e.enforce('alice', 'data2', 'read')) - self.assertTrue(e.enforce('alice', 'data2', 'write')) - - result = adapter.remove_filtered_policy('g', 'g', 6, 'alice', 'data2_admin') + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("alice", "data1", "write")) + self.assertFalse(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertTrue(e.enforce("alice", "data2", "read")) + self.assertTrue(e.enforce("alice", "data2", "write")) + + result = adapter.remove_filtered_policy("g", "g", 6, "alice", "data2_admin") e.load_policy() self.assertFalse(result) - result = adapter.remove_filtered_policy('g', 'g', 0, *[f'v{i}' for i in range(7)]) + result = adapter.remove_filtered_policy( + "g", "g", 0, *[f"v{i}" for i in range(7)] + ) e.load_policy() self.assertFalse(result) - - result = adapter.remove_filtered_policy('g', 'g', 0, 'alice', 'data2_admin') + result = adapter.remove_filtered_policy("g", "g", 0, "alice", "data2_admin") e.load_policy() self.assertTrue(result) - self.assertTrue(e.enforce('alice', 'data1', 'read')) - self.assertFalse(e.enforce('alice', 'data1', 'write')) - self.assertFalse(e.enforce('bob', 'data2', 'read')) - self.assertTrue(e.enforce('bob', 'data2', 'write')) - self.assertFalse(e.enforce('alice', 'data2', 'read')) - self.assertFalse(e.enforce('alice', 'data2', 'write')) - + self.assertTrue(e.enforce("alice", "data1", "read")) + self.assertFalse(e.enforce("alice", "data1", "write")) + self.assertFalse(e.enforce("bob", "data2", "read")) + self.assertTrue(e.enforce("bob", "data2", "write")) + self.assertFalse(e.enforce("alice", "data2", "read")) + self.assertFalse(e.enforce("alice", "data2", "write")) def test_str(self): - ''' + """ test __str__ function - ''' - rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read') - self.assertEqual(rule.__str__(), 'p, alice, data1, read') + """ + rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read") + self.assertEqual(rule.__str__(), "p, alice, data1, read") def test_dict(self): - ''' + """ test __str__ function - ''' - rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read') - self.assertEqual(rule.dict(), {"ptype": 'p', "v0": 'alice', "v1": 'data1', "v2": 'read'}) + """ + rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read") + self.assertEqual( + rule.dict(), {"ptype": "p", "v0": "alice", "v1": "data1", "v2": "read"} + ) def test_repr(self): - ''' + """ test __repr__ function - ''' - adapter = Adapter('mongodb://localhost:27017', 'casbin_test') - rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read') + """ + adapter = Adapter("mongodb://localhost:27017", "casbin_test") + rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read") self.assertEqual(repr(rule), '') # adapter.save_policy(rule) - # self.assertRegex(repr(rule), r'') \ No newline at end of file + # self.assertRegex(repr(rule), r'')