1453315733

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/casbin/pycasbin/commit/<a class=hub.com/casbin/pycasbin/commit/4ce910f021e5a504b798e6d63879be160d9ed296">4ce910f02<a href="https://github.com/casbin/pycasbin/commit/4ce910f021e5a504b798e6d63879be160d9ed296">">Merge </a><a class="double-link" href="https://github.com/casbin/pycasbin/commit/<a class="double-link" href="https://github.com/casbin/pycasbin/commit/44cf35d9fa7ac85ac799faf05775b79685fd6bef">44cf35d9f</a>">44cf35d9f</a><a href="https://github.com/casbin/pycasbin/commit/4ce910f021e5a504b798e6d63879be160d9ed296"> into 786938821">786938821</a>

Pull Request Pull Request #210: fix: replace build_role_links with build_incremental_role_links

Run Details

17 of 24 new or added lines in 2 files covered. (70.83%)

1703 of 2025 relevant lines covered (84.1%)

10.09 hits per line

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

56.32

/casbin/internal_enforcer.py

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 _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	from casbin.core_enforcer import CoreEnforcer	12✔
2	from casbin.model.policy_op import PolicyOp	12✔
3
4
5	class InternalEnforcer(CoreEnforcer):	12✔
6	"""
7	InternalEnforcer = CoreEnforcer + Internal API.
8	"""
9
10	def _add_policy(self, sec, ptype, rule):	12✔
11	"""adds a rule to the current policy."""
12	rule_added = self.model.add_policy(sec, ptype, rule)	12✔
13	if not rule_added:	12✔
14	return rule_added	×
15
16	if self.adapter and self.auto_save:	12✔
17	if self.adapter.add_policy(sec, ptype, rule) is False:	12✔
18	return False	×
19
20	if self.watcher:	12✔
21	self.watcher.update()	×
22
23	return rule_added	12✔
24
25	def _add_policies(self, sec, ptype, rules):	12✔
26	"""adds rules to the current policy."""
27	rules_added = self.model.add_policies(sec, ptype, rules)	12✔
28	if not rules_added:	12✔
29	return rules_added	×
30
31	if self.adapter and self.auto_save:	12✔
32	if hasattr(self.adapter, "add_policies") is False:	12✔
33	return False	×
34
35	if self.adapter.add_policies(sec, ptype, rules) is False:	12✔
36	return False	×
37
38	if self.watcher:	12✔
39	self.watcher.update()	×
40
41	return rules_added	12✔
42
43	def _update_policy(self, sec, ptype, old_rule, new_rule):	12✔
44	"""updates a rule from the current policy."""
45	rule_updated = self.model.update_policy(sec, ptype, old_rule, new_rule)	×
46
47	if not rule_updated:	×
48	return rule_updated	×
49
50	if self.adapter and self.auto_save:	×
51
52	if self.adapter.update_policy(sec, ptype, old_rule, new_rule) is False:	×
53	return False	×
54
55	if self.watcher:	×
56	self.watcher.update()	×
57
58	return rule_updated	×
59
60	def _update_policies(self, sec, ptype, old_rules, new_rules):	12✔
61	"""updates rules from the current policy."""
62	rules_updated = self.model.update_policies(sec, ptype, old_rules, new_rules)	×
63
64	if not rules_updated:	×
65	return rules_updated	×
66
67	if self.adapter and self.auto_save:	×
68
69	if self.adapter.update_policies(sec, ptype, old_rules, new_rules) is False:	×
70	return False	×
71
72	if self.watcher:	×
73	self.watcher.update()	×
74
75	return rules_updated	×
76
77	def _remove_policy(self, sec, ptype, rule):	12✔
78	"""removes a rule from the current policy."""
79	rule_removed = self.model.remove_policy(sec, ptype, rule)	12✔
80	if not rule_removed:	12✔
81	return rule_removed	×
82
83	if self.adapter and self.auto_save:	12✔
84	if self.adapter.remove_policy(sec, ptype, rule) is False:	12✔
85	return False	×
86
87	if self.watcher:	12✔
88	self.watcher.update()	×
89
90	return rule_removed	12✔
91
92	def _remove_policies(self, sec, ptype, rules):	12✔
93	"""RemovePolicies removes policy rules from the model."""
94	rules_removed = self.model.remove_policies(sec, ptype, rules)	12✔
95	if not rules_removed:	12✔
96	return rules_removed	×
97
98	if self.adapter and self.auto_save:	12✔
99	if hasattr(self.adapter, "remove_policies") is False:	12✔
100	return False	×
101
102	if self.adapter.remove_policies(sec, ptype, rules) is False:	12✔
103	return False	×
104
105	if self.watcher:	12✔
106	self.watcher.update()	×
107
108	return rules_removed	12✔
109
110	def _remove_filtered_policy(self, sec, ptype, field_index, *field_values):	12✔
111	"""removes rules based on field filters from the current policy."""
112	rule_removed = self.model.remove_filtered_policy(	12✔
113	sec, ptype, field_index, *field_values
114	)
115	if not rule_removed:	12✔
116	return rule_removed	×
117
118	if self.adapter and self.auto_save:	12✔
119	if (	12✔
120	self.adapter.remove_filtered_policy(
121	sec, ptype, field_index, *field_values
122	)
123	is False
124	):
125	return False	×
126
127	if self.watcher:	12✔
128	self.watcher.update()	×
129
130	return rule_removed	12✔
131
132	def _remove_filtered_policy_returns_effects(self, sec, ptype, field_index, *field_values):	12✔
133	"""removes rules based on field filters from the current policy."""
134	rule_removed = self.model.remove_filtered_policy_returns_effects(	12✔
135	sec, ptype, field_index, *field_values
136	)
137	if len(rule_removed) == 0:	12✔
NEW 138	return rule_removed	×
139
140	if self.adapter and self.auto_save:	12✔
141	if (	12✔
142	self.adapter.remove_filtered_policy(
143	sec, ptype, field_index, *field_values
144	)
145	is False
146	):
NEW 147	return False	×
148
149	if self.watcher:	12✔
NEW 150	self.watcher.update()	×
151
152	return rule_removed	12✔

casbin / pycasbin / 1453315733

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