Tests of the schema checking

4 files changed, 40 insertions, 18 deletions

diff --git a/coverage/sqldata.py b/coverage/sqldata.py
index e102f294..0037d76d 100644
--- a/coverage/sqldata.py
+++ b/coverage/sqldata.py
@@ -4,7 +4,7 @@
 """Sqlite coverage data."""
 
 # TODO: get sys_info for data class, so we can see sqlite version etc
-# TODO: get rid of skip_unless_data_storage_is_json
+# TODO: get rid of skip_unless_data_storage_is
 # TODO: get rid of "JSON message" and "SQL message" in the tests
 # TODO: factor out dataop debugging to a wrapper class?
 # TODO: make sure all dataop debugging is in place somehow
@@ -107,7 +107,7 @@ class CoverageSqliteData(SimpleRepr):
         with self._db:
             try:
                 schema_version, = self._db.execute("select version from coverage_schema").fetchone()
-            except (TypeError, sqlite3.Error) as exc:
+            except Exception as exc:
                 raise CoverageException(
                     "Data file {!r} doesn't seem to be a coverage data file: {}".format(
                         self.filename, exc
@@ -116,7 +116,7 @@ class CoverageSqliteData(SimpleRepr):
             else:
                 if schema_version != SCHEMA_VERSION:
                     raise CoverageException(
-                        "Data file {!r} is wrong schema: {} instead of {}".format(
+                        "Couldn't use data file {!r}: wrong schema: {} instead of {}".format(
                             self.filename, schema_version, SCHEMA_VERSION
                         )
                     )
diff --git a/tests/coveragetest.py b/tests/coveragetest.py
index 6e308718..15c61ece 100644
--- a/tests/coveragetest.py
+++ b/tests/coveragetest.py
@@ -97,9 +97,9 @@ class CoverageTest(
         self.last_command_output = None
         self.last_module_name = None
 
-    def skip_unless_data_storage_is_json(self):
-        if STORAGE != "json":
-            self.skipTest("Not using JSON for data storage")
+    def skip_unless_data_storage_is(self, storage):
+        if STORAGE != storage:
+            self.skipTest("Not using {} for data storage".format(storage))
 
     def clean_local_file_imports(self):
         """Clean up the results of calls to `import_local_file`.
diff --git a/tests/test_data.py b/tests/test_data.py
index 1e6ce027..15da32e5 100644
--- a/tests/test_data.py
+++ b/tests/test_data.py
@@ -8,6 +8,7 @@ import json
 import os
 import os.path
 import re
+import sqlite3
 
 import mock
 
@@ -190,7 +191,7 @@ class CoverageDataTest(DataTestHelpers, CoverageTest):
         self.assertIsNone(covdata.lines('no_such_file.py'))
 
     def test_run_info(self):
-        self.skip_unless_data_storage_is_json()
+        self.skip_unless_data_storage_is("json")
         covdata = CoverageData()
         self.assertEqual(covdata.run_infos(), [])
         covdata.add_run_info(hello="there")
@@ -269,7 +270,7 @@ class CoverageDataTest(DataTestHelpers, CoverageTest):
         self.assertEqual(covdata3.run_infos(), [])
 
     def test_update_run_info(self):
-        self.skip_unless_data_storage_is_json()
+        self.skip_unless_data_storage_is("json")
         covdata1 = CoverageData()
         covdata1.add_arcs(ARCS_3)
         covdata1.add_run_info(hello="there", count=17)
@@ -475,15 +476,36 @@ class CoverageDataTestInTempDir(DataTestHelpers, CoverageTest):
             covdata.read()
         self.assertFalse(covdata)
 
-        if STORAGE == "json":
-            self.make_file("misleading.dat", CoverageData._GO_AWAY + " this isn't JSON")
-            with self.assertRaisesRegex(CoverageException, msg.format("misleading.dat")):
-                covdata = CoverageData("misleading.dat")
-                covdata.read()
-            self.assertFalse(covdata)
+    def test_read_json_errors(self):
+        self.skip_unless_data_storage_is("json")
+        self.make_file("misleading.dat", CoverageData._GO_AWAY + " this isn't JSON")
+        msg = r"Couldn't .* '.*[/\\]{0}': \S+"
+        with self.assertRaisesRegex(CoverageException, msg.format("misleading.dat")):
+            covdata = CoverageData("misleading.dat")
+            covdata.read()
+        self.assertFalse(covdata)
+
+    def test_read_sql_errors(self):
+        self.skip_unless_data_storage_is("sql")
+        with sqlite3.connect("wrong_schema.db") as con:
+            con.execute("create table coverage_schema (version integer)")
+            con.execute("insert into coverage_schema (version) values (99)")
+        msg = r"Couldn't .* '.*[/\\]{0}': wrong schema: 99 instead of \d+".format("wrong_schema.db")
+        with self.assertRaisesRegex(CoverageException, msg):
+            covdata = CoverageData("wrong_schema.db")
+            covdata.read()
+        self.assertFalse(covdata)
+
+        with sqlite3.connect("no_schema.db") as con:
+            con.execute("create table foobar (baz text)")
+        msg = r"Couldn't .* '.*[/\\]{0}': \S+".format("no_schema.db")
+        with self.assertRaisesRegex(CoverageException, msg):
+            covdata = CoverageData("no_schema.db")
+            covdata.read()
+        self.assertFalse(covdata)
 
     def test_debug_main(self):
-        self.skip_unless_data_storage_is_json()
+        self.skip_unless_data_storage_is("json")
         covdata1 = CoverageData(".coverage")
         covdata1.add_lines(LINES_1)
         covdata1.write()
@@ -660,7 +682,7 @@ class CoverageDataFilesTest(DataTestHelpers, CoverageTest):
 
     def read_json_data_file(self, fname):
         """Read a JSON data file for testing the JSON directly."""
-        self.skip_unless_data_storage_is_json()
+        self.skip_unless_data_storage_is("json")
         with open(fname, 'r') as fdata:
             go_away = fdata.read(len(CoverageData._GO_AWAY))
             self.assertEqual(go_away, CoverageData._GO_AWAY)
diff --git a/tests/test_process.py b/tests/test_process.py
index 49919b0f..1ac18ffe 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -516,7 +516,7 @@ class ProcessTest(CoverageTest):
         if not hasattr(os, 'fork'):
             self.skipTest("Can't test os.fork since it doesn't exist.")
         # See https://nedbatchelder.com/blog/201808/sqlite_data_storage_for_coveragepy.html
-        self.skip_unless_data_storage_is_json()
+        self.skip_unless_data_storage_is("json")
 
         self.make_file("fork.py", """\
             import os
@@ -655,7 +655,7 @@ class ProcessTest(CoverageTest):
         if env.PYPY and env.PY3 and env.PYPYVERSION[:3] == (5, 10, 0):      # pragma: obscure
             # https://bitbucket.org/pypy/pypy/issues/2729/pypy3-510-incorrectly-decodes-astral-plane
             self.skipTest("Avoid incorrect decoding astral plane JSON chars")
-        self.skip_unless_data_storage_is_json()
+        self.skip_unless_data_storage_is("json")
 
         self.make_file(".coveragerc", """\
             [run]