988363370

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/<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/a5ed9176cf013169160e4e01a3c65d8406f17c54">a5ed9176c">&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/</a><a class="double-link" href="https://github.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/casbin/pycasbin/commit/a5ed9176cf013169160e4e01a3c65d8406f17c54">a5ed9176c</a><a href="https://github.com/casbin/pycasbin/commit/a5ed9176cf013169160e4e01a3c65d8406f17c54">&lt;a href=&quot;https://github.com/casbin/pycasbin/commit/a5ed9176cf013169160e4e01a3c65d8406f17c54&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/casbin/pycasbin/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/casbin/pycasbin/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/casbin/pycasbin/commit/a5ed9176cf013169160e4e01a3c65d8406f17c54&amp;quot;&amp;gt;a5ed9176c&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/casbin/pycasbin/commit/a5ed9176cf013169160e4e01a3c65d8406f17c54&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/casbin/pycasbin/commit/a5ed9176cf013169160e4e01a3c65d84&amp;lt;/a&amp;gt;06f</a>17c54&quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/casbin/pycasbin/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/casbin/pycasbin/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/casbin/pycasbin/commit/913870d39&quot;&gt;913870d39&lt;/a&gt;&lt;a href=&quot;htt... (continued)

Pull Request Pull Request #174: refactor: modify model structure

Run Details

62 of 63 new or added lines in 3 files covered. (98.41%)

1586 of 1890 relevant lines covered (83.92%)

10.07 hits per line

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

85.26

/casbin/model/policy.py

import logging


class Policy:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.model = {}

    def __getitem__(self, item):
        return self.model.get(item)

    def __setitem__(self, key, value):
        self.model[key] = value

    def keys(self):
        return self.model.keys()

    def values(self):
        return self.model.values()

    def items(self):
        return self.model.items()

    def build_role_links(self, rm_map):
        """initializes the roles in RBAC."""

        if "g" not in self.keys():
            return

        for ptype, ast in self["g"].items():
            rm = rm_map[ptype]
            ast.build_role_links(rm)

    def build_incremental_role_links(self, rm, op, sec, ptype, rules):
        if sec == "g":
            self[sec].get(ptype).build_incremental_role_links(rm, op, rules)

    def print_policy(self):
        """Log using info"""

        self.logger.info("Policy:")
        for sec in ["p", "g"]:
            if sec not in self.keys():
                continue

            for key, ast in self[sec].items():
                self.logger.info("{} : {} : {}".format(key, ast.value, ast.policy))

    def clear_policy(self):
        """clears all current policy."""

        for sec in ["p", "g"]:
            if sec not in self.keys():
                continue

            for key in self[sec].keys():
                self[sec][key].policy = []

    def get_policy(self, sec, ptype):
        """gets all rules in a policy."""

        return self[sec][ptype].policy

    def get_filtered_policy(self, sec, ptype, field_index, *field_values):
        """gets rules based on field filters from a policy."""
        return [
            rule
            for rule in self[sec][ptype].policy
            if all(
                value == "" or rule[field_index + i] == value
                for i, value in enumerate(field_values)
            )
        ]

    def has_policy(self, sec, ptype, rule):
        """determines whether a model has the specified policy rule."""
        if sec not in self.keys():
            return False
        if ptype not in self[sec]:
            return False

        return rule in self[sec][ptype].policy

    def add_policy(self, sec, ptype, rule):
        """adds a policy rule to the model."""

        if not self.has_policy(sec, ptype, rule):
            self[sec][ptype].policy.append(rule)
            return True

        return False

    def add_policies(self, sec, ptype, rules):
        """adds policy rules to the model."""

        for rule in rules:
            if self.has_policy(sec, ptype, rule):
                return False

        for rule in rules:
            self[sec][ptype].policy.append(rule)

        return True

    def update_policy(self, sec, ptype, old_rule, new_rule):
        """update a policy rule from the model."""

        if sec not in self.keys():
            return False
        if ptype not in self[sec]:
            return False

        ast = self[sec][ptype]

        if old_rule in ast.policy:
            rule_index = ast.policy.index(old_rule)
        else:
            return False

        if "p_priority" in ast.tokens:
            priority_index = ast.tokens.index("p_priority")
            if old_rule[priority_index] == new_rule[priority_index]:
                ast.policy[rule_index] = new_rule
            else:
                raise Exception("New rule should have the same priority with old rule.")
        else:
            ast.policy[rule_index] = new_rule

        return True

    def update_policies(self, sec, ptype, old_rules, new_rules):
        """update policy rules from the model."""

        if sec not in self.keys():
            return False
        if ptype not in self[sec]:
            return False
        if len(old_rules) != len(new_rules):
            return False

        ast = self[sec][ptype]
        old_rules_index = []

        for old_rule in old_rules:
            if old_rule in ast.policy:
                old_rules_index.append(ast.policy.index(old_rule))
            else:
                return False

        if "p_priority" in ast.tokens:
            priority_index = ast.tokens.index("p_priority")
            for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):
                if old_rule[priority_index] == new_rule[priority_index]:
                    ast.policy[idx] = new_rule
                else:
                    raise Exception(
                        "New rule should have the same priority with old rule."
                    )
        else:
            for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):
                ast.policy[idx] = new_rule

        return True

    def remove_policy(self, sec, ptype, rule):
        """removes a policy rule from the model."""
        if not self.has_policy(sec, ptype, rule):
            return False

        self[sec][ptype].policy.remove(rule)

        return rule not in self[sec][ptype].policy

    def remove_policies(self, sec, ptype, rules):
        """RemovePolicies removes policy rules from the model."""

        for rule in rules:
            if not self.has_policy(sec, ptype, rule):
                return False
            self[sec][ptype].policy.remove(rule)
            if rule in self[sec][ptype].policy:
                return False

        return True

    def remove_policies_with_effected(self, sec, ptype, rules):
        effected = []
        for rule in rules:
            if self.has_policy(sec, ptype, rule):
                effected.append(rule)
                self.remove_policy(sec, ptype, rule)

        return effected

    def remove_filtered_policy_returns_effects(
        self, sec, ptype, field_index, *field_values
    ):
        """
        remove_filtered_policy_returns_effects removes policy rules based on field filters from the model.
        """
        tmp = []
        effects = []

        if len(field_values) == 0:
            return []
        if sec not in self.keys():
            return []
        if ptype not in self[sec]:
            return []

        for rule in self[sec][ptype].policy:
            if all(
                value == "" or rule[field_index + i] == value
                for i, value in enumerate(field_values[0])
            ):
                effects.append(rule)
            else:
                tmp.append(rule)

        self[sec][ptype].policy = tmp

        return effects

    def remove_filtered_policy(self, sec, ptype, field_index, *field_values):
        """removes policy rules based on field filters from the model."""
        tmp = []
        res = False

        if sec not in self.keys():
            return res
        if ptype not in self[sec]:
            return res

        for rule in self[sec][ptype].policy:
            if all(
                value == "" or rule[field_index + i] == value
                for i, value in enumerate(field_values)
            ):
                res = True
            else:
                tmp.append(rule)

        self[sec][ptype].policy = tmp

        return res

    def get_values_for_field_in_policy(self, sec, ptype, field_index):
        """gets all values for a field for all rules in a policy, duplicated values are removed."""
        values = []
        if sec not in self.keys():
            return values
        if ptype not in self[sec]:
            return values

        for rule in self[sec][ptype].policy:
            value = rule[field_index]
            if value not in values:
                values.append(value)

        return values

1	import logging	12✔
2
3
4	class Policy:	12✔
5	def __init__(self):	12✔
6	self.logger = logging.getLogger(__name__)	12✔
7	self.model = {}	12✔
8
9	def __getitem__(self, item):	12✔
10	return self.model.get(item)	12✔
11
12	def __setitem__(self, key, value):	12✔
13	self.model[key] = value	12✔
14
15	def keys(self):	12✔
16	return self.model.keys()	12✔
17
18	def values(self):	12✔
NEW 19	return self.model.values()	×
20
21	def items(self):	12✔
22	return self.model.items()	12✔
23
24	def build_role_links(self, rm_map):	12✔
25	"""initializes the roles in RBAC."""
26
27	if "g" not in self.keys():	12✔
28	return	12✔
29
30	for ptype, ast in self["g"].items():	12✔
31	rm = rm_map[ptype]	12✔
32	ast.build_role_links(rm)	12✔
33
34	def build_incremental_role_links(self, rm, op, sec, ptype, rules):	12✔
35	if sec == "g":	12✔
36	self[sec].get(ptype).build_incremental_role_links(rm, op, rules)	12✔
37
38	def print_policy(self):	12✔
39	"""Log using info"""
40
41	self.logger.info("Policy:")	12✔
42	for sec in ["p", "g"]:	12✔
43	if sec not in self.keys():	12✔
44	continue	12✔
45
46	for key, ast in self[sec].items():	12✔
47	self.logger.info("{} : {} : {}".format(key, ast.value, ast.policy))	12✔
48
49	def clear_policy(self):	12✔
50	"""clears all current policy."""
51
52	for sec in ["p", "g"]:	12✔
53	if sec not in self.keys():	12✔
54	continue	12✔
55
56	for key in self[sec].keys():	12✔
57	self[sec][key].policy = []	12✔
58
59	def get_policy(self, sec, ptype):	12✔
60	"""gets all rules in a policy."""
61
62	return self[sec][ptype].policy	12✔
63
64	def get_filtered_policy(self, sec, ptype, field_index, *field_values):	12✔
65	"""gets rules based on field filters from a policy."""
66	return [	12✔
67	rule
68	for rule in self[sec][ptype].policy
69	if all(
70	value == "" or rule[field_index + i] == value
71	for i, value in enumerate(field_values)
72	)
73	]
74
75	def has_policy(self, sec, ptype, rule):	12✔
76	"""determines whether a model has the specified policy rule."""
77	if sec not in self.keys():	12✔
78	return False	×
79	if ptype not in self[sec]:	12✔
80	return False	×
81
82	return rule in self[sec][ptype].policy	12✔
83
84	def add_policy(self, sec, ptype, rule):	12✔
85	"""adds a policy rule to the model."""
86
87	if not self.has_policy(sec, ptype, rule):	12✔
88	self[sec][ptype].policy.append(rule)	12✔
89	return True	12✔
90
91	return False	×
92
93	def add_policies(self, sec, ptype, rules):	12✔
94	"""adds policy rules to the model."""
95
96	for rule in rules:	12✔
97	if self.has_policy(sec, ptype, rule):	12✔
98	return False	×
99
100	for rule in rules:	12✔
101	self[sec][ptype].policy.append(rule)	12✔
102
103	return True	12✔
104
105	def update_policy(self, sec, ptype, old_rule, new_rule):	12✔
106	"""update a policy rule from the model."""
107
108	if sec not in self.keys():	12✔
109	return False	×
110	if ptype not in self[sec]:	12✔
111	return False	×
112
113	ast = self[sec][ptype]	12✔
114
115	if old_rule in ast.policy:	12✔
116	rule_index = ast.policy.index(old_rule)	12✔
117	else:
118	return False	×
119
120	if "p_priority" in ast.tokens:	12✔
121	priority_index = ast.tokens.index("p_priority")	12✔
122	if old_rule[priority_index] == new_rule[priority_index]:	12✔
123	ast.policy[rule_index] = new_rule	12✔
124	else:
125	raise Exception("New rule should have the same priority with old rule.")	×
126	else:
127	ast.policy[rule_index] = new_rule	12✔
128
129	return True	12✔
130
131	def update_policies(self, sec, ptype, old_rules, new_rules):	12✔
132	"""update policy rules from the model."""
133
134	if sec not in self.keys():	12✔
135	return False	×
136	if ptype not in self[sec]:	12✔
137	return False	×
138	if len(old_rules) != len(new_rules):	12✔
139	return False	×
140
141	ast = self[sec][ptype]	12✔
142	old_rules_index = []	12✔
143
144	for old_rule in old_rules:	12✔
145	if old_rule in ast.policy:	12✔
146	old_rules_index.append(ast.policy.index(old_rule))	12✔
147	else:
148	return False	×
149
150	if "p_priority" in ast.tokens:	12✔
151	priority_index = ast.tokens.index("p_priority")	12✔
152	for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):	12✔
153	if old_rule[priority_index] == new_rule[priority_index]:	12✔
154	ast.policy[idx] = new_rule	12✔
155	else:
156	raise Exception(	×
157	"New rule should have the same priority with old rule."
158	)
159	else:
160	for idx, old_rule, new_rule in zip(old_rules_index, old_rules, new_rules):	12✔
161	ast.policy[idx] = new_rule	12✔
162
163	return True	12✔
164
165	def remove_policy(self, sec, ptype, rule):	12✔
166	"""removes a policy rule from the model."""
167	if not self.has_policy(sec, ptype, rule):	12✔
168	return False	12✔
169
170	self[sec][ptype].policy.remove(rule)	12✔
171
172	return rule not in self[sec][ptype].policy	12✔
173
174	def remove_policies(self, sec, ptype, rules):	12✔
175	"""RemovePolicies removes policy rules from the model."""
176
177	for rule in rules:	12✔
178	if not self.has_policy(sec, ptype, rule):	12✔
179	return False	×
180	self[sec][ptype].policy.remove(rule)	12✔
181	if rule in self[sec][ptype].policy:	12✔
182	return False	×
183
184	return True	12✔
185
186	def remove_policies_with_effected(self, sec, ptype, rules):	12✔
187	effected = []	12✔
188	for rule in rules:	12✔
189	if self.has_policy(sec, ptype, rule):	12✔
190	effected.append(rule)	12✔
191	self.remove_policy(sec, ptype, rule)	12✔
192
193	return effected	12✔
194
195	def remove_filtered_policy_returns_effects(	12✔
196	self, sec, ptype, field_index, *field_values
197	):
198	"""
199	remove_filtered_policy_returns_effects removes policy rules based on field filters from the model.
200	"""
201	tmp = []	12✔
202	effects = []	12✔
203
204	if len(field_values) == 0:	12✔
205	return []	×
206	if sec not in self.keys():	12✔
207	return []	×
208	if ptype not in self[sec]:	12✔
209	return []	×
210
211	for rule in self[sec][ptype].policy:	12✔
212	if all(	12✔
213	value == "" or rule[field_index + i] == value
214	for i, value in enumerate(field_values[0])
215	):
216	effects.append(rule)	12✔
217	else:
218	tmp.append(rule)	12✔
219
220	self[sec][ptype].policy = tmp	12✔
221
222	return effects	12✔
223
224	def remove_filtered_policy(self, sec, ptype, field_index, *field_values):	12✔
225	"""removes policy rules based on field filters from the model."""
226	tmp = []	12✔
227	res = False	12✔
228
229	if sec not in self.keys():	12✔
230	return res	×
231	if ptype not in self[sec]:	12✔
232	return res	×
233
234	for rule in self[sec][ptype].policy:	12✔
235	if all(	12✔
236	value == "" or rule[field_index + i] == value
237	for i, value in enumerate(field_values)
238	):
239	res = True	12✔
240	else:
241	tmp.append(rule)	12✔
242
243	self[sec][ptype].policy = tmp	12✔
244
245	return res	12✔
246
247	def get_values_for_field_in_policy(self, sec, ptype, field_index):	12✔
248	"""gets all values for a field for all rules in a policy, duplicated values are removed."""
249	values = []	12✔
250	if sec not in self.keys():	12✔
251	return values	×
252	if ptype not in self[sec]:	12✔
253	return values	×
254
255	for rule in self[sec][ptype].policy:	12✔
256	value = rule[field_index]	12✔
257	if value not in values:	12✔
258	values.append(value)	12✔
259
260	return values	12✔

casbin / pycasbin / 988363370

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