Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
**Changed**

- **The config-extension loader split into four modules.** `_load_toml_module` (with the `_toml` binding it produces) moved to `dirsql/toml_module.py`, `_load_extension_entries` to `dirsql/load_extension_entries.py` and `_has_bare_name` to `dirsql/has_bare_name.py`; `dirsql/resolve_config_extensions.py` keeps the intervene-or-not decision, `resolve_config_extension_specs`. Every function keeps its name and behavior — internal reorganisation with no change to the published API or CLI behavior. (#1079)
14 changes: 14 additions & 0 deletions packages/python/dirsql/has_bare_name.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
"""Whether any of a config's extension entries names a package."""

from __future__ import annotations

from .is_bare_name import is_bare_name


def _has_bare_name(entries):
return any(
isinstance(e, dict)
and isinstance(e.get("path"), str)
and is_bare_name(e["path"])
for e in entries
)
41 changes: 41 additions & 0 deletions packages/python/dirsql/has_bare_name_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
"""Unit tests for `_has_bare_name`.

The one collaborator, the pure `is_bare_name`, is mocked so the probe's own
filtering -- non-dict entries, non-string paths, short-circuiting -- is what
these observe.
"""

from unittest import mock

import dirsql.has_bare_name as mod


def _patch():
return mock.patch.object(
mod, "is_bare_name", side_effect=lambda path: not path.endswith(".so")
)


def describe_has_bare_name():
def it_is_false_for_no_entries():
with _patch() as is_bare_name:
assert mod._has_bare_name([]) is False
is_bare_name.assert_not_called()

def it_is_false_when_every_path_is_literal():
with _patch():
assert mod._has_bare_name([{"path": "ext/a.so"}]) is False

def it_is_true_when_some_path_is_a_bare_name():
with _patch():
assert mod._has_bare_name([{"path": "ext/a.so"}, {"path": "vec"}]) is True

def it_skips_an_entry_that_is_not_a_table():
with _patch() as is_bare_name:
assert mod._has_bare_name([1, "vec"]) is False
is_bare_name.assert_not_called()

def it_skips_an_entry_whose_path_is_not_a_string():
with _patch() as is_bare_name:
assert mod._has_bare_name([{"path": 42}, {}]) is False
is_bare_name.assert_not_called()
28 changes: 28 additions & 0 deletions packages/python/dirsql/load_extension_entries.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
"""Reading one TOML config's ``[[dirsql.extension]]`` array off disk."""

from __future__ import annotations

import os

from .toml_module import _toml


def _load_extension_entries(config_path):
"""Return ``(entries, base_dir)`` for a config's ``[[dirsql.extension]]``.

``None`` when the config is missing, unreadable/malformed, or declares no
extension array -- the caller should leave such configs to the core.
"""
if not os.path.isfile(config_path):
return None
try:
with open(config_path, "rb") as f:
doc = _toml.load(f)
except (OSError, _toml.TOMLDecodeError):
# Leave a malformed / unreadable config for the core to report.
return None

entries = (doc.get("dirsql") or {}).get("extension")
if not isinstance(entries, list):
return None
return entries, os.path.dirname(os.path.abspath(config_path))
92 changes: 92 additions & 0 deletions packages/python/dirsql/load_extension_entries_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Unit tests for `_load_extension_entries`.

Real `.dirsql.toml` files under a temp dir are parsed by the real TOML module:
the unit under test is the shape-checking around the parse, and a stub parser
would only restate it. The unreadable-file arm mocks `open`.
"""

import os
import tempfile
from unittest import mock

import pytest

import dirsql.load_extension_entries as mod


@pytest.fixture
def cfg_dir():
with tempfile.TemporaryDirectory() as d:
yield d


def _write(cfg_dir, body):
path = os.path.join(cfg_dir, ".dirsql.toml")
with open(path, "w") as f:
f.write(body)
return path


def describe_configs_with_no_usable_extension_array():
def it_returns_none_when_the_config_is_missing():
assert mod._load_extension_entries("/nope/.dirsql.toml") is None

def it_returns_none_on_malformed_toml(cfg_dir):
path = _write(cfg_dir, "this is not = valid = toml")
assert mod._load_extension_entries(path) is None

def it_returns_none_when_the_config_cannot_be_read(cfg_dir):
path = _write(cfg_dir, '[[dirsql.extension]]\npath = "vec"\n')
with mock.patch("builtins.open", side_effect=OSError("denied")):
assert mod._load_extension_entries(path) is None

def it_returns_none_when_there_is_no_dirsql_section(cfg_dir):
path = _write(
cfg_dir,
'[[table]]\nname = "t"\nddl = "CREATE TABLE t (x TEXT)"\nglob = "*"\n',
)
assert mod._load_extension_entries(path) is None

def it_returns_none_when_no_extensions_are_declared(cfg_dir):
path = _write(cfg_dir, '[dirsql]\nignore = ["x"]\n')
assert mod._load_extension_entries(path) is None

def it_returns_none_when_extension_is_not_a_list(cfg_dir):
path = _write(cfg_dir, '[dirsql]\nextension = "not-a-list"\n')
assert mod._load_extension_entries(path) is None


def describe_configs_declaring_extensions():
def it_returns_the_entries_and_the_configs_own_directory(cfg_dir):
path = _write(
cfg_dir,
"[[dirsql.extension]]\n"
'path = "sqlite_vec"\n'
'entrypoint = "sqlite3_vec_init"\n\n'
"[[dirsql.extension]]\n"
'path = "ext/local.so"\n',
)
assert mod._load_extension_entries(path) == (
[
{"path": "sqlite_vec", "entrypoint": "sqlite3_vec_init"},
{"path": "ext/local.so"},
],
cfg_dir,
)

def it_returns_entries_that_are_not_tables_verbatim(cfg_dir):
# Shape-checking each entry is the caller's job; this only asserts the
# array is a list.
path = _write(cfg_dir, "[dirsql]\nextension = [1, 2]\n")
assert mod._load_extension_entries(path) == ([1, 2], cfg_dir)

def it_normalizes_the_base_directory(cfg_dir):
# A traversal segment in the config path must not survive into the
# base directory every extension is then resolved against.
sub = os.path.join(cfg_dir, "sub")
os.mkdir(sub)
_write(sub, '[[dirsql.extension]]\npath = "vec"\n')
_, base = mod._load_extension_entries(
os.path.join(cfg_dir, "sub", "..", "sub", ".dirsql.toml")
)
assert base == sub
54 changes: 2 additions & 52 deletions packages/python/dirsql/resolve_config_extensions.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,59 +14,9 @@

from __future__ import annotations

import os
import sys
from importlib import import_module

from .has_bare_name import _has_bare_name
from .load_extension_entries import _load_extension_entries
from .resolve_entries import _resolve_entries
from .resolve_extension import is_bare_name


def _load_toml_module():
"""Return the TOML parser module for the running interpreter.

``tomllib`` is stdlib only on 3.11+; on 3.10 the ``tomli`` backport it was
derived from (a version-gated dependency) provides the same surface.
Imported by name so a unit test can exercise both arms on any
interpreter -- a literal ``import tomllib`` is unreachable on 3.10 no
matter what ``sys.version_info`` claims.
"""
if sys.version_info >= (3, 11):
return import_module("tomllib")
return import_module("tomli")


_toml = _load_toml_module()


def _load_extension_entries(config_path):
"""Return ``(entries, base_dir)`` for a config's ``[[dirsql.extension]]``.

``None`` when the config is missing, unreadable/malformed, or declares no
extension array -- the caller should leave such configs to the core.
"""
if not os.path.isfile(config_path):
return None
try:
with open(config_path, "rb") as f:
doc = _toml.load(f)
except (OSError, _toml.TOMLDecodeError):
# Leave a malformed / unreadable config for the core to report.
return None

entries = (doc.get("dirsql") or {}).get("extension")
if not isinstance(entries, list):
return None
return entries, os.path.dirname(os.path.abspath(config_path))


def _has_bare_name(entries):
return any(
isinstance(e, dict)
and isinstance(e.get("path"), str)
and is_bare_name(e["path"])
for e in entries
)


def resolve_config_extension_specs(config_path):
Expand Down
Loading
Loading