7589020954

Committed 19 Jan 2024 08:48PM UTC coverage: 89.866%. Remained the same

Build # 7589020954

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1091 from nolar/fix-liveness-tests

Satisfy aiohttp's deprecation of shutdown timeout kwargs

Run Details

2093 of 2401 branches covered (0.0%)

Branch coverage included in aggregate %.

2 of 2 new or added lines in 1 file covered. (100.0%)

46 existing lines in 6 files now uncovered.

5471 of 6016 relevant lines covered (90.94%)

8.11 hits per line

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

100.0

/kopf/_cogs/structs/dicts.py

"""
Some basic dicts and field-in-a-dict manipulation helpers.
"""
import collections.abc
import enum
from typing import Any, Callable, Generic, Iterable, Iterator, List, \
                   Mapping, MutableMapping, Optional, Tuple, TypeVar, Union

from kopf._cogs.helpers import thirdparty

FieldPath = Tuple[str, ...]
FieldSpec = Union[None, str, FieldPath, List[str]]

_T = TypeVar('_T')
_K = TypeVar('_K', bound=str)  # int & bool keys are possible but discouraged
_V = TypeVar('_V')


class _UNSET(enum.Enum):
    token = enum.auto()


def parse_field(
        field: FieldSpec,
) -> FieldPath:
    """
    Convert any field into a tuple of nested sub-fields.

    Supported notations:

    * ``None`` (for root of a dict).
    * ``"field.subfield"``
    * ``("field", "subfield")``
    * ``["field", "subfield"]``
    """
    if field is None:
        return ()
    elif isinstance(field, str):
        return tuple(field.split('.'))
    elif isinstance(field, (list, tuple)):
        return tuple(field)
    else:
        raise ValueError(f"Field must be either a str, or a list/tuple. Got {field!r}")


def resolve_obj(
        d: Union[None, thirdparty.KubernetesModel, Mapping[Any, Any]],
        field: FieldSpec,
        default: Union[_T, _UNSET] = _UNSET.token,
) -> Union[Any, _T]:
    """
    Mirrors `resolve`, but for a nested mix of dict keys & object attributes.

    While `resolve` is used mostly in certain dictionaries (e.g. diffs),
    this function is used for walking over 3rd-party API objects & models
    with nested structures. The algorithm is essentially the same.
    """
    path = parse_field(field)
    try:
        result = d
        for key in path:
            if isinstance(result, collections.abc.Mapping):
                result = result[key]
            elif isinstance(result, thirdparty.KubernetesModel):
                attrmap: Mapping[str, str] = getattr(result, 'attribute_map', {})
                attrs = [attr for attr, schema_key in attrmap.items() if schema_key == key]
                key = attrs[0] if attrs else key
                result = getattr(result, key)
            elif not isinstance(result, (tuple, list, set, frozenset, str, bytes)):
                result = getattr(result, key)
            elif not isinstance(default, _UNSET):
                return default
            else:
                raise TypeError(f"The structure has no field {key!r}: {result!r}")
        return result
    except (AttributeError, KeyError):
        if not isinstance(default, _UNSET):
            return default
        raise


def resolve(
        d: Optional[Mapping[Any, Any]],
        field: FieldSpec,
        default: Union[_T, _UNSET] = _UNSET.token,
) -> Union[Any, _T]:
    """
    Retrieve a nested sub-field from a dict.

    If ``default`` is provided, then all non-existent and non-mapping values
    are assumed to be empty dictionaries, and ``default`` is returned.

    Otherwise (with no default), attempts to get the inexistent keys will
    raise either a ``TypeError`` or ``KeyError``:

    * ``KeyError`` for actual absence of keys while the structures are correct.
    * ``TypeError`` for attempting to get a key for a non-dictionary:
      e.g. ``None['key']``, ``"string"['key']``, ``123['key']``, etc.

    This essentially means the "safe" mode of resolution, where the obvious
    errors are silenced, as we cannot dive deep into non-dictionary values.

    Silencing errors goes against The Zen of Python, but we need this for K8s:
    if the resources are corrupted externally (e.g. by editing manually),
    we ignore that corrupted data as if there is no data at all,
    and continue running instead of unrecoverably failing the processing.

    Examples of data that can be corrupted:

    * diff-base and progress in the status stanza (``status.kopf.progress``);
      this does not apply to annotations, as their structure is ensured by K8s.
    * field-specific handers for fields (e.g. ``spec.struct.field``)
      when the parent structure (``spec.struct`` or even ``spec``)
      is not a mapping or absent.
    """
    path = parse_field(field)
    try:
        result = d
        for key in path:
            if isinstance(result, collections.abc.Mapping):
                result = result[key]
            elif not isinstance(default, _UNSET):
                return default
            else:
                raise TypeError(f"The structure is not a dict with field {key!r}: {result!r}")
        return result
    except KeyError:
        if not isinstance(default, _UNSET):
            return default
        raise


def ensure(
        d: MutableMapping[Any, Any],
        field: FieldSpec,
        value: Any,
) -> None:
    """
    Force-set a nested sub-field in a dict.

    If some levels of parents are missing, they are created as empty dicts
    (this what makes it "ensuring", not just "setting").
    """
    result = d
    path = parse_field(field)
    if not path:
        raise ValueError("Setting a root of a dict is impossible. Provide the specific fields.")
    for key in path[:-1]:
        try:
            result = result[key]
        except KeyError:
            result = result.setdefault(key, {})
    result[path[-1]] = value


def remove(
        d: MutableMapping[Any, Any],
        field: FieldSpec,
) -> None:
    """
    Remove a nested sub-field from a dict, and all empty parents (optionally).

    All intermediate parents that become empty after the removal are also
    removed, making the whole original dict cleaner. For single-field removal,
    use a built-in ``del d[key]`` operation.

    If the target key is absent already, or any of the intermediate parents
    is absent (which implies that the target key is also absent), no error
    is raised, since the goal of deletion is achieved. The empty parents are
    anyway removed.
    """
    path = parse_field(field)
    if not path:
        raise ValueError("Removing a root of a dict is impossible. Provide a specific field.")

    elif len(path) == 1:
        try:
            del d[path[0]]
        except KeyError:
            pass  # already absent

    else:
        try:
            # Recursion is the easiest way to implement it, assuming the bodies/patches are shallow.
            remove(d[path[0]], path[1:])
        except KeyError:
            pass  # already absent
        else:
            # Clean the parent dict if it has become empty due to deletion of the only sub-key.
            # Upper parents will be handled by upper recursion functions.
            if d[path[0]] == {}:  # but not None, and not False, etc.
                del d[path[0]]


def cherrypick(
        src: Mapping[Any, Any],
        dst: MutableMapping[Any, Any],
        fields: Optional[Iterable[FieldSpec]],
        picker: Optional[Callable[[_T], _T]] = None,
) -> None:
    """
    Copy all specified fields between dicts (from src to dst).
    """
    picker = picker if picker is not None else lambda x: x
    fields = fields if fields is not None else []
    for field in fields:
        try:
            ensure(dst, field, picker(resolve(src, field)))
        except KeyError:
            pass  # absent in the source, nothing to merge


def walk(
        objs: Union[_T,
                    Iterable[_T],
                    Iterable[Union[_T,
                                   Iterable[_T]]]],
        *,
        nested: Optional[Iterable[FieldSpec]] = None,
) -> Iterator[_T]:
    """
    Iterate over objects, flattening the lists/tuples/iterables recursively.

    In plain English, the source is either an object, or a list/tuple/iterable
    of objects with any level of nesting. The dicts/mappings are excluded,
    despite they are iterables too, as they are treated as objects themselves.

    For the output, it yields all the objects in a flat iterable suitable for::

        for obj in walk(objs):
            pass

    The type declares only 2-level nesting, but this is done only
    for type-checker's limitations. The actual nesting can be infinite.
    It is highly unlikely that there will be anything deeper than one level.
    """
    if objs is None:
        pass
    elif isinstance(objs, thirdparty.PykubeObject):
        # Pykube is yielded as an underlying dict, never as its own class.
        yield from walk(objs.obj, nested=nested)
    elif isinstance(objs, thirdparty.KubernetesModel):
        yield objs  # type: ignore
        for subfield in (nested if nested is not None else []):
            try:
                yield resolve_obj(objs, parse_field(subfield))
            except (AttributeError, KeyError):
                pass  # do not dive deep into non-existent fields or non-dicts
    elif isinstance(objs, collections.abc.Mapping):
        yield objs  # type: ignore
        for subfield in (nested if nested is not None else []):
            try:
                yield resolve(objs, parse_field(subfield))
            except KeyError:
                pass  # avoid diving into non-dicts, ignore them
    elif isinstance(objs, collections.abc.Iterable):
        for obj in objs:
            yield from walk(obj, nested=nested)
    else:
        yield objs  # NB: not a mapping or a known type => no nested sub-fields.


class MappingView(Mapping[_K, _V], Generic[_K, _V]):
    """
    A lazy resolver for the "on-demand" dict keys.

    This is needed to have :kwarg:`spec`, :kwarg:`status`, and other fields
    to be *assumed* as dicts, even if they are actually not present.
    And to prevent their implicit creation with ``.setdefault('spec', {})``,
    which produces unwanted side-effects (actually adds this field).

    >>> body = {}
    >>> spec = MappingView(body, 'spec')
    >>> spec.get('field', 'default')
    ... 'default'
    >>> body['spec'] = {'field': 'value'}
    >>> spec.get('field', 'default')
    ... 'value'
    """
    _src: Mapping[_K, _V]

    def __init__(self, __src: Mapping[Any, Any], __path: FieldSpec = None) -> None:
        super().__init__()
        self._src = __src
        self._path = parse_field(__path)

    def __repr__(self) -> str:
        return repr(dict(self))

    def __len__(self) -> int:
        return len(resolve(self._src, self._path, {}))

    def __iter__(self) -> Iterator[Any]:
        return iter(resolve(self._src, self._path, {}))

    def __getitem__(self, item: _K) -> _V:
        return resolve(self._src, self._path + (item,))


class MutableMappingView(MappingView[_K, _V], MutableMapping[_K, _V], Generic[_K, _V]):
    """
    A mapping view with values stored and sub-dicts auto-created.

    >>> patch = {}
    >>> status = MutableMappingView(patch, 'status')
    >>> status.get('field', 'default')
    ... 'default'
    >>> patch
    ... {}
    >>> status['field'] = 'value'
    >>> patch
    ... {'status': {'field': 'value'}}
    >>> status.get('field', 'default')
    ... 'value'
    """
    _src: MutableMapping[_K, _V]  # typing clarification

    def __delitem__(self, item: _K) -> None:
        d = resolve(self._src, self._path)
        del d[item]

    def __setitem__(self, item: _K, value: _V) -> None:
        ensure(self._src, self._path + (item,), value)


class ReplaceableMappingView(MappingView[_K, _V], Generic[_K, _V]):
    """
    A mapping view where the whole source can be replaced atomically.

    All derived mapping views that use this mapping view as their source will
    immediately notice the change.

    The method names are intentionally long and multi-word -- to not have
    potential collisions with regular expected attributes/properties.

    >>> body = ReplaceableMappingView()
    >>> spec = MappingView(body, 'spec')
    >>> spec.get('field', 'default')
    ... 'default'
    >>> body._replace_with({'spec': {'field': 'value'}})
    >>> spec.get('field', 'default')
    ... 'value'
    """

    def _replace_from(self, __src: MappingView[_K, _V]) -> None:
        self._src = __src._src

    def _replace_with(self, __src: Mapping[_K, _V]) -> None:
        self._src = __src

1	"""
2	Some basic dicts and field-in-a-dict manipulation helpers.
3	"""
4	import collections.abc	9✔
5	import enum	9✔
6	from typing import Any, Callable, Generic, Iterable, Iterator, List, \	9✔
7	Mapping, MutableMapping, Optional, Tuple, TypeVar, Union
8
9	from kopf._cogs.helpers import thirdparty	9✔
10
11	FieldPath = Tuple[str, ...]	9✔
12	FieldSpec = Union[None, str, FieldPath, List[str]]	9✔
13
14	_T = TypeVar('_T')	9✔
15	_K = TypeVar('_K', bound=str) # int & bool keys are possible but discouraged	9✔
16	_V = TypeVar('_V')	9✔
17
18
19	class _UNSET(enum.Enum):	9✔
20	token = enum.auto()	9✔
21
22
23	def parse_field(	9✔
24	field: FieldSpec,
25	) -> FieldPath:
26	"""
27	Convert any field into a tuple of nested sub-fields.
28
29	Supported notations:
30
31	* ``None`` (for root of a dict).
32	* ``"field.subfield"``
33	* ``("field", "subfield")``
34	* ``["field", "subfield"]``
35	"""
36	if field is None:	9✔
37	return ()	9✔
38	elif isinstance(field, str):	9✔
39	return tuple(field.split('.'))	9✔
40	elif isinstance(field, (list, tuple)):	9✔
41	return tuple(field)	9✔
42	else:
43	raise ValueError(f"Field must be either a str, or a list/tuple. Got {field!r}")	9✔
44
45
46	def resolve_obj(	9✔
47	d: Union[None, thirdparty.KubernetesModel, Mapping[Any, Any]],
48	field: FieldSpec,
49	default: Union[_T, _UNSET] = _UNSET.token,
50	) -> Union[Any, _T]:
51	"""
52	Mirrors `resolve`, but for a nested mix of dict keys & object attributes.
53
54	While `resolve` is used mostly in certain dictionaries (e.g. diffs),
55	this function is used for walking over 3rd-party API objects & models
56	with nested structures. The algorithm is essentially the same.
57	"""
58	path = parse_field(field)	9✔
59	try:	9✔
60	result = d	9✔
61	for key in path:	9✔
62	if isinstance(result, collections.abc.Mapping):	9✔
63	result = result[key]	9✔
64	elif isinstance(result, thirdparty.KubernetesModel):	9✔
65	attrmap: Mapping[str, str] = getattr(result, 'attribute_map', {})	9✔
66	attrs = [attr for attr, schema_key in attrmap.items() if schema_key == key]	9✔
67	key = attrs[0] if attrs else key	9✔
68	result = getattr(result, key)	9✔
69	elif not isinstance(result, (tuple, list, set, frozenset, str, bytes)):	9✔
70	result = getattr(result, key)	9✔
71	elif not isinstance(default, _UNSET):	9✔
72	return default	9✔
73	else:
74	raise TypeError(f"The structure has no field {key!r}: {result!r}")	9✔
75	return result	9✔
76	except (AttributeError, KeyError):	9✔
77	if not isinstance(default, _UNSET):	9✔
78	return default	9✔
79	raise	9✔
80
81
82	def resolve(	9✔
83	d: Optional[Mapping[Any, Any]],
84	field: FieldSpec,
85	default: Union[_T, _UNSET] = _UNSET.token,
86	) -> Union[Any, _T]:
87	"""
88	Retrieve a nested sub-field from a dict.
89
90	If ``default`` is provided, then all non-existent and non-mapping values
91	are assumed to be empty dictionaries, and ``default`` is returned.
92
93	Otherwise (with no default), attempts to get the inexistent keys will
94	raise either a ``TypeError`` or ``KeyError``:
95
96	* ``KeyError`` for actual absence of keys while the structures are correct.
97	* ``TypeError`` for attempting to get a key for a non-dictionary:
98	e.g. ``None['key']``, ``"string"['key']``, ``123['key']``, etc.
99
100	This essentially means the "safe" mode of resolution, where the obvious
101	errors are silenced, as we cannot dive deep into non-dictionary values.
102
103	Silencing errors goes against The Zen of Python, but we need this for K8s:
104	if the resources are corrupted externally (e.g. by editing manually),
105	we ignore that corrupted data as if there is no data at all,
106	and continue running instead of unrecoverably failing the processing.
107
108	Examples of data that can be corrupted:
109
110	* diff-base and progress in the status stanza (``status.kopf.progress``);
111	this does not apply to annotations, as their structure is ensured by K8s.
112	* field-specific handers for fields (e.g. ``spec.struct.field``)
113	when the parent structure (``spec.struct`` or even ``spec``)
114	is not a mapping or absent.
115	"""
116	path = parse_field(field)	9✔
117	try:	9✔
118	result = d	9✔
119	for key in path:	9✔
120	if isinstance(result, collections.abc.Mapping):	9✔
121	result = result[key]	9✔
122	elif not isinstance(default, _UNSET):	9✔
123	return default	9✔
124	else:
125	raise TypeError(f"The structure is not a dict with field {key!r}: {result!r}")	9✔
126	return result	9✔
127	except KeyError:	9✔
128	if not isinstance(default, _UNSET):	9✔
129	return default	9✔
130	raise	9✔
131
132
133	def ensure(	9✔
134	d: MutableMapping[Any, Any],
135	field: FieldSpec,
136	value: Any,
137	) -> None:
138	"""
139	Force-set a nested sub-field in a dict.
140
141	If some levels of parents are missing, they are created as empty dicts
142	(this what makes it "ensuring", not just "setting").
143	"""
144	result = d	9✔
145	path = parse_field(field)	9✔
146	if not path:	9✔
147	raise ValueError("Setting a root of a dict is impossible. Provide the specific fields.")	9✔
148	for key in path[:-1]:	9✔
149	try:	9✔
150	result = result[key]	9✔
151	except KeyError:	9✔
152	result = result.setdefault(key, {})	9✔
153	result[path[-1]] = value	9✔
154
155
156	def remove(	9✔
157	d: MutableMapping[Any, Any],
158	field: FieldSpec,
159	) -> None:
160	"""
161	Remove a nested sub-field from a dict, and all empty parents (optionally).
162
163	All intermediate parents that become empty after the removal are also
164	removed, making the whole original dict cleaner. For single-field removal,
165	use a built-in ``del d[key]`` operation.
166
167	If the target key is absent already, or any of the intermediate parents
168	is absent (which implies that the target key is also absent), no error
169	is raised, since the goal of deletion is achieved. The empty parents are
170	anyway removed.
171	"""
172	path = parse_field(field)	9✔
173	if not path:	9✔
174	raise ValueError("Removing a root of a dict is impossible. Provide a specific field.")	9✔
175
176	elif len(path) == 1:	9✔
177	try:	9✔
178	del d[path[0]]	9✔
179	except KeyError:	9✔
180	pass # already absent	9✔
181
182	else:
183	try:	9✔
184	# Recursion is the easiest way to implement it, assuming the bodies/patches are shallow.
185	remove(d[path[0]], path[1:])	9✔
186	except KeyError:	9✔
187	pass # already absent	9✔
188	else:
189	# Clean the parent dict if it has become empty due to deletion of the only sub-key.
190	# Upper parents will be handled by upper recursion functions.
191	if d[path[0]] == {}: # but not None, and not False, etc.	9✔
192	del d[path[0]]	9✔
193
194
195	def cherrypick(	9✔
196	src: Mapping[Any, Any],
197	dst: MutableMapping[Any, Any],
198	fields: Optional[Iterable[FieldSpec]],
199	picker: Optional[Callable[[_T], _T]] = None,
200	) -> None:
201	"""
202	Copy all specified fields between dicts (from src to dst).
203	"""
204	picker = picker if picker is not None else lambda x: x	9✔
205	fields = fields if fields is not None else []	9✔
206	for field in fields:	9✔
207	try:	9✔
208	ensure(dst, field, picker(resolve(src, field)))	9✔
209	except KeyError:	9✔
210	pass # absent in the source, nothing to merge	9✔
211
212
213	def walk(	9✔
214	objs: Union[_T,
215	Iterable[_T],
216	Iterable[Union[_T,
217	Iterable[_T]]]],
218	*,
219	nested: Optional[Iterable[FieldSpec]] = None,
220	) -> Iterator[_T]:
221	"""
222	Iterate over objects, flattening the lists/tuples/iterables recursively.
223
224	In plain English, the source is either an object, or a list/tuple/iterable
225	of objects with any level of nesting. The dicts/mappings are excluded,
226	despite they are iterables too, as they are treated as objects themselves.
227
228	For the output, it yields all the objects in a flat iterable suitable for::
229
230	for obj in walk(objs):
231	pass
232
233	The type declares only 2-level nesting, but this is done only
234	for type-checker's limitations. The actual nesting can be infinite.
235	It is highly unlikely that there will be anything deeper than one level.
236	"""
237	if objs is None:	9✔
238	pass	9✔
239	elif isinstance(objs, thirdparty.PykubeObject):	9✔
240	# Pykube is yielded as an underlying dict, never as its own class.
UNCOV 241	yield from walk(objs.obj, nested=nested)	4✔
242	elif isinstance(objs, thirdparty.KubernetesModel):	9✔
UNCOV 243	yield objs # type: ignore	4✔
UNCOV 244	for subfield in (nested if nested is not None else []):	4✔
UNCOV 245	try:	4✔
UNCOV 246	yield resolve_obj(objs, parse_field(subfield))	4✔
UNCOV 247	except (AttributeError, KeyError):	4✔
UNCOV 248	pass # do not dive deep into non-existent fields or non-dicts	4✔
249	elif isinstance(objs, collections.abc.Mapping):	9✔
250	yield objs # type: ignore	9✔
251	for subfield in (nested if nested is not None else []):	9✔
252	try:	9✔
253	yield resolve(objs, parse_field(subfield))	9✔
254	except KeyError:	9✔
255	pass # avoid diving into non-dicts, ignore them	9✔
256	elif isinstance(objs, collections.abc.Iterable):	9✔
257	for obj in objs:	9✔
258	yield from walk(obj, nested=nested)	9✔
259	else:
260	yield objs # NB: not a mapping or a known type => no nested sub-fields.	9✔
261
262
263	class MappingView(Mapping[_K, _V], Generic[_K, _V]):	9✔
264	"""
265	A lazy resolver for the "on-demand" dict keys.
266
267	This is needed to have :kwarg:`spec`, :kwarg:`status`, and other fields
268	to be assumed as dicts, even if they are actually not present.
269	And to prevent their implicit creation with ``.setdefault('spec', {})``,
270	which produces unwanted side-effects (actually adds this field).
271
272	>>> body = {}
273	>>> spec = MappingView(body, 'spec')
274	>>> spec.get('field', 'default')
275	... 'default'
276	>>> body['spec'] = {'field': 'value'}
277	>>> spec.get('field', 'default')
278	... 'value'
279	"""
280	_src: Mapping[_K, _V]	9✔
281
282	def __init__(self, __src: Mapping[Any, Any], __path: FieldSpec = None) -> None:	9✔
283	super().__init__()	9✔
284	self._src = __src	9✔
285	self._path = parse_field(__path)	9✔
286
287	def __repr__(self) -> str:	9✔
288	return repr(dict(self))	9✔
289
290	def __len__(self) -> int:	9✔
291	return len(resolve(self._src, self._path, {}))	9✔
292
293	def __iter__(self) -> Iterator[Any]:	9✔
294	return iter(resolve(self._src, self._path, {}))	9✔
295
296	def __getitem__(self, item: _K) -> _V:	9✔
297	return resolve(self._src, self._path + (item,))	9✔
298
299
300	class MutableMappingView(MappingView[_K, _V], MutableMapping[_K, _V], Generic[_K, _V]):	9✔
301	"""
302	A mapping view with values stored and sub-dicts auto-created.
303
304	>>> patch = {}
305	>>> status = MutableMappingView(patch, 'status')
306	>>> status.get('field', 'default')
307	... 'default'
308	>>> patch
309	... {}
310	>>> status['field'] = 'value'
311	>>> patch
312	... {'status': {'field': 'value'}}
313	>>> status.get('field', 'default')
314	... 'value'
315	"""
316	_src: MutableMapping[_K, _V] # typing clarification	9✔
317
318	def __delitem__(self, item: _K) -> None:	9✔
319	d = resolve(self._src, self._path)	9✔
320	del d[item]	9✔
321
322	def __setitem__(self, item: _K, value: _V) -> None:	9✔
323	ensure(self._src, self._path + (item,), value)	9✔
324
325
326	class ReplaceableMappingView(MappingView[_K, _V], Generic[_K, _V]):	9✔
327	"""
328	A mapping view where the whole source can be replaced atomically.
329
330	All derived mapping views that use this mapping view as their source will
331	immediately notice the change.
332
333	The method names are intentionally long and multi-word -- to not have
334	potential collisions with regular expected attributes/properties.
335
336	>>> body = ReplaceableMappingView()
337	>>> spec = MappingView(body, 'spec')
338	>>> spec.get('field', 'default')
339	... 'default'
340	>>> body._replace_with({'spec': {'field': 'value'}})
341	>>> spec.get('field', 'default')
342	... 'value'
343	"""
344
345	def _replace_from(self, __src: MappingView[_K, _V]) -> None:	9✔
346	self._src = __src._src	9✔
347
348	def _replace_with(self, __src: Mapping[_K, _V]) -> None:	9✔
349	self._src = __src	9✔

nolar / kopf / 7589020954

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