feat: simplify purges_files

Also, move tests to test_data.py, and finish covering the code.

Author: nedbat

CHANGES.rst

@@ -20,6 +20,9 @@ development at the same time, such as 4.5.x and 5.0.
 Unreleased
 ----------
 
+- Added a :meth:`.CoverageData.purge_files` method to remove recorded data for
+  a particular file.  Contributed by `Stephan Deibel <pull 1547_>`_.
+
 - Fix: In some embedded environments, an IndexError could occur on stop() when
   the originating thread exits before completion.  This is now fixed, thanks to
   `Russell Keith-Magee <pull 1543_>`_, closing `issue 1542`_.
@@ -29,6 +32,7 @@ Unreleased
 
 .. _issue 1542: https://github.com/nedbat/coveragepy/issues/1542
 .. _pull 1543: https://github.com/nedbat/coveragepy/pull/1543
+.. _pull 1547: https://github.com/nedbat/coveragepy/pull/1547
 .. _pull 1550: https://github.com/nedbat/coveragepy/pull/1550
 
 

CONTRIBUTORS.txt

@@ -149,6 +149,7 @@ Sigve Tjora
 Simon Willison
 Stan Hu
 Stefan Behnel
+Stephan Deibel
 Stephan Richter
 Stephen Finucane
 Steve Dower

coverage/sqldata.py

@@ -197,9 +197,11 @@ class CoverageData(AutoReprMixin):
 
     Write the data to its file with :meth:`write`.
 
-    You can clear the data in memory with :meth:`erase`.  Two data collections
-    can be combined by using :meth:`update` on one :class:`CoverageData`,
-    passing it the other.
+    You can clear the data in memory with :meth:`erase`.  Data for specific
+    files can be removed from the database with :meth:`purge_files`.
+
+    Two data collections can be combined by using :meth:`update` on one
+    :class:`CoverageData`, passing it the other.
 
     Data in a :class:`CoverageData` can be serialized and deserialized with
     :meth:`dumps` and :meth:`loads`.
@@ -615,41 +617,29 @@ def touch_files(self, filenames: Collection[str], plugin_name: Optional[str] = N
                     # 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:
+    def purge_files(self, filenames: Collection[str]) -> None:
         """Purge any existing coverage data for the given `filenames`.
 
-        If `context` is given, purge only data associated with that measurement context.
-        """
+        .. versionadded:: 7.2
 
+        """
         if self._debug.should("dataop"):
-            self._debug.write(f"Purging {filenames!r} for context {context}")
+            self._debug.write(f"Purging data for {filenames!r}")
         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'
+                sql = "delete from line_bits where file_id=?"
             elif self._has_arcs:
-                table = 'arcs'
+                sql = "delete from arc where file_id=?"
             else:
-                return
+                raise DataError("Can't purge files in an empty CoverageData")
 
             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)
+                con.execute_void(sql, (file_id,))
 
     def update(self, other_data: CoverageData, aliases: Optional[PathAliases] = None) -> None:
         """Update this data with data from several other :class:`CoverageData` instances.
@@ -868,7 +858,12 @@ def has_arcs(self) -> bool:
         return bool(self._has_arcs)
 
     def measured_files(self) -> Set[str]:
-        """A set of all files that had been measured."""
+        """A set of all files that have been measured.
+
+        Note that a file may be mentioned as measured even though no lines or
+        arcs for that file are present in the data.
+
+        """
         return set(self._file_map)
 
     def measured_contexts(self) -> Set[str]:

tests/test_api.py

@@ -754,112 +754,6 @@ def test_run_debug_sys(self) -> None:
         cov.stop()                      # pragma: nested
         assert cast(str, d['data_file']).endswith(".coverage")
 
-    def test_purge_filenames(self) -> None:
-
-        fn1 = self.make_file("mymain.py", """\
-            import mymod
-            a = 1
-            """)
-        fn1 = os.path.join(self.temp_dir, fn1)
-
-        fn2 = self.make_file("mymod.py", """\
-            fooey = 17
-            """)
-        fn2 = os.path.join(self.temp_dir, fn2)
-
-        cov = coverage.Coverage()
-        self.start_import_stop(cov, "mymain")
-
-        data = cov.get_data()
-
-        # Initial measurement was for two files
-        assert len(data.measured_files()) == 2
-        assert [1, 2] == sorted_lines(data, fn1)
-        assert [1,] == sorted_lines(data, fn2)
-
-        # Purge one file's data and one should remain
-        data.purge_files([fn1])
-        assert len(data.measured_files()) == 1
-        assert [] == sorted_lines(data, fn1)
-        assert [1,] == sorted_lines(data, fn2)
-
-        # Purge second file's data and none should remain
-        data.purge_files([fn2])
-        assert len(data.measured_files()) == 0
-        assert [] == sorted_lines(data, fn1)
-        assert [] == sorted_lines(data, fn2)
-
-    def test_purge_filenames_context(self) -> None:
-
-        fn1 = self.make_file("mymain.py", """\
-            import mymod
-            a = 1
-            """)
-        fn1 = os.path.join(self.temp_dir, fn1)
-
-        fn2 = self.make_file("mymod.py", """\
-            fooey = 17
-            """)
-        fn2 = os.path.join(self.temp_dir, fn2)
-
-        def dummy_function() -> None:
-            unused = 42
-
-        # Start/stop since otherwise cantext
-        cov = coverage.Coverage()
-        cov.start()
-        cov.switch_context('initialcontext')
-        dummy_function()
-        cov.switch_context('testcontext')
-        cov.stop()
-        self.start_import_stop(cov, "mymain")
-
-        data = cov.get_data()
-
-        # Initial measurement was for three files and two contexts
-        assert len(data.measured_files()) == 3
-        assert [1, 2] == sorted_lines(data, fn1)
-        assert [1,] == sorted_lines(data, fn2)
-        assert len(sorted_lines(data, __file__)) == 1
-        assert len(data.measured_contexts()) == 2
-
-        # Remove specifying wrong context should raise exception and not remove anything
-        try:
-            data.purge_files([fn1], 'wrongcontext')
-        except coverage.sqldata.DataError:
-            pass
-        else:
-            assert 0, "exception expected"
-        assert len(data.measured_files()) == 3
-        assert [1, 2] == sorted_lines(data, fn1)
-        assert [1,] == sorted_lines(data, fn2)
-        assert len(sorted_lines(data, __file__)) == 1
-        assert len(data.measured_contexts()) == 2
-
-        # Remove one file specifying correct context
-        data.purge_files([fn1], 'testcontext')
-        assert len(data.measured_files()) == 2
-        assert [] == sorted_lines(data, fn1)
-        assert [1,] == sorted_lines(data, fn2)
-        assert len(sorted_lines(data, __file__)) == 1
-        assert len(data.measured_contexts()) == 2
-
-        # Remove second file with other correct context
-        data.purge_files([__file__], 'initialcontext')
-        assert len(data.measured_files()) == 1
-        assert [] == sorted_lines(data, fn1)
-        assert [1,] == sorted_lines(data, fn2)
-        assert len(sorted_lines(data, __file__)) == 0
-        assert len(data.measured_contexts()) == 2
-
-        # Remove last file specifying correct context
-        data.purge_files([fn2], 'testcontext')
-        assert len(data.measured_files()) == 0
-        assert [] == sorted_lines(data, fn1)
-        assert [] == sorted_lines(data, fn2)
-        assert len(sorted_lines(data, __file__)) == 0
-        assert len(data.measured_contexts()) == 2
-
 
 class CurrentInstanceTest(CoverageTest):
     """Tests of Coverage.current()."""

tests/test_data.py

@@ -588,6 +588,35 @@ def thread_main() -> None:
         assert_lines1_data(covdata)
         assert not exceptions
 
+    def test_purge_files_lines(self) -> None:
+        covdata = DebugCoverageData()
+        covdata.add_lines(LINES_1)
+        covdata.add_lines(LINES_2)
+        assert_line_counts(covdata, SUMMARY_1_2)
+        covdata.purge_files(["a.py", "b.py"])
+        assert_line_counts(covdata, {"a.py": 0, "b.py": 0, "c.py": 1})
+        covdata.purge_files(["c.py"])
+        assert_line_counts(covdata, {"a.py": 0, "b.py": 0, "c.py": 0})
+        # It's OK to "purge" a file that wasn't measured.
+        covdata.purge_files(["xyz.py"])
+        assert_line_counts(covdata, {"a.py": 0, "b.py": 0, "c.py": 0})
+
+    def test_purge_files_arcs(self) -> None:
+        covdata = CoverageData()
+        covdata.add_arcs(ARCS_3)
+        covdata.add_arcs(ARCS_4)
+        assert_line_counts(covdata, SUMMARY_3_4)
+        covdata.purge_files(["x.py", "y.py"])
+        assert_line_counts(covdata, {"x.py": 0, "y.py": 0, "z.py": 1})
+        covdata.purge_files(["z.py"])
+        assert_line_counts(covdata, {"x.py": 0, "y.py": 0, "z.py": 0})
+
+    def test_cant_purge_in_empty_data(self) -> None:
+        covdata = DebugCoverageData()
+        msg = "Can't purge files in an empty CoverageData"
+        with pytest.raises(DataError, match=msg):
+            covdata.purge_files(["abc.py"])
+
 
 class CoverageDataInTempDirTest(CoverageTest):
     """Tests of CoverageData that need a temporary directory to make files."""