LocalConnectionFactory minor cleanup and more tests. #319 #375

Author: lemon24

src/reader/_storage/_sqlite_utils.py

@@ -412,6 +412,9 @@ class UsageError(DBError):
     display_name = "usage error"
 
 
+TEMPORARY_DB_PATHS = {':memory:', ''}
+
+
 class LocalConnectionFactory:
     """Maintain a set of connections to the same database, one per thread.
 
@@ -450,8 +453,6 @@ def __init__(
         self.kwargs = kwargs
         if kwargs.get('uri'):  # pragma: no cover
             raise NotImplementedError("is_private() does not work for uri=True")
-        if path.startswith(':file:'):  # pragma: no cover
-            raise NotImplementedError("file: connections are not supported")
         self.kwargs['uri'] = True
         self.attached: dict[str, str] = {}
         self._local = _LocalConnectionFactoryState()
@@ -595,25 +596,27 @@ def attach(self, name: str, path: str) -> None:
         if name in self.attached:  # pragma: no cover
             raise ValueError(f"database already attached: {name!r}")
 
-        uri_path = self._make_uri(path)
-        self.attached[name] = uri_path
+        self.attached[name] = path
         db = self._local.db
         assert db is not None
-        self._attach(db, name, uri_path)
+        self._attach(db, name, path)
 
     def _attach(self, db: sqlite3.Connection, name: str, path: str) -> None:
-        db.execute("ATTACH DATABASE ? AS ?;", (path, name))
+        db.execute("ATTACH DATABASE ? AS ?;", (self._make_uri(path), name))
 
     def is_private(self) -> bool:
         return self._is_private(self.path)
 
     def _make_uri(self, path: str) -> str:
-        # keeping sqlite special names intact
-        if path in (':memory:', ''):
-            return path
-        # make the path full, otherwise it cannot be turned into URI
-        abs_path = Path(path).resolve(strict=False)
-        uri = abs_path.as_uri()
+        # SQLite allows relative paths by skipping the // after file:,
+        # but pathlib.Path.as_uri() does not support this;
+        # however, temporary databases *must* be relative
+        if path in TEMPORARY_DB_PATHS:
+            # arbitrary paths would need to quote ? to avoid qs injection
+            uri = f"file:{path}"
+        else:
+            # bypasses SQLite's resolution of relative paths, should be fine
+            uri = Path(path).absolute().as_uri()
 
         if self.read_only:
             uri += "?mode=ro"
@@ -637,7 +640,7 @@ def _is_private(path: str) -> bool:
         https://www.sqlite.org/uri.html
 
         """
-        return path in [':memory:', '']
+        return path in TEMPORARY_DB_PATHS
 
 
 class _LocalConnectionFactoryState(threading.local):

tests/test_reader_context.py

@@ -332,11 +332,57 @@ def test_read_only(make_reader, db_path, parser):
     immutable_reader.get_feed(feed)
     assert len(list(immutable_reader.search_entries('one'))) >= 1
 
-    with pytest.raises(StorageError):
+    with pytest.raises(StorageError) as excinfo:
         immutable_reader.add_feed('http://another.feed')
+    assert 'readonly' in str(excinfo.value)
 
     list(immutable_reader.search_entries('one'))
-    with pytest.raises(SearchError):
+    with pytest.raises(SearchError) as excinfo:
         immutable_reader.update_search()
+    assert 'readonly' in str(excinfo.value)
 
     assert len(list(immutable_reader.search_entries('two'))) == 0
+
+
+PATHS_RELATIVE = [
+    'db.sqlite',
+    './db.sqlite',
+    'file:db.sqlite',
+    'db.sqlite?mode=injection',
+]
+
+
+@pytest.mark.parametrize('path', PATHS_RELATIVE)
+@pytest.mark.parametrize('absolute', [False, True])
+def test_paths_shared(make_reader, monkeypatch, tmp_path, path, absolute):
+    monkeypatch.chdir(tmp_path)
+
+    # shouldn't fail
+    make_reader(str(tmp_path / path) if absolute else path)
+
+    # we created the right file
+    assert (tmp_path / path).exists()
+
+
+# test_paths_private covered by the other *_private tests
+
+
+@pytest.mark.parametrize('path', PATHS_RELATIVE)
+def test_paths_read_only_shared(make_reader, monkeypatch, tmp_path, path):
+    monkeypatch.chdir(tmp_path)
+
+    # can't open because database doesn't exist
+    with pytest.raises(StorageError):
+        make_reader(path, read_only=True)
+
+    make_reader(path)
+
+    # shouldn't fail if database already exists
+    reader = make_reader(path, read_only=True)
+
+
+@pytest.mark.parametrize('path', PATHS_PRIVATE)
+def test_paths_read_only_private(make_reader, path):
+    # can open, but can't create tables etc. because read only
+    with pytest.raises(StorageError):
+        make_reader(path, read_only=True)