tests: add coverage for deferred audit findings + isolate test_settings from ambient env

tests/integration/test_dependencies.py

@@ -1,3 +1,5 @@
+from unittest.mock import patch
+
 from pytest import raises
 
 from datajoint import errors
@@ -50,3 +52,58 @@ def test_unique_dependency(thing_tables):
     # duplicate foreign key attributes = not ok
     with raises(errors.DuplicateError):
         c.insert1(dict(a=1, b1=1, b2=1))
+
+
+class TestLoadAllShortCircuit:
+    """Finding #7 (audit deferred): the ``load_all_upstream`` /
+    ``load_all_downstream`` helpers must short-circuit when the graph already
+    covers every schema they would discover. The guard at
+    ``dependencies.py:266-267`` and ``:300-301`` was introduced by PR #1499 as
+    the ~53 ms/key populate speedup; a regression would silently pass CI
+    because the correctness of the graph is unaffected — only its cost.
+
+    These tests count calls to ``Dependencies.load`` on the second invocation.
+    The first call must load once (to populate ``_loaded_schemas``); the
+    second must NOT call ``load`` at all — the ``if self._loaded and
+    known_schemas <= self._loaded_schemas: return`` early-return owns the
+    entire path.
+    """
+
+    def test_load_all_downstream_short_circuits_on_repeat(self, thing_tables):
+        # thing_tables ensures at least one schema is activated on the
+        # connection, so the ``if not known_schemas: self.load(); return``
+        # bail-out branch does not fire.
+        a, _, _, _, _ = thing_tables
+        conn = a.connection
+
+        # Warm up the graph once — this call is allowed to invoke load().
+        conn.dependencies.load_all_downstream()
+        assert conn.dependencies._loaded is True
+        assert conn.dependencies._loaded_schemas, "warm-up must populate _loaded_schemas"
+
+        # Second identical call must hit the early-return: no load(), no rebuild.
+        with patch.object(type(conn.dependencies), "load", wraps=conn.dependencies.load) as spy_load:
+            conn.dependencies.load_all_downstream()
+        assert spy_load.call_count == 0, (
+            "load_all_downstream must short-circuit when _loaded_schemas already "
+            "covers the discovered set (#1493, PR #1499). Deleting the "
+            "`if self._loaded and known_schemas <= self._loaded_schemas: return` "
+            "at dependencies.py:266-267 would make this test fail."
+        )
+
+    def test_load_all_upstream_short_circuits_on_repeat(self, thing_tables):
+        a, _, _, _, _ = thing_tables
+        conn = a.connection
+
+        conn.dependencies.load_all_upstream()
+        assert conn.dependencies._loaded is True
+        assert conn.dependencies._loaded_schemas
+
+        with patch.object(type(conn.dependencies), "load", wraps=conn.dependencies.load) as spy_load:
+            conn.dependencies.load_all_upstream()
+        assert spy_load.call_count == 0, (
+            "load_all_upstream must short-circuit when _loaded_schemas already "
+            "covers the discovered set (#1493, PR #1499). Deleting the "
+            "`if self._loaded and known_schemas <= self._loaded_schemas: return` "
+            "at dependencies.py:300-301 would make this test fail."
+        )

tests/integration/test_gc.py

@@ -843,3 +843,108 @@ def test_scan_errors_in_stats_dict(self):
         assert len(stats["scan_errors"]) == 1
         assert stats["scan_errors"][0].startswith("list_hash_paths(s):")
         assert "s3 access denied" in stats["scan_errors"][0]
+
+
+class TestDeleteSchemaPathPruning:
+    """Finding #8 (audit deferred): ``GarbageCollector.delete_schema_path``
+    prunes now-empty parent directories after removing the orphaned file
+    (``gc.py:277-306``). The pruning walk was previously uncovered — no test
+    exercised the ``while parent and parent != root and parent.startswith(root)``
+    loop, its ``fs.ls`` "not empty" break, its exception guards, or the
+    stop-at-store-root boundary.
+
+    A regression to "delete the file, leave every parent dir behind" would
+    silently pass CI and gradually accumulate empty directories on every
+    store, so a live-data test that inspects the on-disk tree after
+    ``collect(dry_run=False)`` is the right lock.
+    """
+
+    @pytest.fixture
+    def schema_prune(self, connection_test, prefix, mock_stores):
+        schema = dj.Schema(
+            f"{prefix}_test_gc_prune",
+            context={"GcObjectTest": GcObjectTest},
+            connection=connection_test,
+        )
+        schema(GcObjectTest)
+        yield schema
+        schema.drop()
+
+    def test_pruning_removes_empty_pk_directory(self, schema_prune):
+        """Deleting the only file under a pk-dir must remove the file AND
+        prune the now-empty pk-dir; a sibling pk-dir with a live file survives.
+
+        Regressing ``delete_schema_path`` to skip the ``while ... rmdir``
+        pruning loop would leave the empty ``rid=1/`` directory behind and
+        make this test fail.
+        """
+        from pathlib import Path
+
+        # Two rows in separate pk-dirs: rid=1 will be orphaned, rid=2 stays live.
+        GcObjectTest.insert1({"rid": 1, "results": b"orphan-me"})
+        GcObjectTest.insert1({"rid": 2, "results": b"live"})
+
+        collector = _gc(schema_prune)
+        cfg = schema_prune.connection._config
+        store_root = Path(cfg.get_store_spec("local")["location"])
+
+        # Sanity: both pk-dirs exist on disk pre-delete.
+        refs_all = collector.schema_references()
+        assert len(refs_all) == 2
+        for ref in refs_all:
+            assert (store_root / ref).exists(), f"expected {ref} to exist pre-delete"
+
+        # Orphan rid=1 and reclaim.
+        (GcObjectTest & {"rid": 1}).delete(prompt=False)
+        refs_live = collector.schema_references()
+        live_ref = next(iter(refs_live))
+        dead_ref = next(iter(refs_all - refs_live))
+
+        stats = collector.collect(dry_run=False)
+        assert stats["schema_paths_deleted"] >= 1
+
+        # The orphaned file is gone.
+        assert not (store_root / dead_ref).exists(), f"orphaned schema file must be deleted; still present at {dead_ref}"
+        # The pruning loop must have removed the now-empty pk-dir.
+        dead_pk_dir = (store_root / dead_ref).parent
+        assert not dead_pk_dir.exists(), (
+            f"parent pk-directory must be pruned when its last file is removed; "
+            f"still present at {dead_pk_dir} — deleting the pruning loop at "
+            f"gc.py:293-302 would make this test fail"
+        )
+
+        # The sibling pk-dir with a live file must be preserved.
+        live_pk_dir = (store_root / live_ref).parent
+        assert live_pk_dir.exists(), f"pk-directory holding a live file must not be pruned; missing {live_pk_dir}"
+        assert (store_root / live_ref).exists(), f"live schema file must survive collect(); missing {live_ref}"
+
+        # The table directory (grandparent) still holds live content, so it
+        # must be preserved — the pruning walk must stop at "not empty".
+        assert live_pk_dir.parent.exists(), "table directory must survive when it still holds a live pk-dir"
+
+    def test_pruning_stops_at_store_root(self, schema_prune):
+        """After removing the last orphan under a schema, the pruning walk may
+        remove the pk-dir, table-dir, schema-dir, and section-dir but MUST
+        stop at the store root — a regression to "keep walking past the root"
+        would be catastrophic (removes user's storage directory itself).
+
+        The boundary ``while parent and parent != root and parent.startswith(root)``
+        owns this invariant.
+        """
+        from pathlib import Path
+
+        GcObjectTest.insert1({"rid": 1, "results": b"only-row"})
+
+        collector = _gc(schema_prune)
+        cfg = schema_prune.connection._config
+        store_root = Path(cfg.get_store_spec("local")["location"])
+        assert store_root.exists()
+
+        (GcObjectTest & {"rid": 1}).delete(prompt=False)
+        collector.collect(dry_run=False)
+
+        # Store root must survive even when its entire contents were pruned.
+        assert store_root.exists(), (
+            f"pruning walk must stop at store root; root itself was removed at {store_root} — "
+            f"the `parent != root and parent.startswith(root)` boundary in gc.py:295 owns this"
+        )

tests/integration/test_object.py

@@ -130,6 +130,55 @@ def test_build_object_path_with_partition(self):
         # section prefix first, then partition attrs (schema_prefix default "_schema")
         assert path.startswith("_schema/subject_id=1")
 
+    def test_build_object_path_empty_schema_prefix(self):
+        """Finding #19 (audit deferred): ``schema_prefix=""`` reproduces the
+        documented legacy pre-2.3.1 storage layout (no section prefix).
+
+        The guard at ``storage.py:267-269`` (``prefix = schema_prefix.strip("/");
+        if prefix: parts.append(prefix)``) exists specifically so an empty
+        prefix does NOT produce a path with an empty leading segment (e.g.
+        ``"/myschema/MyTable/..."``). Every other test in this class passes
+        ``schema_prefix="_schema"`` and would still pass if the ``if prefix:``
+        guard were deleted, so a regression to always-prepend would go
+        unnoticed. This test locks in the pre-2.3.1 layout.
+        """
+        path, token = build_object_path(
+            schema="myschema",
+            table="MyTable",
+            field="data_file",
+            primary_key={"id": 42},
+            ext=".dat",
+            schema_prefix="",
+        )
+        assert path == f"myschema/MyTable/id=42/data_file_{token}.dat", (
+            f"empty schema_prefix must produce a path with NO leading section "
+            f"segment (documented legacy layout); got {path!r}"
+        )
+        # Regression sentinel: deleting the ``if prefix:`` guard would produce
+        # a leading empty segment, i.e. ``"/myschema/..."``.
+        assert not path.startswith("/"), (
+            f"path must not have a leading empty segment when schema_prefix is empty; " f"got {path!r}"
+        )
+
+    def test_build_object_path_empty_schema_prefix_with_partition(self):
+        """Finding #19 companion: legacy layout must also compose correctly
+        with a partition pattern — partition attrs lead, no empty segment
+        before them."""
+        path, token = build_object_path(
+            schema="myschema",
+            table="MyTable",
+            field="data",
+            primary_key={"subject_id": 1, "session_id": 2},
+            ext=".dat",
+            partition_pattern="{subject_id}",
+            schema_prefix="",
+        )
+        assert path.startswith("subject_id=1/myschema/"), (
+            f"empty schema_prefix + partition must start with the partition "
+            f"segment (no empty leading segment); got {path!r}"
+        )
+        assert not path.startswith("/")
+
 
 class TestObjectRef:
     """Tests for ObjectRef class."""

tests/unit/test_settings.py

@@ -509,6 +509,19 @@ def test_load_store_arbitrary_attr(self, tmp_path):
 class TestStoreEnv:
     """Test DJ_STORES env var and DJ_IGNORE_CONFIG_FILE flag."""
 
+    @pytest.fixture(autouse=True)
+    def _isolate_dj_env(self, monkeypatch):
+        """Clear ambient DJ_* env vars so tests are isolated from the developer's shell.
+
+        pydantic-settings reads any DJ_* prefixed variable and merges it into the
+        settings object; a developer with ``DJ_HOST``/``DJ_USER``/``DJ_PASS``/
+        ``DJ_PORT``/``DJ_BACKEND`` exported (e.g. from a local ``.env``) would
+        otherwise see assertions on the config's defaults or file-loaded values
+        fail here even though the test's own contract is unrelated.
+        """
+        for var in ("DJ_HOST", "DJ_USER", "DJ_PASS", "DJ_PORT", "DJ_BACKEND"):
+            monkeypatch.delenv(var, raising=False)
+
     def _isolate_filesystem(self, monkeypatch, tmp_path):
         """chdir into a tmp_path with a .git sentinel so find_config_file stops there."""
         (tmp_path / ".git").mkdir()
@@ -937,6 +950,19 @@ def test_database_prefix_empty_no_warning(self):
 class TestBackendConfiguration:
     """Test database backend configuration and port auto-detection."""
 
+    @pytest.fixture(autouse=True)
+    def _isolate_dj_env(self, monkeypatch):
+        """Clear ambient DJ_* env vars so tests are isolated from the developer's shell.
+
+        ``DatabaseSettings()`` reads ``DJ_HOST``/``DJ_USER``/``DJ_PASS``/
+        ``DJ_PORT``/``DJ_BACKEND`` from the process env; leaked values would
+        otherwise fail assertions on backend defaults and auto-detected ports.
+        Individual tests may still ``monkeypatch.setenv(...)`` to test explicit
+        settings — the fixture runs first, so their setenv wins.
+        """
+        for var in ("DJ_HOST", "DJ_USER", "DJ_PASS", "DJ_PORT", "DJ_BACKEND"):
+            monkeypatch.delenv(var, raising=False)
+
     def test_backend_default(self):
         """Test default backend is mysql."""
         from datajoint.settings import DatabaseSettings