fix: really close all sqlite dbs #2017

Author: nedbat

CHANGES.rst

@@ -23,7 +23,10 @@ upgrading your version of coverage.py.
 Unreleased
 ----------
 
-Nothing yet.
+- Fix: really close all SQLite databases, even in-memory ones. Closes `issue
+  2017`_.
+
+.. _issue 2017: https://github.com/nedbat/coveragepy/issues/2017
 
 
 .. start-releases

coverage/control.py

@@ -282,6 +282,7 @@ def __init__(                       # pylint: disable=too-many-arguments
         self._plugins: Plugins = Plugins()
         self._plugin_override = cast(Union[Iterable[TCoverageInit], None], plugins)
         self._data: CoverageData | None = None
+        self._data_to_close: list[CoverageData] = []
         self._core: Core | None = None
         self._collector: Collector | None = None
         self._metacov = False
@@ -646,6 +647,7 @@ def _init_data(self, suffix: str | bool | None) -> None:
                 debug=self._debug,
                 no_disk=self._no_disk,
             )
+            self._data_to_close.append(self._data)
 
     def start(self) -> None:
         """Start measuring code coverage.
@@ -718,6 +720,8 @@ def _atexit(self, event: str = "atexit") -> None:
             self.stop()
         if self._auto_save or event == "sigterm":
             self.save()
+        for d in self._data_to_close:
+            d.close(force=True)
 
     def _on_sigterm(self, signum_unused: int, frame_unused: FrameType | None) -> None:
         """A handler for signal.SIGTERM."""
@@ -1048,6 +1052,7 @@ def _prepare_data_for_reporting(self) -> None:
             if self._data is not None:
                 mapped_data.update(self._data, map_path=self._make_aliases().map)
             self._data = mapped_data
+            self._data_to_close.append(mapped_data)
 
     def report(
         self,

coverage/sqldata.py

@@ -271,13 +271,19 @@ def _choose_filename(self) -> None:
     def _reset(self) -> None:
         """Reset our attributes."""
         if not self._no_disk:
-            for db in self._dbs.values():
-                db.close()
-            self._dbs = {}
+            self.close()
         self._file_map = {}
         self._have_used = False
         self._current_context_id = None
 
+    def close(self, force: bool = False) -> None:
+        """Really close all the database objects."""
+        if self._debug.should("dataio"):
+            self._debug.write(f"Closing dbs, force={force}: {self._dbs}")
+        for db in self._dbs.values():
+            db.close(force=force)
+        self._dbs = {}
+
     def _open_db(self) -> None:
         """Open an existing db file, and read its metadata."""
         if self._debug.should("dataio"):

coverage/sqlitedb.py

@@ -76,13 +76,14 @@ def _connect(self) -> None:
         # to keep things going.
         self.execute_void("pragma synchronous=off", fail_ok=True)
 
-    def close(self) -> None:
+    def close(self, force: bool = False) -> None:
         """If needed, close the connection."""
-        if self.con is not None and self.filename != ":memory:":
-            if self.debug.should("sql"):
-                self.debug.write(f"Closing {self.con!r} on {self.filename!r}")
-            self.con.close()
-            self.con = None
+        if self.con is not None:
+            if force or self.filename != ":memory:":
+                if self.debug.should("sql"):
+                    self.debug.write(f"Closing {self.con!r} on {self.filename!r}")
+                self.con.close()
+                self.con = None
 
     def __enter__(self) -> SqliteDb:
         if self.nest == 0:

tests/test_data.py

@@ -670,7 +670,7 @@ def test_error_when_closing(self, klass: TCoverageData) -> None:
             covdata.add_lines(LINES_1)
             # I don't know how to make a real error, so let's fake one.
             sqldb = list(covdata._dbs.values())[0]
-            sqldb.close = lambda: 1/0       # type: ignore[assignment]
+            sqldb.close = lambda: 1/0       # type: ignore
             covdata.add_lines(LINES_1)
 
     def test_wrong_schema_version(self) -> None:
@@ -724,8 +724,11 @@ def test_debug_output_with_debug_option(self) -> None:
         covdata2.read()
         assert_line_counts(covdata2, SUMMARY_1)
 
+        print(debug.get_output())
+
         assert re.search(
-            r"^Erasing data file '.*\.coverage'\n" +
+            r"^Closing dbs, force=False: {}\n" +
+            r"Erasing data file '.*\.coverage'\n" +
             r"Opening data file '.*\.coverage'\n" +
             r"Initing data file '.*\.coverage'\n" +
             r"Opening data file '.*\.coverage'\n$",