exp: enforce waiting for task completion in CI

tests/func/experiments/test_apply.py

@@ -35,7 +35,7 @@ def test_apply_failed(tmp_dir, scm, dvc, failed_exp_stage, session_queue):
     dvc.experiments.run(
         failed_exp_stage.addressing, params=["foo=3"], queue=True, name="foo"
     )
-    dvc.experiments.run(run_all=True)
+    session_queue.wait(["foo"])
 
     with pytest.raises(InvalidArgumentError):
         dvc.experiments.apply("bar")

tests/func/experiments/test_queue.py

@@ -22,8 +22,9 @@ def test_celery_logs(
     test_queue,
 ):
     celery_queue = dvc.experiments.celery_queue
-    dvc.experiments.run(failed_exp_stage.addressing, queue=True)
+    dvc.experiments.run(failed_exp_stage.addressing, queue=True, name="foo")
     dvc.experiments.run(run_all=True)
+    test_queue.wait(["foo"])
 
     done_result = first(celery_queue.iter_done())
 
@@ -34,10 +35,6 @@ def test_celery_logs(
     assert "failed to reproduce 'failed-copy-file'" in captured.out
 
 
-@pytest.mark.xfail(
-    strict=False,
-    reason="https://github.com/iterative/dvc/issues/9143",
-)
 def test_queue_remove_done(
     dvc,
     exp_stage,
@@ -63,6 +60,7 @@ def test_queue_remove_done(
             name=name_fail,
         )
     dvc.experiments.run(run_all=True)
+    test_queue.wait(success_tasks + failed_tasks)
     assert len(celery_queue.failed_stash) == 3
     status = to_dict(celery_queue.status())
     assert len(status) == 6

tests/func/experiments/test_show.py

@@ -156,15 +156,18 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage):
 
 
 @pytest.mark.vscode
-def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage):
+def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage, test_queue):
     baseline_rev = scm.get_rev()
     timestamp = datetime.fromtimestamp(
         scm.gitpython.repo.rev_parse(baseline_rev).committed_date
     )
 
-    dvc.experiments.run(failed_exp_stage.addressing, params=["foo=2"], queue=True)
+    dvc.experiments.run(
+        failed_exp_stage.addressing, params=["foo=2"], queue=True, name="show-failed"
+    )
     exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
     dvc.experiments.run(run_all=True)
+    test_queue.wait(["show-failed"])
     experiments = dvc.experiments.show()[baseline_rev]
 
     expected_baseline = {
@@ -542,15 +545,20 @@ def test_show_running_celery(tmp_dir, scm, dvc, exp_stage, mocker):
     assert results["workspace"]["baseline"]["data"]["status"] == "Success"
 
 
-def test_show_running_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker):
+def test_show_running_checkpoint(
+    tmp_dir, scm, dvc, checkpoint_stage, mocker, test_queue
+):
     from dvc.repo.experiments.executor.local import TempDirExecutor
 
     baseline_rev = scm.get_rev()
-    dvc.experiments.run(checkpoint_stage.addressing, params=["foo=2"], queue=True)
+    dvc.experiments.run(
+        checkpoint_stage.addressing, params=["foo=2"], queue=True, name="foo"
+    )
     queue = dvc.experiments.celery_queue
     entries = list(queue.iter_queued())
 
     run_results = dvc.experiments.run(run_all=True)
+    test_queue.wait(["foo"])
     checkpoint_rev = first(run_results)
     exp_ref = first(exp_refs_by_rev(scm, checkpoint_rev))