3060734495

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/pycasbin/sqlalchemy-adapter/commit/<a class=hub.com/pycasbin/sqlalchemy-adapter/commit/<a class="double-link" href="https://git"><a class=hub.com/pycasbin/sqlalchemy-adapter/commit/<a class="double-link" href="https://git"><a class=hub.com/pycasbin/sqlalchemy-adapter/commit/<a class="double-link" href="https://git"><a class=hub.com/pycasbin/sqlalchemy-adapter/commit/830758a37606bcb07aeb27e3184549d22602d830">830758a37">&lt;a href=&quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/</a><a class="double-link" href="https://github.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/830758a37606bcb07aeb27e3184549d22602d830">830758a37</a><a href="https://github.com/pycasbin/sqlalchemy-adapter/commit/830758a37606bcb07aeb27e3184549d22602d830">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/pycasbin/sqlalchemy-adapter/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/pycasbin/sqlalchemy-adapter/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/pycasbin/sqlalchemy-adapter/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/pycasbin/sqlalchemy-adapter/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/pycasbin/sqlalchemy-adapter/commit/830758a37606bcb07aeb27e3184549d22602d830&quot;&gt;830758a37&lt;/a&gt;&lt;a href=&quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/830758a37606bcb07aeb27e3184549d22602d830&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/pycasbin/sqlal... (continued)

Pull Request Pull Request #58: refactor: add ability to not use internally constructed session

Run Details

93 of 104 new or added lines in 2 files covered. (89.42%)

169 of 184 relevant lines covered (91.85%)

3.67 hits per line

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

91.8

/casbin_sqlalchemy_adapter/adapter.py

from casbin import persist
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()


class CasbinRule(Base):
    __tablename__ = "casbin_rule"

    id = Column(Integer, primary_key=True)
    ptype = Column(String(255))
    v0 = Column(String(255))
    v1 = Column(String(255))
    v2 = Column(String(255))
    v3 = Column(String(255))
    v4 = Column(String(255))
    v5 = Column(String(255))

    def __str__(self):
        arr = [self.ptype]
        for v in (self.v0, self.v1, self.v2, self.v3, self.v4, self.v5):
            if v is None:
                break
            arr.append(v)
        return ", ".join(arr)

    def __repr__(self):
        return '<CasbinRule {}: "{}">'.format(self.id, str(self))


class Filter:
    ptype = []
    v0 = []
    v1 = []
    v2 = []
    v3 = []
    v4 = []
    v5 = []


class CoreAdapter(persist.Adapter, persist.adapters.UpdateAdapter):
    """the interface for Casbin adapters."""

    def __init__(self, engine, db_class=None, filtered=False):
        if isinstance(engine, str):
            self._engine = create_engine(engine)
        else:
            self._engine = engine

        if db_class is None:
            db_class = CasbinRule
        else:
            for attr in (
                "id",
                "ptype",
                "v0",
                "v1",
                "v2",
                "v3",
                "v4",
                "v5",
            ):  # id attr was used by filter
                if not hasattr(db_class, attr):
                    raise Exception(f"{attr} not found in custom DatabaseClass.")
            Base.metadata = db_class.metadata

        self._db_class = db_class
        Base.metadata.create_all(self._engine)
        self._filtered = filtered
        self.session = None

    def set_session(self, session):
        self.session = session

    def _commit(self):
        self.session.commit()

    def _close(self):
        self.session.close()

    def load_policy(self, model):
        """loads all policy rules from the storage."""
        lines = self.session.query(self._db_class).all()
        for line in lines:
            persist.load_policy_line(str(line), model)
        self._close()

    def is_filtered(self):
        return self._filtered

    def load_filtered_policy(self, model, filter) -> None:
        """loads all policy rules from the storage."""
        query = self.session.query(self._db_class)
        filters = self.filter_query(query, filter)
        filters = filters.all()

        for line in filters:
            persist.load_policy_line(str(line), model)
        self._filtered = True
        self._close()

    def filter_query(self, querydb, filter):
        for attr in ("ptype", "v0", "v1", "v2", "v3", "v4", "v5"):
            if len(getattr(filter, attr)) > 0:
                querydb = querydb.filter(
                    getattr(self._db_class, attr).in_(getattr(filter, attr))
                )
        return querydb.order_by(self._db_class.id)

    def _save_policy_line(self, ptype, rule):
        line = self._db_class(ptype=ptype)
        for i, v in enumerate(rule):
            setattr(line, "v{}".format(i), v)
        self.session.add(line)
        # commit in save_policy

    def save_policy(self, model):
        """saves all policy rules to the storage."""
        query = self.session.query(self._db_class)
        query.delete()
        for sec in ["p", "g"]:
            if sec not in model.model.keys():
                continue
            for ptype, ast in model.model[sec].items():
                for rule in ast.policy:
                    self._save_policy_line(ptype, rule)

        self._commit()

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

    def add_policies(self, sec, ptype, rules):
        """adds a policy rules to the storage."""
        for rule in rules:
            self._save_policy_line(ptype, rule)
        self._commit()

    def remove_policy(self, sec, ptype, rule):
        """removes a policy rule from the storage."""
        query = self.session.query(self._db_class)
        query = query.filter(self._db_class.ptype == ptype)
        for i, v in enumerate(rule):
            query = query.filter(getattr(self._db_class, "v{}".format(i)) == v)
        query.delete()

        self._commit()

    def remove_policies(self, sec, ptype, rules):
        """remove policy rules from the storage."""
        if not rules:
            return
        query = self.session.query(self._db_class)
        query = query.filter(self._db_class.ptype == ptype)
        rules = zip(*rules)
        for i, rule in enumerate(rules):
            query = query.filter(
                or_(getattr(self._db_class, "v{}".format(i)) == v for v in rule)
            )
        query.delete()
        self._commit()

    def remove_filtered_policy(self, sec, ptype, field_index, *field_values):
        """removes policy rules that match the filter from the storage.
        This is part of the Auto-Save feature.
        """
        query = self.session.query(self._db_class).filter(self._db_class.ptype == ptype)

        if not (0 <= field_index <= 5):
            return False
        if not (1 <= field_index + len(field_values) <= 6):
            return False
        for i, v in enumerate(field_values):
            if v != "":
                v_value = getattr(self._db_class, "v{}".format(field_index + i))
                query = query.filter(v_value == v)
        query.delete()
        self._commit()

    def update_policy(
        self, sec: str, ptype: str, old_rule: [str], new_rule: [str]
    ) -> None:
        """
        Update the old_rule with the new_rule in the database (storage).

        :param sec: section type
        :param ptype: policy type
        :param old_rule: the old rule that needs to be modified
        :param new_rule: the new rule to replace the old rule

        :return: None
        """

        query = self.session.query(self._db_class).filter(self._db_class.ptype == ptype)

        # locate the old rule
        for index, value in enumerate(old_rule):
            v_value = getattr(self._db_class, "v{}".format(index))
            query = query.filter(v_value == value)

        # need the length of the longest_rule to perform overwrite
        longest_rule = old_rule if len(old_rule) > len(new_rule) else new_rule
        old_rule_line = query.one()

        # overwrite the old rule with the new rule
        for index in range(len(longest_rule)):
            if index < len(new_rule):
                exec(f"old_rule_line.v{index} = new_rule[{index}]")
            else:
                exec(f"old_rule_line.v{index} = None")

        self._commit()
        return old_rule_line

    def update_policies(
        self,
        sec: str,
        ptype: str,
        old_rules: [
            [str],
        ],
        new_rules: [
            [str],
        ],
    ) -> None:
        """
        Update the old_rules with the new_rules in the database (storage).

        :param sec: section type
        :param ptype: policy type
        :param old_rules: the old rules that need to be modified
        :param new_rules: the new rules to replace the old rules

        :return: None
        """
        for i in range(len(old_rules)):
            self.update_policy(sec, ptype, old_rules[i], new_rules[i])

        self._commit()

    def update_filtered_policies(
        self, sec, ptype, new_rules: [[str]], field_index, *field_values
    ) -> [[str]]:
        """update_filtered_policies updates all the policies on the basis of the filter."""

        filter = Filter()
        filter.ptype = ptype

        # Creating Filter from the field_index & field_values provided
        for i in range(len(field_values)):
            if field_index <= i and i < field_index + len(field_values):
                setattr(filter, f"v{i}", field_values[i - field_index])
            else:
                break

        self._update_filtered_policies(new_rules, filter)
        self._commit()

    def _update_filtered_policies(self, new_rules, filter) -> [[str]]:
        """_update_filtered_policies updates all the policies on the basis of the filter."""

        # Load old policies
        query = self.session.query(self._db_class).filter(
            self._db_class.ptype == filter.ptype
        )
        filtered_query = self.filter_query(query, filter)
        old_rules = filtered_query.all()
        self.remove_policies("p", filter.ptype, old_rules)
        self.add_policies("p", filter.ptype, new_rules)
        self._commit()
        return old_rules


class Adapter(CoreAdapter):
    """the interface for Casbin adapters."""

    @staticmethod
    def refresh_session_decorator(func, refresh_session):
        def wrapper(*args):
            refresh_session()
            return func(*args)

        return wrapper

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.load_policy = self.refresh_session_decorator(
            self.load_policy, self.refresh_session
        )
        self.load_filtered_policy = self.refresh_session_decorator(
            self.load_filtered_policy, self.refresh_session
        )
        self.filter_query = self.refresh_session_decorator(
            self.filter_query, self.refresh_session
        )
        self.save_policy = self.refresh_session_decorator(
            self.save_policy, self.refresh_session
        )
        self.add_policy = self.refresh_session_decorator(
            self.add_policy, self.refresh_session
        )
        self.add_policies = self.refresh_session_decorator(
            self.add_policies, self.refresh_session
        )
        self.remove_policy = self.refresh_session_decorator(
            self.remove_policy, self.refresh_session
        )
        self.remove_policies = self.refresh_session_decorator(
            self.remove_policies, self.refresh_session
        )
        self.remove_filtered_policy = self.refresh_session_decorator(
            self.remove_filtered_policy, self.refresh_session
        )
        self.update_policy = self.refresh_session_decorator(
            self.update_policy, self.refresh_session
        )
        self.update_policies = self.refresh_session_decorator(
            self.update_policies, self.refresh_session
        )
        self.update_filtered_policies = self.refresh_session_decorator(
            self.update_filtered_policies, self.refresh_session
        )

    def refresh_session(self):
        if self.session:
            self.session.expire_all()
        self.session = sessionmaker(bind=self._engine)()

1	from casbin import persist	4✔
2	from sqlalchemy import Column, Integer, String	4✔
3	from sqlalchemy import create_engine, or_	4✔
4	from sqlalchemy.ext.declarative import declarative_base	4✔
5	from sqlalchemy.orm import sessionmaker	4✔
6
7	Base = declarative_base()	4✔
8
9
10	class CasbinRule(Base):	4✔
11	__tablename__ = "casbin_rule"	4✔
12
13	id = Column(Integer, primary_key=True)	4✔
14	ptype = Column(String(255))	4✔
15	v0 = Column(String(255))	4✔
16	v1 = Column(String(255))	4✔
17	v2 = Column(String(255))	4✔
18	v3 = Column(String(255))	4✔
19	v4 = Column(String(255))	4✔
20	v5 = Column(String(255))	4✔
21
22	def __str__(self):	4✔
23	arr = [self.ptype]	4✔
24	for v in (self.v0, self.v1, self.v2, self.v3, self.v4, self.v5):	4✔
25	if v is None:	4✔
26	break	4✔
27	arr.append(v)	4✔
28	return ", ".join(arr)	4✔
29
30	def __repr__(self):	4✔
31	return '<CasbinRule {}: "{}">'.format(self.id, str(self))	4✔
32
33
34	class Filter:	4✔
35	ptype = []	4✔
36	v0 = []	4✔
37	v1 = []	4✔
38	v2 = []	4✔
39	v3 = []	4✔
40	v4 = []	4✔
41	v5 = []	4✔
42
43
44	class CoreAdapter(persist.Adapter, persist.adapters.UpdateAdapter):	4✔
45	"""the interface for Casbin adapters."""
46
47	def __init__(self, engine, db_class=None, filtered=False):	4✔
48	if isinstance(engine, str):	4✔
49	self._engine = create_engine(engine)	×
50	else:
51	self._engine = engine	4✔
52
53	if db_class is None:	4✔
54	db_class = CasbinRule	4✔
55	else:
56	for attr in (	4✔
57	"id",
58	"ptype",
59	"v0",
60	"v1",
61	"v2",
62	"v3",
63	"v4",
64	"v5",
65	): # id attr was used by filter
66	if not hasattr(db_class, attr):	4✔
67	raise Exception(f"{attr} not found in custom DatabaseClass.")	×
68	Base.metadata = db_class.metadata	4✔
69
70	self._db_class = db_class	4✔
71	Base.metadata.create_all(self._engine)	4✔
72	self._filtered = filtered	4✔
73	self.session = None	4✔
74
75	def set_session(self, session):	4✔
NEW 76	self.session = session	×
77
78	def _commit(self):	4✔
79	self.session.commit()	4✔
80
81	def _close(self):	4✔
82	self.session.close()	4✔
83
84	def load_policy(self, model):	4✔
85	"""loads all policy rules from the storage."""
86	lines = self.session.query(self._db_class).all()	4✔
87	for line in lines:	4✔
88	persist.load_policy_line(str(line), model)	4✔
89	self._close()	4✔
90
91	def is_filtered(self):	4✔
92	return self._filtered	4✔
93
94	def load_filtered_policy(self, model, filter) -> None:	4✔
95	"""loads all policy rules from the storage."""
96	query = self.session.query(self._db_class)	4✔
97	filters = self.filter_query(query, filter)	4✔
98	filters = filters.all()	4✔
99
100	for line in filters:	4✔
101	persist.load_policy_line(str(line), model)	4✔
102	self._filtered = True	4✔
103	self._close()	4✔
104
105	def filter_query(self, querydb, filter):	4✔
106	for attr in ("ptype", "v0", "v1", "v2", "v3", "v4", "v5"):	4✔
107	if len(getattr(filter, attr)) > 0:	4✔
108	querydb = querydb.filter(	4✔
109	getattr(self._db_class, attr).in_(getattr(filter, attr))
110	)
111	return querydb.order_by(self._db_class.id)	4✔
112
113	def _save_policy_line(self, ptype, rule):	4✔
114	line = self._db_class(ptype=ptype)	4✔
115	for i, v in enumerate(rule):	4✔
116	setattr(line, "v{}".format(i), v)	4✔
117	self.session.add(line)	4✔
118	# commit in save_policy
119
120	def save_policy(self, model):	4✔
121	"""saves all policy rules to the storage."""
122	query = self.session.query(self._db_class)	4✔
123	query.delete()	4✔
124	for sec in ["p", "g"]:	4✔
125	if sec not in model.model.keys():	4✔
NEW 126	continue	×
127	for ptype, ast in model.model[sec].items():	4✔
128	for rule in ast.policy:	4✔
129	self._save_policy_line(ptype, rule)	4✔
130
131	self._commit()	4✔
132
133	def add_policy(self, sec, ptype, rule):	4✔
134	"""adds a policy rule to the storage."""
135	self._save_policy_line(ptype, rule)	4✔
136	self._commit()	4✔
137
138	def add_policies(self, sec, ptype, rules):	4✔
139	"""adds a policy rules to the storage."""
140	for rule in rules:	4✔
141	self._save_policy_line(ptype, rule)	4✔
142	self._commit()	4✔
143
144	def remove_policy(self, sec, ptype, rule):	4✔
145	"""removes a policy rule from the storage."""
146	query = self.session.query(self._db_class)	4✔
147	query = query.filter(self._db_class.ptype == ptype)	4✔
148	for i, v in enumerate(rule):	4✔
149	query = query.filter(getattr(self._db_class, "v{}".format(i)) == v)	4✔
150	query.delete()	4✔
151
152	self._commit()	4✔
153
154	def remove_policies(self, sec, ptype, rules):	4✔
155	"""remove policy rules from the storage."""
156	if not rules:	4✔
157	return	×
158	query = self.session.query(self._db_class)	4✔
159	query = query.filter(self._db_class.ptype == ptype)	4✔
160	rules = zip(*rules)	4✔
161	for i, rule in enumerate(rules):	4✔
162	query = query.filter(	4✔
163	or_(getattr(self._db_class, "v{}".format(i)) == v for v in rule)
164	)
165	query.delete()	4✔
166	self._commit()	4✔
167
168	def remove_filtered_policy(self, sec, ptype, field_index, *field_values):	4✔
169	"""removes policy rules that match the filter from the storage.
170	This is part of the Auto-Save feature.
171	"""
172	query = self.session.query(self._db_class).filter(self._db_class.ptype == ptype)	4✔
173
174	if not (0 <= field_index <= 5):	4✔
NEW 175	return False	×
176	if not (1 <= field_index + len(field_values) <= 6):	4✔
NEW 177	return False	×
178	for i, v in enumerate(field_values):	4✔
179	if v != "":	4✔
180	v_value = getattr(self._db_class, "v{}".format(field_index + i))	4✔
181	query = query.filter(v_value == v)	4✔
182	query.delete()	4✔
183	self._commit()	4✔
184
185	def update_policy(	4✔
186	self, sec: str, ptype: str, old_rule: [str], new_rule: [str]
187	) -> None:
188	"""
189	Update the old_rule with the new_rule in the database (storage).
190
191	:param sec: section type
192	:param ptype: policy type
193	:param old_rule: the old rule that needs to be modified
194	:param new_rule: the new rule to replace the old rule
195
196	:return: None
197	"""
198
199	query = self.session.query(self._db_class).filter(self._db_class.ptype == ptype)	4✔
200
201	# locate the old rule
202	for index, value in enumerate(old_rule):	4✔
203	v_value = getattr(self._db_class, "v{}".format(index))	4✔
204	query = query.filter(v_value == value)	4✔
205
206	# need the length of the longest_rule to perform overwrite
207	longest_rule = old_rule if len(old_rule) > len(new_rule) else new_rule	4✔
208	old_rule_line = query.one()	4✔
209
210	# overwrite the old rule with the new rule
211	for index in range(len(longest_rule)):	4✔
212	if index < len(new_rule):	4✔
213	exec(f"old_rule_line.v{index} = new_rule[{index}]")	4✔
214	else:
NEW 215	exec(f"old_rule_line.v{index} = None")	×
216
217	self._commit()	4✔
218	return old_rule_line	4✔
219
220	def update_policies(	4✔
221	self,
222	sec: str,
223	ptype: str,
224	old_rules: [
225	[str],
226	],
227	new_rules: [
228	[str],
229	],
230	) -> None:
231	"""
232	Update the old_rules with the new_rules in the database (storage).
233
234	:param sec: section type
235	:param ptype: policy type
236	:param old_rules: the old rules that need to be modified
237	:param new_rules: the new rules to replace the old rules
238
239	:return: None
240	"""
241	for i in range(len(old_rules)):	4✔
242	self.update_policy(sec, ptype, old_rules[i], new_rules[i])	4✔
243
244	self._commit()	4✔
245
246	def update_filtered_policies(	4✔
247	self, sec, ptype, new_rules: [[str]], field_index, *field_values
248	) -> [[str]]:
249	"""update_filtered_policies updates all the policies on the basis of the filter."""
250
251	filter = Filter()	4✔
252	filter.ptype = ptype	4✔
253
254	# Creating Filter from the field_index & field_values provided
255	for i in range(len(field_values)):	4✔
256	if field_index <= i and i < field_index + len(field_values):	4✔
257	setattr(filter, f"v{i}", field_values[i - field_index])	4✔
258	else:
259	break	×
260
261	self._update_filtered_policies(new_rules, filter)	4✔
NEW 262	self._commit()	×
263
264	def _update_filtered_policies(self, new_rules, filter) -> [[str]]:	4✔
265	"""_update_filtered_policies updates all the policies on the basis of the filter."""
266
267	# Load old policies
268	query = self.session.query(self._db_class).filter(	4✔
269	self._db_class.ptype == filter.ptype
270	)
271	filtered_query = self.filter_query(query, filter)	4✔
NEW 272	old_rules = filtered_query.all()	×
NEW 273	self.remove_policies("p", filter.ptype, old_rules)	×
NEW 274	self.add_policies("p", filter.ptype, new_rules)	×
NEW 275	self._commit()	×
NEW 276	return old_rules	×
277
278
279	class Adapter(CoreAdapter):	4✔
280	"""the interface for Casbin adapters."""
281
282	@staticmethod	4✔
283	def refresh_session_decorator(func, refresh_session):	2✔
284	def wrapper(*args):	4✔
285	refresh_session()	4✔
286	return func(*args)	4✔
287
288	return wrapper	4✔
289
290	def __init__(self, args, *kwargs):	4✔
291	super().__init__(args, *kwargs)	4✔
292
293	self.load_policy = self.refresh_session_decorator(	4✔
294	self.load_policy, self.refresh_session
295	)
296	self.load_filtered_policy = self.refresh_session_decorator(	4✔
297	self.load_filtered_policy, self.refresh_session
298	)
299	self.filter_query = self.refresh_session_decorator(	4✔
300	self.filter_query, self.refresh_session
301	)
302	self.save_policy = self.refresh_session_decorator(	4✔
303	self.save_policy, self.refresh_session
304	)
305	self.add_policy = self.refresh_session_decorator(	4✔
306	self.add_policy, self.refresh_session
307	)
308	self.add_policies = self.refresh_session_decorator(	4✔
309	self.add_policies, self.refresh_session
310	)
311	self.remove_policy = self.refresh_session_decorator(	4✔
312	self.remove_policy, self.refresh_session
313	)
314	self.remove_policies = self.refresh_session_decorator(	4✔
315	self.remove_policies, self.refresh_session
316	)
317	self.remove_filtered_policy = self.refresh_session_decorator(	4✔
318	self.remove_filtered_policy, self.refresh_session
319	)
320	self.update_policy = self.refresh_session_decorator(	4✔
321	self.update_policy, self.refresh_session
322	)
323	self.update_policies = self.refresh_session_decorator(	4✔
324	self.update_policies, self.refresh_session
325	)
326	self.update_filtered_policies = self.refresh_session_decorator(	4✔
327	self.update_filtered_policies, self.refresh_session
328	)
329
330	def refresh_session(self):	4✔
331	if self.session:	4✔
332	self.session.expire_all()	4✔
333	self.session = sessionmaker(bind=self._engine)()	4✔

pycasbin / sqlalchemy-adapter / 3060734495

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