refactor test fixtures

Author: d-v-b

.pre-commit-config.yaml

@@ -30,7 +30,7 @@ repos:
   rev: 25.1.0 # matching versions in pyproject.toml and github actions
   hooks:
   - id: black
-    args: ["--check", "-v", "src", "tests", "--diff"] # --required-version is conflicting with pre-commit
+    args: ["-v", "src", "tests", "--diff"] # --required-version is conflicting with pre-commit
 - repo: https://github.com/PyCQA/flake8
   rev: 7.3.0
   hooks:

tests/conftest.py

@@ -45,6 +45,57 @@ def pytest_configure(config):
     pass
 
 
+@pytest.fixture
+def clean_autopopulate(experiment, trial, ephys):
+    """
+    Explicit cleanup fixture for autopopulate tests.
+
+    Cleans experiment/trial/ephys tables after test completes.
+    Tests must explicitly request this fixture to get cleanup.
+    """
+    yield
+    # Cleanup after test - delete in reverse dependency order
+    ephys.delete()
+    trial.delete()
+    experiment.delete()
+
+
+@pytest.fixture
+def clean_jobs(schema_any):
+    """
+    Explicit cleanup fixture for jobs tests.
+
+    Cleans jobs table before test runs.
+    Tests must explicitly request this fixture to get cleanup.
+    """
+    try:
+        schema_any.jobs.delete()
+    except DataJointError:
+        pass
+    yield
+
+
+@pytest.fixture
+def clean_test_tables(test, test_extra, test_no_extra):
+    """
+    Explicit cleanup fixture for relation tests using test tables.
+
+    Ensures test table has lookup data and restores clean state after test.
+    Tests must explicitly request this fixture to get cleanup.
+    """
+    # Ensure lookup data exists before test
+    if not test:
+        test.insert(test.contents, skip_duplicates=True)
+
+    yield
+
+    # Restore original state after test
+    test.delete()
+    test.insert(test.contents, skip_duplicates=True)
+    test_extra.delete()
+    test_no_extra.delete()
+
+
 # Global container registry for cleanup
 _active_containers = set()
 _docker_client = None
@@ -547,7 +598,7 @@ def mock_cache(tmpdir_factory):
         dj.config["cache"] = og_cache
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def schema_any(connection_test, prefix):
     schema_any = dj.Schema(
         prefix + "_test1", schema.LOCALS_ANY, connection=connection_test
@@ -603,6 +654,63 @@ def schema_any(connection_test, prefix):
     schema_any.drop()
 
 
+@pytest.fixture
+def schema_any_fresh(connection_test, prefix):
+    """Function-scoped schema_any for tests that need fresh schema state."""
+    schema_any = dj.Schema(
+        prefix + "_test1_fresh", schema.LOCALS_ANY, connection=connection_test
+    )
+    assert schema.LOCALS_ANY, "LOCALS_ANY is empty"
+    try:
+        schema_any.jobs.delete()
+    except DataJointError:
+        pass
+    schema_any(schema.TTest)
+    schema_any(schema.TTest2)
+    schema_any(schema.TTest3)
+    schema_any(schema.NullableNumbers)
+    schema_any(schema.TTestExtra)
+    schema_any(schema.TTestNoExtra)
+    schema_any(schema.Auto)
+    schema_any(schema.User)
+    schema_any(schema.Subject)
+    schema_any(schema.Language)
+    schema_any(schema.Experiment)
+    schema_any(schema.Trial)
+    schema_any(schema.Ephys)
+    schema_any(schema.Image)
+    schema_any(schema.UberTrash)
+    schema_any(schema.UnterTrash)
+    schema_any(schema.SimpleSource)
+    schema_any(schema.SigIntTable)
+    schema_any(schema.SigTermTable)
+    schema_any(schema.DjExceptionName)
+    schema_any(schema.ErrorClass)
+    schema_any(schema.DecimalPrimaryKey)
+    schema_any(schema.IndexRich)
+    schema_any(schema.ThingA)
+    schema_any(schema.ThingB)
+    schema_any(schema.ThingC)
+    schema_any(schema.ThingD)
+    schema_any(schema.ThingE)
+    schema_any(schema.Parent)
+    schema_any(schema.Child)
+    schema_any(schema.ComplexParent)
+    schema_any(schema.ComplexChild)
+    schema_any(schema.SubjectA)
+    schema_any(schema.SessionA)
+    schema_any(schema.SessionStatusA)
+    schema_any(schema.SessionDateA)
+    schema_any(schema.Stimulus)
+    schema_any(schema.Longblob)
+    yield schema_any
+    try:
+        schema_any.jobs.delete()
+    except DataJointError:
+        pass
+    schema_any.drop()
+
+
 @pytest.fixture
 def thing_tables(schema_any):
     a = schema.ThingA()
@@ -623,7 +731,7 @@ def thing_tables(schema_any):
     yield a, b, c, d, e
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def schema_simp(connection_test, prefix):
     schema = dj.Schema(
         prefix + "_relational", schema_simple.LOCALS_SIMPLE, connection=connection_test
@@ -653,7 +761,7 @@ def schema_simp(connection_test, prefix):
     schema.drop()
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def schema_adv(connection_test, prefix):
     schema = dj.Schema(
         prefix + "_advanced",
@@ -694,7 +802,7 @@ def schema_ext(
     schema.drop()
 
 
-@pytest.fixture
+@pytest.fixture(scope="module")
 def schema_uuid(connection_test, prefix):
     schema = dj.Schema(
         prefix + "_test1",

tests/test_alter.py

@@ -14,12 +14,12 @@
 
 
 @pytest.fixture
-def schema_alter(connection_test, schema_any):
-    # Overwrite Experiment and Parent nodes
-    schema_any(Experiment, context=LOCALS_ALTER)
-    schema_any(Parent, context=LOCALS_ALTER)
-    yield schema_any
-    schema_any.drop()
+def schema_alter(connection_test, schema_any_fresh):
+    # Overwrite Experiment and Parent nodes using fresh schema
+    schema_any_fresh(Experiment, context=LOCALS_ALTER)
+    schema_any_fresh(Parent, context=LOCALS_ALTER)
+    yield schema_any_fresh
+    schema_any_fresh.drop()
 
 
 class TestAlter:

tests/test_autopopulate.py

@@ -7,7 +7,7 @@
 from . import schema
 
 
-def test_populate(trial, subject, experiment, ephys, channel):
+def test_populate(clean_autopopulate, trial, subject, experiment, ephys, channel):
     # test simple populate
     assert subject, "root tables are empty"
     assert not experiment, "table already filled?"
@@ -33,7 +33,7 @@ def test_populate(trial, subject, experiment, ephys, channel):
     assert channel
 
 
-def test_populate_with_success_count(subject, experiment, trial):
+def test_populate_with_success_count(clean_autopopulate, subject, experiment, trial):
     # test simple populate
     assert subject, "root tables are empty"
     assert not experiment, "table already filled?"
@@ -51,7 +51,7 @@ def test_populate_with_success_count(subject, experiment, trial):
     assert len(trial.key_source & trial) == success_count
 
 
-def test_populate_key_list(subject, experiment, trial):
+def test_populate_key_list(clean_autopopulate, subject, experiment, trial):
     # test simple populate
     assert subject, "root tables are empty"
     assert not experiment, "table already filled?"
@@ -63,7 +63,7 @@ def test_populate_key_list(subject, experiment, trial):
     assert n == ret["success_count"]
 
 
-def test_populate_exclude_error_and_ignore_jobs(schema_any, subject, experiment):
+def test_populate_exclude_error_and_ignore_jobs(clean_autopopulate, schema_any, subject, experiment):
     # test simple populate
     assert subject, "root tables are empty"
     assert not experiment, "table already filled?"
@@ -79,7 +79,7 @@ def test_populate_exclude_error_and_ignore_jobs(schema_any, subject, experiment)
     assert len(experiment.key_source & experiment) == len(experiment.key_source) - 2
 
 
-def test_allow_direct_insert(subject, experiment):
+def test_allow_direct_insert(clean_autopopulate, subject, experiment):
     assert subject, "root tables are empty"
     key = subject.fetch("KEY", limit=1)[0]
     key["experiment_id"] = 1000
@@ -88,14 +88,14 @@ def test_allow_direct_insert(subject, experiment):
 
 
 @pytest.mark.parametrize("processes", [None, 2])
-def test_multi_processing(subject, experiment, processes):
+def test_multi_processing(clean_autopopulate, subject, experiment, processes):
     assert subject, "root tables are empty"
     assert not experiment, "table already filled?"
     experiment.populate(processes=None)
     assert len(experiment) == len(subject) * experiment.fake_experiments_per_subject
 
 
-def test_allow_insert(subject, experiment):
+def test_allow_insert(clean_autopopulate, subject, experiment):
     assert subject, "root tables are empty"
     key = subject.fetch("KEY")[0]
     key["experiment_id"] = 1001

tests/test_cascading_delete.py

@@ -8,6 +8,17 @@
 
 @pytest.fixture
 def schema_simp_pop(schema_simp):
+    # Clean up tables first to ensure fresh state with module-scoped schema
+    # Delete in reverse dependency order
+    Profile().delete()
+    Website().delete()
+    G().delete()
+    E().delete()
+    D().delete()
+    B().delete()
+    L().delete()
+    A().delete()
+
     A().insert(A.contents, skip_duplicates=True)
     L().insert(L.contents, skip_duplicates=True)
     B().populate()

tests/test_declare.py

@@ -268,7 +268,7 @@ class BadName(dj.Manual):
         schema_any(BadName)
 
 
-def test_bad_fk_rename(schema_any):
+def test_bad_fk_rename(schema_any_fresh):
     """issue #381"""
 
     class A(dj.Manual):
@@ -281,9 +281,9 @@ class B(dj.Manual):
         b -> A    # invalid, the new syntax is (b) -> A
         """
 
-    schema_any(A)
+    schema_any_fresh(A)
     with pytest.raises(dj.DataJointError):
-        schema_any(B)
+        schema_any_fresh(B)
 
 
 def test_primary_nullable_foreign_key(schema_any):
@@ -401,7 +401,7 @@ def test_add_hidden_timestamp_default_value():
     ), "Default value for add_hidden_timestamp is not False"
 
 
-def test_add_hidden_timestamp_enabled(enable_add_hidden_timestamp, schema_any):
+def test_add_hidden_timestamp_enabled(enable_add_hidden_timestamp, schema_any_fresh):
     assert config["add_hidden_timestamp"], "add_hidden_timestamp is not enabled"
     msg = f"{Experiment().heading._attributes=}"
     assert any(
@@ -414,7 +414,7 @@ def test_add_hidden_timestamp_enabled(enable_add_hidden_timestamp, schema_any):
     assert not any(a.is_hidden for a in Experiment().heading.attributes.values()), msg
 
 
-def test_add_hidden_timestamp_disabled(disable_add_hidden_timestamp, schema_any):
+def test_add_hidden_timestamp_disabled(disable_add_hidden_timestamp, schema_any_fresh):
     assert not config[
         "add_hidden_timestamp"
     ], "expected add_hidden_timestamp to be False"

tests/test_jobs.py

@@ -9,7 +9,7 @@
 from . import schema
 
 
-def test_reserve_job(subject, schema_any):
+def test_reserve_job(clean_jobs, subject, schema_any):
     assert subject
     table_name = "fake_table"
 
@@ -47,7 +47,7 @@ def test_reserve_job(subject, schema_any):
     assert not schema_any.jobs, "failed to clear error jobs"
 
 
-def test_restrictions(schema_any):
+def test_restrictions(clean_jobs, schema_any):
     jobs = schema_any.jobs
     jobs.delete()
     jobs.reserve("a", {"key": "a1"})
@@ -62,7 +62,7 @@ def test_restrictions(schema_any):
     jobs.delete()
 
 
-def test_sigint(schema_any):
+def test_sigint(clean_jobs, schema_any):
     try:
         schema.SigIntTable().populate(reserve_jobs=True)
     except KeyboardInterrupt:
@@ -74,7 +74,7 @@ def test_sigint(schema_any):
     assert error_message == "KeyboardInterrupt"
 
 
-def test_sigterm(schema_any):
+def test_sigterm(clean_jobs, schema_any):
     try:
         schema.SigTermTable().populate(reserve_jobs=True)
     except SystemExit:
@@ -86,14 +86,14 @@ def test_sigterm(schema_any):
     assert error_message == "SystemExit: SIGTERM received"
 
 
-def test_suppress_dj_errors(schema_any):
+def test_suppress_dj_errors(clean_jobs, schema_any):
     """test_suppress_dj_errors: dj errors suppressible w/o native py blobs"""
     with dj.config(enable_python_native_blobs=False):
         schema.ErrorClass.populate(reserve_jobs=True, suppress_errors=True)
     assert len(schema.DjExceptionName()) == len(schema_any.jobs) > 0
 
 
-def test_long_error_message(subject, schema_any):
+def test_long_error_message(clean_jobs, subject, schema_any):
     # create long error message
     long_error_message = "".join(
         random.choice(string.ascii_letters) for _ in range(ERROR_MESSAGE_LENGTH + 100)
@@ -129,7 +129,7 @@ def test_long_error_message(subject, schema_any):
     schema_any.jobs.delete()
 
 
-def test_long_error_stack(subject, schema_any):
+def test_long_error_stack(clean_jobs, subject, schema_any):
     # create long error stack
     STACK_SIZE = (
         89942  # Does not fit into small blob (should be 64k, but found to be higher)