2183610547

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/casbin/pycasbin/commit/<a class=hub.com/casbin/pycasbin/commit/<a class="double-link" href="https://git"><a class=hub.com/casbin/pycasbin/commit/<a class="double-link" href="https://git"><a class=hub.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90437f17ee">c0ef96884<a href="https://github.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90437f17ee">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90437f17ee&quot;&gt;c0ef96884&lt;/a&gt;&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90437f17ee&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90&lt;/a&gt;437f17ee">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/<a class="double-link" href="https://github.com/casbin/pycasbin/commit/db0183596ff7bff67a1737647441d1614cd17c64">db0183596</a><a href="https://github.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90437f17ee">&amp;quot;&amp;gt;db0183596&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;db0183596&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/c0ef96884e0c5fdce05d4ead2010fc90437f1&lt;/a&gt;7ee&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/<a class="double-link" href="https://github.com/casbin/pycasbin/commit/f75702401">f75702401">f75702401</a>

Pull Request Pull Request #249: feat: update_filtered_policies

Run Details

20 of 25 new or added lines in 3 files covered. (80.0%)

1864 of 2183 relevant lines covered (85.39%)

10.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.38

/casbin/internal_enforcer.py

# Copyright 2021 The casbin Authors. All Rights Reserved.
#
# 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 casbin.core_enforcer import CoreEnforcer
from casbin.model.policy_op import PolicyOp


class InternalEnforcer(CoreEnforcer):
    """
    InternalEnforcer = CoreEnforcer + Internal API.
    """

    def _add_policy(self, sec, ptype, rule):
        """adds a rule to the current policy."""
        rule_added = self.model.add_policy(sec, ptype, rule)
        if not rule_added:
            return rule_added

        if self.adapter and self.auto_save:
            if self.adapter.add_policy(sec, ptype, rule) is False:
                return False

            if self.watcher:
                self.watcher.update()

        return rule_added

    def _add_policies(self, sec, ptype, rules):
        """adds rules to the current policy."""
        rules_added = self.model.add_policies(sec, ptype, rules)
        if not rules_added:
            return rules_added

        if self.adapter and self.auto_save:
            if hasattr(self.adapter, "add_policies") is False:
                return False

            if self.adapter.add_policies(sec, ptype, rules) is False:
                return False

            if self.watcher:
                self.watcher.update()

        return rules_added

    def _update_policy(self, sec, ptype, old_rule, new_rule):
        """updates a rule from the current policy."""
        rule_updated = self.model.update_policy(sec, ptype, old_rule, new_rule)

        if not rule_updated:
            return rule_updated

        if self.adapter and self.auto_save:

            if self.adapter.update_policy(sec, ptype, old_rule, new_rule) is False:
                return False

            if self.watcher:
                self.watcher.update()

        return rule_updated

    def _update_policies(self, sec, ptype, old_rules, new_rules):
        """updates rules from the current policy."""
        rules_updated = self.model.update_policies(sec, ptype, old_rules, new_rules)

        if not rules_updated:
            return rules_updated

        if self.adapter and self.auto_save:

            if self.adapter.update_policies(sec, ptype, old_rules, new_rules) is False:
                return False

            if self.watcher:
                self.watcher.update()

        return rules_updated

    def _update_filtered_policies(
        self, sec, ptype, new_rules, field_index, *field_values
    ):
        """_update_filtered_policies deletes old rules and adds new rules."""

        old_rules = self.model.get_filtered_policy(
            sec, ptype, field_index, *field_values
        )

        if self.adapter and self.auto_save:
            try:
                old_rules = self.adapter.update_filtered_policies(
                    sec, ptype, new_rules, field_index, *field_values
                )
            except:
                pass

        if not old_rules:
            return False

        is_rule_changed = self.model.remove_policies(sec, ptype, old_rules)
        self.model.add_policies(sec, ptype, new_rules)
        is_rule_changed = is_rule_changed and len(new_rules) != 0
        if not is_rule_changed:
            return is_rule_changed
        if sec == "g":
            self.build_role_links()
        if self.watcher:
            self.watcher.update()
        return is_rule_changed

    def _remove_policy(self, sec, ptype, rule):
        """removes a rule from the current policy."""
        rule_removed = self.model.remove_policy(sec, ptype, rule)
        if not rule_removed:
            return rule_removed

        if self.adapter and self.auto_save:
            if self.adapter.remove_policy(sec, ptype, rule) is False:
                return False

            if self.watcher:
                self.watcher.update()

        return rule_removed

    def _remove_policies(self, sec, ptype, rules):
        """RemovePolicies removes policy rules from the model."""
        rules_removed = self.model.remove_policies(sec, ptype, rules)
        if not rules_removed:
            return rules_removed

        if self.adapter and self.auto_save:
            if hasattr(self.adapter, "remove_policies") is False:
                return False

            if self.adapter.remove_policies(sec, ptype, rules) is False:
                return False

            if self.watcher:
                self.watcher.update()

        return rules_removed

    def _remove_filtered_policy(self, sec, ptype, field_index, *field_values):
        """removes rules based on field filters from the current policy."""
        rule_removed = self.model.remove_filtered_policy(
            sec, ptype, field_index, *field_values
        )
        if not rule_removed:
            return rule_removed

        if self.adapter and self.auto_save:
            if (
                self.adapter.remove_filtered_policy(
                    sec, ptype, field_index, *field_values
                )
                is False
            ):
                return False

            if self.watcher:
                self.watcher.update()

        return rule_removed

    def _remove_filtered_policy_returns_effects(
        self, sec, ptype, field_index, *field_values
    ):
        """removes rules based on field filters from the current policy."""
        rule_removed = self.model.remove_filtered_policy_returns_effects(
            sec, ptype, field_index, *field_values
        )
        if len(rule_removed) == 0:
            return rule_removed

        if self.adapter and self.auto_save:
            if (
                self.adapter.remove_filtered_policy(
                    sec, ptype, field_index, *field_values
                )
                is False
            ):
                return False

            if self.watcher:
                self.watcher.update()

        return rule_removed

1	# Copyright 2021 The casbin Authors. All Rights Reserved.
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# http://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	from casbin.core_enforcer import CoreEnforcer	12✔
16	from casbin.model.policy_op import PolicyOp	12✔
17
18
19	class InternalEnforcer(CoreEnforcer):	12✔
20	"""
21	InternalEnforcer = CoreEnforcer + Internal API.
22	"""
23
24	def _add_policy(self, sec, ptype, rule):	12✔
25	"""adds a rule to the current policy."""
26	rule_added = self.model.add_policy(sec, ptype, rule)	12✔
27	if not rule_added:	12✔
28	return rule_added	×
29
30	if self.adapter and self.auto_save:	12✔
31	if self.adapter.add_policy(sec, ptype, rule) is False:	12✔
32	return False	×
33
34	if self.watcher:	12✔
35	self.watcher.update()	×
36
37	return rule_added	12✔
38
39	def _add_policies(self, sec, ptype, rules):	12✔
40	"""adds rules to the current policy."""
41	rules_added = self.model.add_policies(sec, ptype, rules)	12✔
42	if not rules_added:	12✔
43	return rules_added	×
44
45	if self.adapter and self.auto_save:	12✔
46	if hasattr(self.adapter, "add_policies") is False:	12✔
47	return False	×
48
49	if self.adapter.add_policies(sec, ptype, rules) is False:	12✔
50	return False	×
51
52	if self.watcher:	12✔
53	self.watcher.update()	×
54
55	return rules_added	12✔
56
57	def _update_policy(self, sec, ptype, old_rule, new_rule):	12✔
58	"""updates a rule from the current policy."""
59	rule_updated = self.model.update_policy(sec, ptype, old_rule, new_rule)	×
60
61	if not rule_updated:	×
62	return rule_updated	×
63
64	if self.adapter and self.auto_save:	×
65
66	if self.adapter.update_policy(sec, ptype, old_rule, new_rule) is False:	×
67	return False	×
68
69	if self.watcher:	×
70	self.watcher.update()	×
71
72	return rule_updated	×
73
74	def _update_policies(self, sec, ptype, old_rules, new_rules):	12✔
75	"""updates rules from the current policy."""
76	rules_updated = self.model.update_policies(sec, ptype, old_rules, new_rules)	×
77
78	if not rules_updated:	×
79	return rules_updated	×
80
81	if self.adapter and self.auto_save:	×
82
83	if self.adapter.update_policies(sec, ptype, old_rules, new_rules) is False:	×
84	return False	×
85
86	if self.watcher:	×
87	self.watcher.update()	×
88
89	return rules_updated	×
90
91	def _update_filtered_policies(	12✔
92	self, sec, ptype, new_rules, field_index, *field_values
93	):
94	"""_update_filtered_policies deletes old rules and adds new rules."""
95
96	old_rules = self.model.get_filtered_policy(	12✔
97	sec, ptype, field_index, *field_values
98	)
99
100	if self.adapter and self.auto_save:	12✔
101	try:	12✔
102	old_rules = self.adapter.update_filtered_policies(	12✔
103	sec, ptype, new_rules, field_index, *field_values
104	)
105	except:	12✔
106	pass	12✔
107
108	if not old_rules:	12✔
NEW 109	return False	×
110
111	is_rule_changed = self.model.remove_policies(sec, ptype, old_rules)	12✔
112	self.model.add_policies(sec, ptype, new_rules)	12✔
113	is_rule_changed = is_rule_changed and len(new_rules) != 0	12✔
114	if not is_rule_changed:	12✔
NEW 115	return is_rule_changed	×
116	if sec == "g":	12✔
NEW 117	self.build_role_links()	×
118	if self.watcher:	12✔
NEW 119	self.watcher.update()	×
120	return is_rule_changed	12✔
121
122	def _remove_policy(self, sec, ptype, rule):	12✔
123	"""removes a rule from the current policy."""
124	rule_removed = self.model.remove_policy(sec, ptype, rule)	12✔
125	if not rule_removed:	12✔
126	return rule_removed	×
127
128	if self.adapter and self.auto_save:	12✔
129	if self.adapter.remove_policy(sec, ptype, rule) is False:	12✔
130	return False	×
131
132	if self.watcher:	12✔
133	self.watcher.update()	×
134
135	return rule_removed	12✔
136
137	def _remove_policies(self, sec, ptype, rules):	12✔
138	"""RemovePolicies removes policy rules from the model."""
139	rules_removed = self.model.remove_policies(sec, ptype, rules)	12✔
140	if not rules_removed:	12✔
141	return rules_removed	×
142
143	if self.adapter and self.auto_save:	12✔
144	if hasattr(self.adapter, "remove_policies") is False:	12✔
145	return False	×
146
147	if self.adapter.remove_policies(sec, ptype, rules) is False:	12✔
148	return False	×
149
150	if self.watcher:	12✔
151	self.watcher.update()	×
152
153	return rules_removed	12✔
154
155	def _remove_filtered_policy(self, sec, ptype, field_index, *field_values):	12✔
156	"""removes rules based on field filters from the current policy."""
157	rule_removed = self.model.remove_filtered_policy(	12✔
158	sec, ptype, field_index, *field_values
159	)
160	if not rule_removed:	12✔
161	return rule_removed	×
162
163	if self.adapter and self.auto_save:	12✔
164	if (	12✔
165	self.adapter.remove_filtered_policy(
166	sec, ptype, field_index, *field_values
167	)
168	is False
169	):
170	return False	×
171
172	if self.watcher:	12✔
173	self.watcher.update()	×
174
175	return rule_removed	12✔
176
177	def _remove_filtered_policy_returns_effects(	12✔
178	self, sec, ptype, field_index, *field_values
179	):
180	"""removes rules based on field filters from the current policy."""
181	rule_removed = self.model.remove_filtered_policy_returns_effects(	12✔
182	sec, ptype, field_index, *field_values
183	)
184	if len(rule_removed) == 0:	12✔
185	return rule_removed	×
186
187	if self.adapter and self.auto_save:	12✔
188	if (	12✔
189	self.adapter.remove_filtered_policy(
190	sec, ptype, field_index, *field_values
191	)
192	is False
193	):
194	return False	×
195
196	if self.watcher:	12✔
197	self.watcher.update()	×
198
199	return rule_removed	12✔

casbin / pycasbin / 2183610547

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous