chrisjsewell / pytest-notebook / 104

pending completion

Build # 104

Build Type

Pull #11

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/chrisjsewell/pytest-notebook/commit/<a class=hub.com/chrisjsewell/pytest-notebook/commit/<a class="double-link" href="https://git"><a class=hub.com/chrisjsewell/pytest-notebook/commit/<a class="double-link" href="https://git"><a class=hub.com/chrisjsewell/pytest-notebook/commit/<a class="double-link" href="https://git"><a class=hub.com/chrisjsewell/pytest-notebook/commit/bcb66f0dca31192e1e6e59ee8c19a124247b6331">bcb66f0dc">&lt;a href=&quot;https://github.com/chrisjsewell/pytest-notebook/commit/</a><a class="double-link" href="https://github.com/chrisjsewell/pytest-notebook/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/chrisjsewell/pytest-notebook/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/chrisjsewell/pytest-notebook/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/chrisjsewell/pytest-notebook/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/chrisjsewell/pytest-notebook/commit/bcb66f0dca31192e1e6e59ee8c19a124247b6331">bcb66f0dc</a><a href="https://github.com/chrisjsewell/pytest-notebook/commit/bcb66f0dca31192e1e6e59ee8c19a124247b6331">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/chrisjsewell/pytest-notebook/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/chrisjsewell/pytest-notebook/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/chrisjsewell/pytest-notebook/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/chrisjsewell/pytest-notebook/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/chrisjsewell/pytest-notebook/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/chrisjsewell/pytest-notebook/commit/bcb66f0dca31192e1e6e59ee8c19a124247b6331&quot;&gt;bcb66f0dc&lt;/a&gt;&lt;a href=&quot;https://github.com/chrisjsewell/pytest-notebook/commit/bcb66f0dca31192e1e6e59ee8c19a124247b6331&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com... (continued)

Pull Request Pull Request #11: Fix for PyTest6 API: use from_parent rather than constructor

Run Details

5 of 5 new or added lines in 4 files covered. (100.0%)

717 of 863 relevant lines covered (83.08%)

1.66 hits per line

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

77.22

/pytest_notebook/nb_regression.py

"""Jupyter Notebook Regression Test Class."""
import copy
import logging
import os
import sys
from typing import Any, List, TextIO, Tuple, Union

import attr
from attr.validators import instance_of
from nbdime.diff_format import DiffEntry
import nbformat
from nbformat import NotebookNode

try:
    # coverage is an optional dependency
    from coverage import Coverage as CoverageType
except ImportError:
    CoverageType = Any

from pytest_notebook.diffing import diff_notebooks, diff_to_string, filter_diff
from pytest_notebook.execution import (
    execute_notebook,
    HELP_COVERAGE,
    HELP_COVERAGE_CONFIG,
    HELP_COVERAGE_SOURCE,
)
from pytest_notebook.notebook import (
    load_notebook_with_config,
    regex_replace_nb,
    validate_regex_replace,
)
from pytest_notebook.post_processors import (
    ENTRY_POINT_NAME,
    list_processor_names,
    load_processor,
)
from pytest_notebook.utils import autodoc

logger = logging.getLogger(__name__)

HELP_EXEC_NOTEBOOK = (
    "Create a new notebook, by executing all cells in the original notebook"
)
HELP_EXEC_CWD = (
    "Path to the directory which the notebook will run in "
    "(defaults to directory of notebook)."
)
HELP_EXEC_TIMEOUT = "The maximum time to wait (in seconds) for execution of each cell."
HELP_EXEC_ALLOW_ERRORS = (
    "Do not stop execution after the first unexpected exception "
    "(where cell is not tagged ``raises-exception``)."
)
HELP_DIFF_REPLACE = (
    "A list of regex replacements to apply before diffing, "
    r"e.g. ``[('/cells/*/outputs', '\d{2,4}-\d{1,2}-\d{1,2}', 'DATE-STAMP')]``."
)
HELP_DIFF_IGNORE = (
    "List of diff paths to ignore, e.g. '/cells/1/outputs' or '/cells/\\*/metadata'."
)
HELP_DIFF_USE_COLOR = "Use ANSI color code escapes for text output."
HELP_DIFF_COLOR_WORDS = "Highlight changed words using only colors."
HELP_FORCE_REGEN = (
    "Re-generate notebook files, if no unexpected execution errors, "
    "and an output path has been supplied."
)
HELP_POST_PROCS = (
    "post-processors to apply to the new workbook, "
    f"relating to entry points in the '{ENTRY_POINT_NAME}' group"
)
HELP_COVERAGE_MERGE = "A coverage.Coverage instance, to merge coverage results with."


class NBRegressionError(Exception):
    """Exception to signal a regression test fail."""


@autodoc
@attr.s(frozen=True, slots=True, repr=False)
class NBRegressionResult:
    """A class to store the result of ``NBRegressionFixture.check``."""

    nb_initial: NotebookNode = attr.ib(
        validator=instance_of(NotebookNode), metadata={"help": "Initial notebook."}
    )
    nb_final: NotebookNode = attr.ib(
        validator=instance_of(NotebookNode),
        metadata={"help": "Notebook after execution and post-processing."},
    )

    diff_full: List[DiffEntry] = attr.ib(
        metadata={"help": "Full diff of initial/final notebooks."}
    )
    diff_filtered: List[DiffEntry] = attr.ib(
        metadata={
            "help": (
                "Diff of initial/final notebooks, "
                "filtered according to the parsed configuration."
            )
        }
    )
    diff_string: str = attr.ib(
        validator=instance_of(str),
        metadata={"help": "The formatte string of diff_filtered."},
    )

    process_resources: dict = attr.ib(
        attr.Factory(dict),
        instance_of(dict),
        metadata={"help": "Resources returned from notebook processors."},
    )

    def __repr__(self):
        """Represent the class instance."""
        return (
            f"NBRegressionResult(diff_full_length={len(self.diff_full)},"
            f"diff_filtered_length={len(self.diff_filtered)})"
        )


@autodoc
@attr.s
class NBRegressionFixture:
    """Class to perform Jupyter Notebook Regression tests."""

    exec_notebook: bool = attr.ib(
        True, instance_of(bool), metadata={"help": HELP_EXEC_NOTEBOOK}
    )
    exec_cwd: Union[str, None] = attr.ib(
        None, instance_of((type(None), str)), metadata={"help": HELP_EXEC_CWD}
    )

    @exec_cwd.validator
    def _validate_exec_cwd(self, attribute, value):
        if value is None:
            return
        if not isinstance(value, str):
            raise TypeError("exec_cwd must be None or a string")
        if not os.path.isdir(value):
            raise IOError("exec_cwd='{}' is not an existing directory".format(value))

    exec_allow_errors: bool = attr.ib(
        False, instance_of(bool), metadata={"help": HELP_EXEC_ALLOW_ERRORS}
    )
    exec_timeout: int = attr.ib(
        120, instance_of((int, float)), metadata={"help": HELP_EXEC_TIMEOUT}
    )

    @exec_timeout.validator
    def _validate_exec_timeout(self, attribute, value):
        if not isinstance(value, int):
            raise TypeError("exec_timeout must be an integer")
        if value <= 0:
            raise ValueError("exec_timeout must be larger than 0")

    coverage: bool = attr.ib(False, metadata={"help": HELP_COVERAGE})

    @coverage.validator
    def _validate_coverage(self, attribute, value):
        if not isinstance(value, bool):
            raise TypeError("coverage must be an boolean")
        if value:
            try:
                import coverage  # noqa: F401
            except ImportError:
                raise ImportError("The 'coverage' package must be installed.")

    cov_config: Union[str, None] = attr.ib(
        None, instance_of((type(None), str)), metadata={"help": HELP_COVERAGE_CONFIG}
    )
    cov_source: Union[str, Tuple[str]] = attr.ib(
        None, instance_of((type(None), tuple)), metadata={"help": HELP_COVERAGE_SOURCE}
    )

    cov_merge: Union[CoverageType, None] = attr.ib(
        None, metadata={"help": HELP_COVERAGE_MERGE}, hash=True
    )

    @cov_merge.validator
    def _validate_cov_merge(self, attribute, value):
        if value is None:
            return
        try:
            from coverage import Coverage
        except ImportError:
            raise ImportError("The 'coverage' package must be installed")
        if not isinstance(value, Coverage):
            raise TypeError("cov_merge must be an instance of coverage.Coverage")

    post_processors: tuple = attr.ib(
        ("coalesce_streams",), metadata={"help": HELP_POST_PROCS}
    )

    @post_processors.validator
    def _validate_post_processors(self, attribute, values):
        if not isinstance(values, tuple):
            raise TypeError(f"post_processors must be a tuple: {values}")
        for name in values:
            if name not in list_processor_names():
                raise TypeError(
                    f"name '{name}' not found in entry points: {list_processor_names()}"
                )

    process_resources: dict = attr.ib(
        attr.Factory(dict),
        instance_of(dict),
        metadata={"help": "Resources to parse to processor functions."},
    )

    diff_replace: tuple = attr.ib((), metadata={"help": HELP_DIFF_REPLACE})

    @diff_replace.validator
    def _validate_diff_replace(self, attribute, values):
        if not isinstance(values, tuple):
            raise TypeError(f"diff_replace must be a tuple: {values}")
        for i, args in enumerate(values):
            validate_regex_replace(args, i)

    diff_ignore: tuple = attr.ib(
        # TODO replace this default with a diff_replace one?
        ("/cells/*/outputs/*/traceback",),
        metadata={"help": HELP_DIFF_IGNORE},
    )

    @diff_ignore.validator
    def _validate_diff_ignore(self, attribute, values):
        if not isinstance(values, tuple):
            raise TypeError(f"diff_ignore must be a tuple: {values}")
        for item in values:
            if not isinstance(item, str):
                raise TypeError(f"diff_ignore item '{item}' must a strings")
            if not item.startswith("/"):
                raise ValueError(f"diff_ignore item '{item}' must start with '/'")

    diff_use_color: bool = attr.ib(
        True, instance_of(bool), metadata={"help": HELP_DIFF_USE_COLOR}
    )
    diff_color_words: bool = attr.ib(
        False, instance_of(bool), metadata={"help": HELP_DIFF_COLOR_WORDS}
    )

    force_regen: bool = attr.ib(
        False, instance_of(bool), metadata={"help": HELP_FORCE_REGEN}
    )

    def __setattr__(self, key, value):
        """Add validation when setting attributes."""
        x_attr = getattr(attr.fields(self.__class__), key)
        if x_attr.validator:
            x_attr.validator(self, x_attr, value)

        super(NBRegressionFixture, self).__setattr__(key, value)

    def check(
        self, path: Union[TextIO, str], raise_errors: bool = True
    ) -> NBRegressionResult:
        """Execute the Notebook and compare its initial vs. final contents.

        if ``force_regen`` is True, the new notebook will be written to ``path``

        if ``raise_errors`` is True:

        :raise nbconvert.preprocessors.CellExecutionError: if error in execution
        :raise NBConfigValidationError: if the notebook metadata is invalid
        :raise NBRegressionError: if diffs present

        :rtype: NBRegressionResult

        """
        __tracebackhide__ = True
        if hasattr(path, "name"):
            abspath = os.path.abspath(path.name)
        else:
            abspath = os.path.abspath(str(path))
        logger.debug(f"Checking file: {abspath}")

        nb_initial, nb_config = load_notebook_with_config(path)

        resources = copy.deepcopy(self.process_resources)
        if not self.exec_cwd:
            self.exec_cwd = os.path.dirname(abspath)

        if self.exec_notebook:
            logger.debug("Executing notebook.")
            exec_results = execute_notebook(
                nb_initial,
                resources=resources,
                cwd=self.exec_cwd,
                timeout=self.exec_timeout,
                allow_errors=self.exec_allow_errors,
                with_coverage=self.coverage,
                cov_config_file=self.cov_config,
                cov_source=self.cov_source,
            )
            exec_error = exec_results.exec_error
            nb_final = exec_results.notebook
            resources = exec_results.resources
        else:
            exec_error = None
            nb_final = nb_initial

        # TODO merge on fail option (using pytest-cov --no-cov-on-fail)
        if self.cov_merge and exec_results.has_coverage:
            logger.info("Merging coverage.")
            self.cov_merge.data.update(
                exec_results.coverage_data(self.cov_merge.debug),
                aliases=_get_coverage_aliases(self.cov_merge),
            )
            # we also take this opportunity to remove ''
            # from the unmatched source packages, which is caused by using `--cov=`
            self.cov_merge.source_pkgs_unmatched = [
                p for p in self.cov_merge.source_pkgs_unmatched if p
            ]

        for proc_name in self.post_processors:
            logger.debug(f"Applying post processor: {proc_name}")
            post_proc = load_processor(proc_name)
            nb_final, resources = post_proc(nb_final, resources)

        regex_replace = list(self.diff_replace) + list(nb_config.diff_replace)

        if regex_replace:
            logger.debug(f"Applying replacements: {regex_replace}")
            nb_initial_replace = regex_replace_nb(nb_initial, regex_replace)
            nb_final_replace = regex_replace_nb(nb_final, regex_replace)
        else:
            nb_initial_replace = nb_initial
            nb_final_replace = nb_final

        full_diff = diff_notebooks(nb_initial_replace, nb_final_replace)

        diff_ignore = copy.deepcopy(nb_config.diff_ignore)
        diff_ignore.update(self.diff_ignore)
        logger.debug(f"filtering diff by ignoring: {diff_ignore}")
        filtered_diff = filter_diff(full_diff, diff_ignore)

        diff_string = diff_to_string(
            nb_initial_replace,
            filtered_diff,
            use_color=self.diff_use_color,
            color_words=self.diff_color_words,
        )
        # TODO optionally write diff to file

        regen_exc = None
        if filtered_diff and self.force_regen and not exec_error:

            if hasattr(path, "close") and hasattr(path, "name"):
                path.close()
                with open(path.name, "w") as handle:
                    nbformat.write(nb_final, handle)
            else:
                nbformat.write(nb_final, str(path))

            regen_exc = NBRegressionError(
                f"Files differ and --nb-force-regen set, "
                f"regenerating file at:\n- {abspath}"
            )

        if not raise_errors:
            pass
        elif exec_error:
            print("Diff up to exception:\n" + diff_string, file=sys.stderr)
            raise exec_error
        elif regen_exc:
            print("Diff before regeneration:\n" + diff_string, file=sys.stderr)
            raise regen_exc
        elif filtered_diff:
            raise NBRegressionError(diff_string)

        return NBRegressionResult(
            nb_initial, nb_final, full_diff, filtered_diff, diff_string, resources
        )


def _get_coverage_aliases(cov):
    """Retrieve path aliases from coverage.Coverage object."""
    from coverage.files import PathAliases

    aliases = None
    if cov.config.paths:
        aliases = PathAliases()
        for paths in cov.config.paths.values():
            result = paths[0]
            for pattern in paths[1:]:
                aliases.add(pattern, result)
    return aliases

1	"""Jupyter Notebook Regression Test Class."""
2	import copy	2✔
3	import logging	2✔
4	import os	2✔
5	import sys	2✔
6	from typing import Any, List, TextIO, Tuple, Union	2✔
7
8	import attr	2✔
9	from attr.validators import instance_of	2✔
10	from nbdime.diff_format import DiffEntry	2✔
11	import nbformat	2✔
12	from nbformat import NotebookNode	2✔
13
14	try:	2✔
15	# coverage is an optional dependency
16	from coverage import Coverage as CoverageType	2✔
17	except ImportError:	×
18	CoverageType = Any	×
19
20	from pytest_notebook.diffing import diff_notebooks, diff_to_string, filter_diff	2✔
21	from pytest_notebook.execution import (	2✔
22	execute_notebook,
23	HELP_COVERAGE,
24	HELP_COVERAGE_CONFIG,
25	HELP_COVERAGE_SOURCE,
26	)
27	from pytest_notebook.notebook import (	2✔
28	load_notebook_with_config,
29	regex_replace_nb,
30	validate_regex_replace,
31	)
32	from pytest_notebook.post_processors import (	2✔
33	ENTRY_POINT_NAME,
34	list_processor_names,
35	load_processor,
36	)
37	from pytest_notebook.utils import autodoc	2✔
38
39	logger = logging.getLogger(__name__)	2✔
40
41	HELP_EXEC_NOTEBOOK = (	2✔
42	"Create a new notebook, by executing all cells in the original notebook"
43	)
44	HELP_EXEC_CWD = (	2✔
45	"Path to the directory which the notebook will run in "
46	"(defaults to directory of notebook)."
47	)
48	HELP_EXEC_TIMEOUT = "The maximum time to wait (in seconds) for execution of each cell."	2✔
49	HELP_EXEC_ALLOW_ERRORS = (	2✔
50	"Do not stop execution after the first unexpected exception "
51	"(where cell is not tagged ``raises-exception``)."
52	)
53	HELP_DIFF_REPLACE = (	2✔
54	"A list of regex replacements to apply before diffing, "
55	r"e.g. ``[('/cells/*/outputs', '\d{2,4}-\d{1,2}-\d{1,2}', 'DATE-STAMP')]``."
56	)
57	HELP_DIFF_IGNORE = (	2✔
58	"List of diff paths to ignore, e.g. '/cells/1/outputs' or '/cells/\\*/metadata'."
59	)
60	HELP_DIFF_USE_COLOR = "Use ANSI color code escapes for text output."	2✔
61	HELP_DIFF_COLOR_WORDS = "Highlight changed words using only colors."	2✔
62	HELP_FORCE_REGEN = (	2✔
63	"Re-generate notebook files, if no unexpected execution errors, "
64	"and an output path has been supplied."
65	)
66	HELP_POST_PROCS = (	2✔
67	"post-processors to apply to the new workbook, "
68	f"relating to entry points in the '{ENTRY_POINT_NAME}' group"
69	)
70	HELP_COVERAGE_MERGE = "A coverage.Coverage instance, to merge coverage results with."	2✔
71
72
73	class NBRegressionError(Exception):	2✔
74	"""Exception to signal a regression test fail."""
75
76
77	@autodoc	2✔
78	@attr.s(frozen=True, slots=True, repr=False)	2✔
79	class NBRegressionResult:
80	"""A class to store the result of ``NBRegressionFixture.check``."""
81
82	nb_initial: NotebookNode = attr.ib(	2✔
83	validator=instance_of(NotebookNode), metadata={"help": "Initial notebook."}
84	)
85	nb_final: NotebookNode = attr.ib(	2✔
86	validator=instance_of(NotebookNode),
87	metadata={"help": "Notebook after execution and post-processing."},
88	)
89
90	diff_full: List[DiffEntry] = attr.ib(	2✔
91	metadata={"help": "Full diff of initial/final notebooks."}
92	)
93	diff_filtered: List[DiffEntry] = attr.ib(	2✔
94	metadata={
95	"help": (
96	"Diff of initial/final notebooks, "
97	"filtered according to the parsed configuration."
98	)
99	}
100	)
101	diff_string: str = attr.ib(	2✔
102	validator=instance_of(str),
103	metadata={"help": "The formatte string of diff_filtered."},
104	)
105
106	process_resources: dict = attr.ib(	2✔
107	attr.Factory(dict),
108	instance_of(dict),
109	metadata={"help": "Resources returned from notebook processors."},
110	)
111
112	def __repr__(self):	2✔
113	"""Represent the class instance."""
114	return (	×
115	f"NBRegressionResult(diff_full_length={len(self.diff_full)},"
116	f"diff_filtered_length={len(self.diff_filtered)})"
117	)
118
119
120	@autodoc	2✔
121	@attr.s	2✔
122	class NBRegressionFixture:
123	"""Class to perform Jupyter Notebook Regression tests."""
124
125	exec_notebook: bool = attr.ib(	2✔
126	True, instance_of(bool), metadata={"help": HELP_EXEC_NOTEBOOK}
127	)
128	exec_cwd: Union[str, None] = attr.ib(	2✔
129	None, instance_of((type(None), str)), metadata={"help": HELP_EXEC_CWD}
130	)
131
132	@exec_cwd.validator	2✔
133	def _validate_exec_cwd(self, attribute, value):
134	if value is None:	2✔
135	return	2✔
136	if not isinstance(value, str):	2✔
137	raise TypeError("exec_cwd must be None or a string")	×
138	if not os.path.isdir(value):	2✔
139	raise IOError("exec_cwd='{}' is not an existing directory".format(value))	×
140
141	exec_allow_errors: bool = attr.ib(	2✔
142	False, instance_of(bool), metadata={"help": HELP_EXEC_ALLOW_ERRORS}
143	)
144	exec_timeout: int = attr.ib(	2✔
145	120, instance_of((int, float)), metadata={"help": HELP_EXEC_TIMEOUT}
146	)
147
148	@exec_timeout.validator	2✔
149	def _validate_exec_timeout(self, attribute, value):
150	if not isinstance(value, int):	2✔
151	raise TypeError("exec_timeout must be an integer")	×
152	if value <= 0:	2✔
153	raise ValueError("exec_timeout must be larger than 0")	×
154
155	coverage: bool = attr.ib(False, metadata={"help": HELP_COVERAGE})	2✔
156
157	@coverage.validator	2✔
158	def _validate_coverage(self, attribute, value):
159	if not isinstance(value, bool):	2✔
160	raise TypeError("coverage must be an boolean")	×
161	if value:	2✔
162	try:	2✔
163	import coverage # noqa: F401	2✔
164	except ImportError:	×
165	raise ImportError("The 'coverage' package must be installed.")	×
166
167	cov_config: Union[str, None] = attr.ib(	2✔
168	None, instance_of((type(None), str)), metadata={"help": HELP_COVERAGE_CONFIG}
169	)
170	cov_source: Union[str, Tuple[str]] = attr.ib(	2✔
171	None, instance_of((type(None), tuple)), metadata={"help": HELP_COVERAGE_SOURCE}
172	)
173
174	cov_merge: Union[CoverageType, None] = attr.ib(	2✔
175	None, metadata={"help": HELP_COVERAGE_MERGE}, hash=True
176	)
177
178	@cov_merge.validator	2✔
179	def _validate_cov_merge(self, attribute, value):
180	if value is None:	2✔
181	return	2✔
182	try:	×
183	from coverage import Coverage	×
184	except ImportError:	×
185	raise ImportError("The 'coverage' package must be installed")	×
186	if not isinstance(value, Coverage):	×
187	raise TypeError("cov_merge must be an instance of coverage.Coverage")	×
188
189	post_processors: tuple = attr.ib(	2✔
190	("coalesce_streams",), metadata={"help": HELP_POST_PROCS}
191	)
192
193	@post_processors.validator	2✔
194	def _validate_post_processors(self, attribute, values):
195	if not isinstance(values, tuple):	2✔
196	raise TypeError(f"post_processors must be a tuple: {values}")	×
197	for name in values:	2✔
198	if name not in list_processor_names():	2✔
199	raise TypeError(	×
200	f"name '{name}' not found in entry points: {list_processor_names()}"
201	)
202
203	process_resources: dict = attr.ib(	2✔
204	attr.Factory(dict),
205	instance_of(dict),
206	metadata={"help": "Resources to parse to processor functions."},
207	)
208
209	diff_replace: tuple = attr.ib((), metadata={"help": HELP_DIFF_REPLACE})	2✔
210
211	@diff_replace.validator	2✔
212	def _validate_diff_replace(self, attribute, values):
213	if not isinstance(values, tuple):	2✔
214	raise TypeError(f"diff_replace must be a tuple: {values}")	×
215	for i, args in enumerate(values):	2✔
216	validate_regex_replace(args, i)	2✔
217
218	diff_ignore: tuple = attr.ib(	2✔
219	# TODO replace this default with a diff_replace one?
220	("/cells//outputs//traceback",),
221	metadata={"help": HELP_DIFF_IGNORE},
222	)
223
224	@diff_ignore.validator	2✔
225	def _validate_diff_ignore(self, attribute, values):
226	if not isinstance(values, tuple):	2✔
227	raise TypeError(f"diff_ignore must be a tuple: {values}")	×
228	for item in values:	2✔
229	if not isinstance(item, str):	2✔
230	raise TypeError(f"diff_ignore item '{item}' must a strings")	×
231	if not item.startswith("/"):	2✔
232	raise ValueError(f"diff_ignore item '{item}' must start with '/'")	×
233
234	diff_use_color: bool = attr.ib(	2✔
235	True, instance_of(bool), metadata={"help": HELP_DIFF_USE_COLOR}
236	)
237	diff_color_words: bool = attr.ib(	2✔
238	False, instance_of(bool), metadata={"help": HELP_DIFF_COLOR_WORDS}
239	)
240
241	force_regen: bool = attr.ib(	2✔
242	False, instance_of(bool), metadata={"help": HELP_FORCE_REGEN}
243	)
244
245	def __setattr__(self, key, value):	2✔
246	"""Add validation when setting attributes."""
247	x_attr = getattr(attr.fields(self.__class__), key)	2✔
248	if x_attr.validator:	2✔
249	x_attr.validator(self, x_attr, value)	2✔
250
251	super(NBRegressionFixture, self).__setattr__(key, value)	2✔
252
253	def check(	2✔
254	self, path: Union[TextIO, str], raise_errors: bool = True
255	) -> NBRegressionResult:
256	"""Execute the Notebook and compare its initial vs. final contents.
257
258	if ``force_regen`` is True, the new notebook will be written to ``path``
259
260	if ``raise_errors`` is True:
261
262	:raise nbconvert.preprocessors.CellExecutionError: if error in execution
263	:raise NBConfigValidationError: if the notebook metadata is invalid
264	:raise NBRegressionError: if diffs present
265
266	:rtype: NBRegressionResult
267
268	"""
269	__tracebackhide__ = True	2✔
270	if hasattr(path, "name"):	2✔
271	abspath = os.path.abspath(path.name)	×
272	else:
273	abspath = os.path.abspath(str(path))	2✔
274	logger.debug(f"Checking file: {abspath}")	2✔
275
276	nb_initial, nb_config = load_notebook_with_config(path)	2✔
277
278	resources = copy.deepcopy(self.process_resources)	2✔
279	if not self.exec_cwd:	2✔
280	self.exec_cwd = os.path.dirname(abspath)	2✔
281
282	if self.exec_notebook:	2✔
283	logger.debug("Executing notebook.")	2✔
284	exec_results = execute_notebook(	2✔
285	nb_initial,
286	resources=resources,
287	cwd=self.exec_cwd,
288	timeout=self.exec_timeout,
289	allow_errors=self.exec_allow_errors,
290	with_coverage=self.coverage,
291	cov_config_file=self.cov_config,
292	cov_source=self.cov_source,
293	)
294	exec_error = exec_results.exec_error	2✔
295	nb_final = exec_results.notebook	2✔
296	resources = exec_results.resources	2✔
297	else:
298	exec_error = None	×
299	nb_final = nb_initial	×
300
301	# TODO merge on fail option (using pytest-cov --no-cov-on-fail)
302	if self.cov_merge and exec_results.has_coverage:	2✔
303	logger.info("Merging coverage.")	×
304	self.cov_merge.data.update(	×
305	exec_results.coverage_data(self.cov_merge.debug),
306	aliases=_get_coverage_aliases(self.cov_merge),
307	)
308	# we also take this opportunity to remove ''
309	# from the unmatched source packages, which is caused by using `--cov=`
310	self.cov_merge.source_pkgs_unmatched = [	×
311	p for p in self.cov_merge.source_pkgs_unmatched if p
312	]
313
314	for proc_name in self.post_processors:	2✔
315	logger.debug(f"Applying post processor: {proc_name}")	2✔
316	post_proc = load_processor(proc_name)	2✔
317	nb_final, resources = post_proc(nb_final, resources)	2✔
318
319	regex_replace = list(self.diff_replace) + list(nb_config.diff_replace)	2✔
320
321	if regex_replace:	2✔
322	logger.debug(f"Applying replacements: {regex_replace}")	2✔
323	nb_initial_replace = regex_replace_nb(nb_initial, regex_replace)	2✔
324	nb_final_replace = regex_replace_nb(nb_final, regex_replace)	2✔
325	else:
326	nb_initial_replace = nb_initial	2✔
327	nb_final_replace = nb_final	2✔
328
329	full_diff = diff_notebooks(nb_initial_replace, nb_final_replace)	2✔
330
331	diff_ignore = copy.deepcopy(nb_config.diff_ignore)	2✔
332	diff_ignore.update(self.diff_ignore)	2✔
333	logger.debug(f"filtering diff by ignoring: {diff_ignore}")	2✔
334	filtered_diff = filter_diff(full_diff, diff_ignore)	2✔
335
336	diff_string = diff_to_string(	2✔
337	nb_initial_replace,
338	filtered_diff,
339	use_color=self.diff_use_color,
340	color_words=self.diff_color_words,
341	)
342	# TODO optionally write diff to file
343
344	regen_exc = None	2✔
345	if filtered_diff and self.force_regen and not exec_error:	2✔
346
347	if hasattr(path, "close") and hasattr(path, "name"):	2✔
348	path.close()	×
349	with open(path.name, "w") as handle:	×
350	nbformat.write(nb_final, handle)	×
351	else:
352	nbformat.write(nb_final, str(path))	2✔
353
354	regen_exc = NBRegressionError(	2✔
355	f"Files differ and --nb-force-regen set, "
356	f"regenerating file at:\n- {abspath}"
357	)
358
359	if not raise_errors:	2✔
360	pass	×
361	elif exec_error:	2✔
362	print("Diff up to exception:\n" + diff_string, file=sys.stderr)	2✔
363	raise exec_error	2✔
364	elif regen_exc:	2✔
365	print("Diff before regeneration:\n" + diff_string, file=sys.stderr)	2✔
366	raise regen_exc	2✔
367	elif filtered_diff:	2✔
368	raise NBRegressionError(diff_string)	2✔
369
370	return NBRegressionResult(	2✔
371	nb_initial, nb_final, full_diff, filtered_diff, diff_string, resources
372	)
373
374
375	def _get_coverage_aliases(cov):	2✔
376	"""Retrieve path aliases from coverage.Coverage object."""
377	from coverage.files import PathAliases	×
378
379	aliases = None	×
380	if cov.config.paths:	×
381	aliases = PathAliases()	×
382	for paths in cov.config.paths.values():	×
383	result = paths[0]	×
384	for pattern in paths[1:]:	×
385	aliases.add(pattern, result)	×
386	return aliases	×

chrisjsewell / pytest-notebook / 104

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