tests: use session-scoped exp celery fixtures when possible

tests/func/experiments/conftest.py

@@ -5,6 +5,9 @@ from tests.unit.repo.experiments.conftest import (  # noqa, pylint disable=unuse
     exp_stage,
     failed_checkpoint_stage,
     failed_exp_stage,
+    session_app,
+    session_queue,
+    session_worker,
     test_queue,
 )
 
@@ -24,7 +27,7 @@ def http_auth_patch(mocker):
 
 
 @pytest.fixture(params=[True, False])
-def workspace(request, test_queue) -> bool:  # noqa: F811
+def workspace(request, session_queue) -> bool:  # noqa: F811
     return request.param
 
 

tests/func/experiments/test_apply.py

@@ -26,7 +26,7 @@ def test_apply(tmp_dir, scm, dvc, exp_stage):
     assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 3"
 
 
-def test_apply_failed(tmp_dir, scm, dvc, failed_exp_stage, test_queue):
+def test_apply_failed(tmp_dir, scm, dvc, failed_exp_stage, session_queue):
     from dvc.exceptions import InvalidArgumentError
 
     dvc.experiments.run(failed_exp_stage.addressing, params=["foo=2"], queue=True)

tests/func/experiments/test_experiments.py

@@ -131,7 +131,7 @@ def test_get_baseline(tmp_dir, scm, dvc, exp_stage):
     assert dvc.experiments.get_baseline(f"{CELERY_STASH}@{{1}}") == init_rev
 
 
-def test_update_py_params(tmp_dir, scm, dvc, test_queue, copy_script):
+def test_update_py_params(tmp_dir, scm, dvc, session_queue, copy_script):
     tmp_dir.gen("params.py", "INT = 1\n")
     stage = dvc.run(
         cmd="python copy.py params.py metrics.py",

tests/func/experiments/test_queue.py

@@ -19,6 +19,7 @@ def test_celery_logs(
     failed_exp_stage,
     follow,
     capsys,
+    test_queue,
 ):
     celery_queue = dvc.experiments.celery_queue
     dvc.experiments.run(failed_exp_stage.addressing, queue=True)
@@ -37,6 +38,7 @@ def test_queue_remove_done(
     dvc,
     exp_stage,
     failed_exp_stage,
+    test_queue,
 ):
     queue_length = 3
     success_tasks = []

tests/unit/repo/experiments/queue/test_celery.py

@@ -49,10 +49,10 @@ def test_shutdown_with_kill(test_queue, mocker):
     )
 
 
-def test_post_run_after_kill(test_queue):
+def test_post_run_after_kill(session_queue):
     from celery import chain
 
-    sig_bar = test_queue.proc.run_signature(
+    sig_bar = session_queue.proc.run_signature(
         ["python3", "-c", "import time; time.sleep(10)"], name="bar"
     )
     sig_bar.freeze()
@@ -65,7 +65,7 @@ def test_post_run_after_kill(test_queue):
 
     while True:
         try:
-            test_queue.proc.kill("bar")
+            session_queue.proc.kill("bar")
             assert result_foo.status == "PENDING"
             break
         except ProcessLookupError: