From a4cd6a01f766359fbe8ff159e0ed87e03e8023ed Mon Sep 17 00:00:00 2001 From: Wingware Development <59840981+wingware-dev@users.noreply.github.com> Date: Sat, 4 Feb 2023 18:20:20 -0500 Subject: feat: add purge_files method to CoverageData + unit tests for it (#1547) * Add purge_files method to CoverageData, to allow for selective removal and update of coverage data. * Fix assert syntax so it's not true; this code isn't reached in the test unless it fails and then it would have failed to fail. * Remove trailing whitespace; did not expect this would matter on a blank line. * Add type annotations required by mypy --------- Co-authored-by: Stephan Deibel --- coverage/sqldata.py | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) (limited to 'coverage/sqldata.py') diff --git a/coverage/sqldata.py b/coverage/sqldata.py index 77577437..12676d0b 100644 --- a/coverage/sqldata.py +++ b/coverage/sqldata.py @@ -615,6 +615,42 @@ class CoverageData(AutoReprMixin): # Set the tracer for this file self.add_file_tracers({filename: plugin_name}) + def purge_files(self, filenames: Iterable[str], context: Optional[str] = None) -> None: + """Purge any existing coverage data for the given `filenames`. + + If `context` is given, purge only data associated with that measurement context. + """ + + if self._debug.should("dataop"): + self._debug.write(f"Purging {filenames!r} for context {context}") + self._start_using() + with self._connect() as con: + + if context is not None: + context_id = self._context_id(context) + if context_id is None: + raise DataError("Unknown context {context}") + else: + context_id = None + + if self._has_lines: + table = 'line_bits' + elif self._has_arcs: + table = 'arcs' + else: + return + + for filename in filenames: + file_id = self._file_id(filename, add=False) + if file_id is None: + continue + self._file_map.pop(filename, None) + if context_id is None: + q = f'delete from {table} where file_id={file_id}' + else: + q = f'delete from {table} where file_id={file_id} and context_id={context_id}' + con.execute(q) + def update(self, other_data: CoverageData, aliases: Optional[PathAliases] = None) -> None: """Update this data with data from several other :class:`CoverageData` instances. -- cgit v1.2.1