From 9f639ba2a982ca7f0143c88a55f601bbd1e53fd5 Mon Sep 17 00:00:00 2001 From: tligui_y Date: Sun, 23 Nov 2025 17:40:38 +0100 Subject: [PATCH] Add tests/test_core_scanbackend.py --- tests/test_core_scanbackend.py | 852 +++++++++++++++++++++++++++++++++ 1 file changed, 852 insertions(+) create mode 100644 tests/test_core_scanbackend.py diff --git a/tests/test_core_scanbackend.py b/tests/test_core_scanbackend.py new file mode 100644 index 00000000..e7901f5a --- /dev/null +++ b/tests/test_core_scanbackend.py @@ -0,0 +1,852 @@ +import pytest +from pathlib import Path +import os + +from slic.core.scanner.scanbackend import ( + ScanBackend, + is_sfdaq, is_only_sfdaq, + print_all_current_values, get_all_current_values, + set_all_target_values_and_wait, set_all_target_values, + wait_for_all, stop_all, +) + +from slic.core.acquisition import SFAcquisition +from slic.core.acquisition.fakeacquisition import FakeAcquisition +from slic.core.adjustable.dummyadjustable import DummyAdjustable +from slic.core.task import DAQTask + + +# Dummies for conditions/sensors + +class DummyCondition: + def __init__(self, repeats=0): + self.repeats = repeats + self._stopped = False + def wants_repeat(self): + self.repeats -= 1 + return self.repeats >= 0 + def stop(self): + self._stopped = True + + +class DummySensor: + counter = 0 + def __init__(self, name=None): + DummySensor.counter += 1 + self.name = name or f"sensor_{DummySensor.counter}" + self.started = False + self.stopped = False + def start(self): + self.started = True + def stop(self): + self.stopped = True + def get(self): + return 3.14 + + +class DummyRemotePlot: + def __init__(self, fail=False): + self.fail = fail + self.created = False + self.appended = False + self.last_data = None + self.last_filename = None + + def new_plot(self, filename, cfg): + self.created = True + if self.fail: + raise ConnectionRefusedError + + def append_data(self, filename, data): + self.appended = True + self.last_data = data + self.last_filename = filename + if self.fail: + raise ConnectionRefusedError + + +# Main ScanBackend tests + +def test_is_sfdaq_and_only_sfdaq(): + # Test SFDAQ detection functions with different acquisition types + s1, s2 = SFAcquisition(), SFAcquisition() + f1 = FakeAcquisition() + random_obj = object() + + assert is_sfdaq(s1) + assert is_sfdaq(f1) + assert not is_sfdaq(random_obj) + + assert is_only_sfdaq([s1, s2]) + assert is_only_sfdaq([f1, s1]) + assert not is_only_sfdaq([s1, random_obj]) + +def test_get_filename(tmp_path): + # Test filename generation with both make_scan_sub_dir modes + adjs = [DummyAdjustable(name="AX", ID="AX")] + acqs = [FakeAcquisition()] + + # Case 1 - make_scan_sub_dir = False + sb1 = ScanBackend( + adjs, [[1]], acqs, "scanfile", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=False, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + f1 = sb1.get_filename(7) + assert f1.endswith("scanfile_step0007") + assert os.path.basename(f1).startswith("scanfile") + + # Case 2 - make_scan_sub_dir = True + sb2 = ScanBackend( + adjs, [[1]], acqs, "scanfile", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=True, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + f2 = sb2.get_filename(3) + expected_sub = os.path.join("scanfile", "scanfile_step0003") + assert f2.endswith(expected_sub) + + sb3 = ScanBackend( + adjs, [[1]], acqs, "/tmp/path/to/custom_name", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=True, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + f3 = sb3.get_filename(1) + assert f3.endswith(os.path.join("custom_name", "custom_name_step0001")) + + +def test_create_output_dirs(tmp_path): + # Test directory creation behavior with different acquisition types and settings + adjs = [DummyAdjustable(ID="A1")] + + # Test case 1: SFDAQ only, make_scan_sub_dir=True + sfdaq_acq = SFAcquisition() + sb = ScanBackend( + adjs, [[1]], [sfdaq_acq], + filename="scan_sfdaq", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=True, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + sb.create_output_dirs() + for root, dirs, files in os.walk(tmp_path): + assert not dirs + + # Test case 2: FakeAcquisition only, make_scan_sub_dir=False + fake_acq = FakeAcquisition("non_sfdaq", "p1") + fake_acq.default_dir = str(tmp_path / "fake_default") + sb = ScanBackend( + adjs, [[1]], [fake_acq], + filename="scan_fake", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=False, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + sb.create_output_dirs() + assert (tmp_path / sb.scan_info.base_dir).exists() + expected_data_dir = os.path.join(fake_acq.default_dir, sb.data_base_dir) + assert os.path.isdir(expected_data_dir) + assert not os.path.exists(os.path.join(expected_data_dir, "scan_fake")) + + # Test case 3: FakeAcquisition only, make_scan_sub_dir=True + fake_acq2 = FakeAcquisition("non_sfdaq2", "p2") + fake_acq2.default_dir = str(tmp_path / "fake_default2") + sb = ScanBackend( + adjs, [[1]], [fake_acq2], + filename="scan_fake_subdir", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=True, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + sb.create_output_dirs() + expected_subdir = os.path.join(fake_acq2.default_dir, sb.data_base_dir, "scan_fake_subdir") + assert os.path.isdir(expected_subdir) + + # Test case 4: Mixed SFDAQ + FakeAcquisition, make_scan_sub_dir=False + fake_acq3 = FakeAcquisition("mix_non_sfdaq", "p3") + fake_acq3.default_dir = str(tmp_path / "mix_default") + sb = ScanBackend( + adjs, [[1]], [sfdaq_acq, fake_acq3], + filename="scan_mix_no_subdir", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=False, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + sb.create_output_dirs() + expected_data_dir_no_sub = os.path.join(fake_acq3.default_dir, sb.data_base_dir) + assert os.path.isdir(expected_data_dir_no_sub) + assert not os.path.exists(os.path.join(expected_data_dir_no_sub, "scan_mix_no_subdir")) + + # Test case 5: Mixed SFDAQ + FakeAcquisition, make_scan_sub_dir=True + fake_acq4 = FakeAcquisition("mix_non_sfdaq2", "p4") + fake_acq4.default_dir = str(tmp_path / "mix_default2") + sb = ScanBackend( + adjs, [[1]], [sfdaq_acq, fake_acq4], + filename="scan_mix_subdir", + detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=True, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + sb.create_output_dirs() + expected_subdir_mix = os.path.join(fake_acq4.default_dir, sb.data_base_dir, "scan_mix_subdir") + assert os.path.isdir(expected_subdir_mix) + sfdaq_dir = os.path.join("data", "sfdaq") + assert not os.path.exists(sfdaq_dir) + + +def test_store_and_change_initial_values_restores_correctly(tmp_path): + # Test that initial values are properly stored and restored + adjs = [ + DummyAdjustable(ID="A", initial_value=9, process_time=0), + DummyAdjustable(ID="B", initial_value=8, process_time=0) + ] + + sb = ScanBackend( + adjs, [[1]], [FakeAcquisition()], + filename="fn", detectors=[], channels=[], pvs=[], + n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path, + make_scan_sub_dir=True, condition=None, + return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + # Store initial values + sb.store_initial_values() + initial_values = [a.get_current_value() for a in adjs] + assert initial_values == [9, 8] + assert sb.initial_values == [9, 8] + + # Change values + for a, new_val in zip(adjs, [100, 200]): + a.set_target_value(new_val) + + changed_values = [a.get_current_value() for a in adjs] + assert changed_values == [100, 200] + + # Restore initial values + sb.change_to_initial_values() + + restored_values = [a.get_current_value() for a in adjs] + assert restored_values == [9, 8] + + +def test_acquire_all_with_fake_acquisitions(tmp_path): + # Test acquisition with multiple fake acquisitions + adjs = [DummyAdjustable(name="A", ID="A")] + + fake_acq1 = FakeAcquisition(name="fake_acq1") + fake_acq2 = FakeAcquisition(name="fake_acq2") + acqs = [fake_acq1, fake_acq2] + + sb = ScanBackend( + adjs, [[1]], acqs, "test_scan", + ["detector1"], ["bs_channel1"], ["pv1"], 3, + "data", tmp_path, True, None, True, 1, None, None + ) + + # Test acquire_all + filenames = sb.acquire_all("test_filename") + + # Verify tasks were created and stored + assert hasattr(sb, 'current_tasks') + assert len(sb.current_tasks) == 2 + + # Verify all tasks completed + assert all(t.done for t in sb.current_tasks) + + # Verify each fake acquisition was called with correct parameters + for acq in acqs: + assert acq.acquire_called + assert acq.last_filename == "test_filename" + assert acq.last_data_base_dir == "data" + assert acq.last_channels == ["bs_channel1"] + assert acq.last_n_pulses == 3 + assert acq.last_wait == False + + # Verify filenames returned + assert len(filenames) >= 2 + assert all(isinstance(fname, str) for fname in filenames) + assert all(len(fname) > 0 for fname in filenames) + + # Test stop cleans up properly + sb.stop() + assert not sb.running + assert sb.current_tasks == [] + + +def test_do_step_with_fake_acquisitions(tmp_path): + # Test individual scan step execution + adjs = [DummyAdjustable(name="motor1", ID="M1"), DummyAdjustable(name="motor2", ID="M2")] + + fake_acq = FakeAcquisition(name="fake_acq") + + sb = ScanBackend( + adjs, [[1, 2], [3, 4]], [fake_acq], "test_scan", + [], ["bs_channel"], [], 1, + "data", tmp_path, True, None, True, 1, None, None + ) + + # Test do_step for first step + step_values = [1, 3] + n_step = 0 + + sb.do_step(n_step, step_values) + + # Verify adjustables were moved to target values + current_values = [adj.get_current_value() for adj in adjs] + assert current_values == step_values + + # Verify scan_info_sfdaq was updated + assert hasattr(sb.scan_info_sfdaq, 'steps') + assert len(sb.scan_info_sfdaq.steps) > 0 + + # Verify acquisition was called with correct filename + expected_filename = sb.get_filename(n_step) + assert fake_acq.last_filename == expected_filename + + # Verify scan_info was updated + assert hasattr(sb.scan_info, 'steps') + assert len(sb.scan_info.steps) > 0 + + +def test_do_step_with_sensor_and_remote_plot(tmp_path): + # Test scan step with sensor and remote plot integration + adjs = [DummyAdjustable(name="motor", ID="M1")] + + fake_acq = FakeAcquisition(name="fake_acq") + dummy_sensor = DummySensor() + dummy_remote_plot = DummyRemotePlot() + + sb = ScanBackend( + adjs, [[1, 2, 3]], [fake_acq], "test_scan", + [], ["bs_channel"], [], 1, + "data", tmp_path, True, None, True, 1, dummy_sensor, dummy_remote_plot + ) + + step_values = [2] + n_step = 1 + + sb.do_step(n_step, step_values) + + # Verify sensor was started and stopped + assert dummy_sensor.started + assert dummy_sensor.stopped + + # Verify remote plot received data + assert dummy_remote_plot.appended + + # Verify data sent to plot matches current motor position and sensor reading + x_value = adjs[0].get_current_value() + y_value = dummy_sensor.get() + expected_data = (float(x_value), float(y_value)) + assert dummy_remote_plot.last_data == expected_data + + +def test_do_step_multiple_steps(tmp_path): + # Test multiple consecutive scan steps + adjs = [DummyAdjustable(name="motor", ID="M1")] + + fake_acq = FakeAcquisition(name="fake_acq") + + sb = ScanBackend( + adjs, [[1, 2, 3]], [fake_acq], "test_scan", + [], ["bs_channel"], [], 1, + "data", tmp_path, True, None, True, 1, None, None + ) + + # Test multiple steps + step_sequences = [ + (0, [1]), + (1, [2]), + (2, [3]) + ] + + for n_step, step_values in step_sequences: + fake_acq.reset() + + sb.do_step(n_step, step_values) + + # Verify motor moved to correct position + assert adjs[0].get_current_value() == step_values[0] + + # Verify acquisition was called with correct step filename + expected_filename = sb.get_filename(n_step) + assert fake_acq.last_filename == expected_filename + + # Verify scan info was updated for each step + assert len(sb.scan_info.steps) == n_step + 1 + + +def test_do_checked_step_with_condition_repeats(tmp_path): + # Test step execution with condition that requires repeats + adjs = [DummyAdjustable(name="motor", ID="M1")] + fake_acq = FakeAcquisition(name="fake_acq") + + # Condition that wants 2 repeats + condition = DummyCondition(repeats=2) + + sb = ScanBackend( + adjs, [[1, 2]], [fake_acq], "test_scan", + [], ["bs_channel"], [], 1, + "data", tmp_path, True, condition, True, 1, None, None + ) + + step_values = [1] + n_step = 0 + + # Track do_step calls directly + do_step_call_count = 0 + original_do_step = sb.do_step + + def mock_do_step(*args, **kwargs): + nonlocal do_step_call_count + do_step_call_count += 1 + return original_do_step(*args, **kwargs) + + sb.do_step = mock_do_step + sb.running = True + + sb.do_checked_step(n_step, step_values) + + # Verify do_step was called 3 times (original + 2 repeats) + assert do_step_call_count == 3 + + # Verify condition was checked + assert condition.repeats == -1 + + +def test_make_summary_and_repr(tmp_path): + # Test summary generation and string representation + adjs = [DummyAdjustable(name="A", ID="A")] + sb = ScanBackend( + adjs, [[1, 2]], [FakeAcquisition()], + "fn", [], [], [], 2, + "data", tmp_path, True, None, True, 2, None, None + ) + s = sb._make_summary() + assert "record" in s and "pulse" in s + assert isinstance(repr(sb), str) + + +def test_make_summary_single_repeat(tmp_path): + # Test summary generation with single repeat + adjs = [DummyAdjustable(name="motor1", ID="M1"), DummyAdjustable(name="motor2", ID="M2")] + fake_acq = FakeAcquisition(name="fake_daq") + + sb = ScanBackend( + adjs, [[1, 2], [3, 4]], [fake_acq], "test_scan", + [], ["bs_channel"], [], 5, + "data", tmp_path, True, None, True, 1, + None, None + ) + + summary = sb._make_summary() + + # Verify single repeat wording + assert "perform the following scan" in summary + + # Verify adjustable names are included + assert "motor1" in summary + assert "motor2" in summary + + # Verify pulse count + assert "5 pulses" in summary + + # Verify filename + assert "test_scan" in summary + + # Verify acquisition is mentioned + assert "fake_daq" in summary + + +def test_make_summary_multiple_repeats(tmp_path): + # Test summary generation with multiple repeats + adjs = [DummyAdjustable(name="motor", ID="M1")] + fake_acq = FakeAcquisition(name="daq_system") + + sb = ScanBackend( + adjs, [[1]], [fake_acq], "multi_scan", + [], ["bs_channel"], [], 1, + "data", tmp_path, True, None, True, 3, + None, None + ) + + summary = sb._make_summary() + + # Verify multiple repeats wording + assert "repeat the following scan 3 times" in summary + + # Verify pulse count + assert "1 pulse" in summary + + # Verify filename + assert "multi_scan" in summary + + +def test_scan_loop_fake_only(tmp_path, capsys): + # Test complete scan loop execution with fake acquisition + adjs = [ + DummyAdjustable(name="A", ID="A", initial_value=0), + DummyAdjustable(name="B", ID="B", initial_value=0), + ] + + fake = FakeAcquisition() + + values = [[1, 10], [2, 20], [3, 30]] + + sb = ScanBackend( + adjs, values, [fake], "scan1", + [], ["ch"], [], 1, + "data", tmp_path, False, + condition=None, return_to_initial_values=True, n_repeat=1, + sensor=None, remote_plot=None + ) + + sb.running = True + sb.scan_loop() + + out = capsys.readouterr().out + + # Verify steps printed + assert "Scan step 1 of 3" in out + assert "Scan step 2 of 3" in out + assert "Scan step 3 of 3" in out + assert "All scan steps done" in out + + # Verify acquisitions done + assert fake.call_count == 3 + + # Verify final positions + assert adjs[0].get_current_value() == 3 + assert adjs[1].get_current_value() == 30 + + +def test_repeated_scan_loop_fake_only(tmp_path, capsys): + # Test repeated scan loop execution + adjs = [DummyAdjustable(name="A", ID="A")] + fake = FakeAcquisition() + + values = [[1], [2]] + + sb = ScanBackend( + adjs, values, [fake], "rscan", + [], ["ch"], [], 1, + "data", tmp_path, False, + condition=None, + return_to_initial_values=True, n_repeat=3, + sensor=None, remote_plot=None + ) + + sb.running = True + sb.repeated_scan_loop() + + out = capsys.readouterr().out + + assert "Repetition 1 of 3" in out + assert "Repetition 2 of 3" in out + assert "Repetition 3 of 3" in out + + # Verify fake acquisitions: 3 reps * 2 steps = 6 + assert fake.call_count == 6 + + # Verify filename restored + assert sb.filename == "rscan" + + +def test_full_multidimensional_scan_end_to_end(tmp_path): + # Comprehensive test of multi-dimensional scan with verification of all components + d = 4 # scan dimension + n_steps = 5 # number of steps + + # Create d adjustables + adjs = [ + DummyAdjustable(name=f"M{i}", ID=f"ID{i}", initial_value=0, process_time=0) + for i in range(d) + ] + + # Create values + values = [ + list(range(t, t + d)) + for t in range(1, n_steps + 1) + ] + + # Create fake acquisition + fake = FakeAcquisition(name="fake_master") + + # Create ScanBackend + sb = ScanBackend( + adjs, values, [fake], + filename="multidim_test", + detectors=[], channels=["ch"], pvs=[], + n_pulses=3, + data_base_dir="data", + scan_info_dir=tmp_path, + make_scan_sub_dir=True, + condition=None, + return_to_initial_values=True, + n_repeat=1, + sensor=None, + remote_plot=None + ) + + # Execute scan + sb.run() + + # Verification checks + + # 1. All acquisitions done + assert fake.call_count == n_steps + + # 2. Adjustables returned to initial values + assert [a.get_current_value() for a in adjs] == [0] * d + + # 3. Scan info has correct number of steps + assert len(sb.scan_info.steps) == n_steps + assert len(sb.scan_info_sfdaq.steps) == n_steps + + # 4. Each step has correct values + for i in range(n_steps): + step_values = sb.scan_info.steps[i]["target_values"] + assert step_values == values[i] + + # 5. Filenames are coherent + base = sb.filename + filebase = os.path.basename(base) + for i in range(n_steps): + expected = os.path.join(base, filebase + f"_step{i:04d}") + assert expected.endswith(f"{filebase}_step{i:04d}") + + # 6. Directories created + data_root = tmp_path / "fake_master" / "data" + assert data_root.exists() + + expected_subfolder = data_root / "multidim_test" + assert expected_subfolder.exists() + + # 7. All tasks finished + assert all(t.done for t in sb.current_tasks) + + # 8. Each task has at least one file + all_files = [] + for t in sb.current_tasks: + assert len(t.filenames) > 0 + all_files += t.filenames + + assert len(all_files) >= n_steps + + # 9. Intermediate values correctly positioned + for i in range(n_steps): + target = values[i] + readback = sb.scan_info_sfdaq.steps[i]["readback_values"] + assert readback == target + + # 10. Step order respected + assert sb.scan_info.steps[0]["target_values"] == values[0] + assert sb.scan_info.steps[-1]["target_values"] == values[-1] + + +def test_scanND_relative_positions_only(tmp_path): + # Test relative position calculation for N-dimensional scans + d = 5 + + # Create adjustables with distinct initial values + initial_values = [10 * (i + 1) for i in range(d)] + adjustables = [ + DummyAdjustable(name=f"M{i}", ID=f"ID{i}", initial_value=initial_values[i], process_time=0) + for i in range(d) + ] + + # Create N-dim grid + positions_per_dim = [list(range(-1, 2))] * d + + fake = FakeAcquisition(name="fake") + + # Create ScanBackend + sb = ScanBackend( + adjustables, + values=positions_per_dim, + acquisitions=[fake], + filename="relND", + detectors=[], channels=["ch"], pvs=[], + n_pulses=1, + data_base_dir="data", + scan_info_dir=tmp_path, + make_scan_sub_dir=False, + condition=None, + return_to_initial_values=True, + n_repeat=1, + sensor=None, remote_plot=None + ) + + # Simulate relative position calculation + offset_values = [ + [p + initial_values[i] for p in [-1, 0, 1]] + for i in range(d) + ] + + # Apply relative offset + sb.values = offset_values + + # Verification + + # Check each dimension is correctly offset + for i in range(d): + expected = [ + initial_values[i] - 1, + initial_values[i], + initial_values[i] + 1, + ] + assert sb.values[i] == expected + + # Verify dimension count + assert len(sb.values) == d + + # Verify each dimension has 3 positions + assert all(len(axis) == 3 for axis in sb.values) + + +# Utility function tests + +def test_print_current_values_displays_correct_output(capsys): + # Test current values printing functionality + class Obj: + def __init__(self, adjustables): + self.adjustables = adjustables + + def print_current_values(self): + print_all_current_values(self.adjustables) + + adjs = [ + DummyAdjustable(name="MotorA", ID="A1", value=10), + DummyAdjustable(name="MotorB", ID="B2", value=20), + ] + + obj = Obj(adjs) + obj.print_current_values() + + captured = capsys.readouterr().out + + assert "Current values" in captured + assert "A1" in captured and "B2" in captured + assert "10" in captured and "20" in captured + + +def test_get_all_current_values_returns_correct_list(): + # Test current values retrieval + adjs = [ + DummyAdjustable(name="M1", ID="M1", value=5), + DummyAdjustable(name="M2", ID="M2", value=15), + ] + values = get_all_current_values(adjs) + assert values == [5, 15] + + +def test_set_all_target_values_and_wait_full_chain(monkeypatch): + # Test complete chain of setting target values and waiting + adjs = [ + DummyAdjustable(ID="1", initial_value=0, process_time=0), + DummyAdjustable(ID="2", initial_value=0, process_time=0), + ] + + called = [] + + class DummyTask: + def __init__(self, name): + self.name = name + def wait(self): + called.append(self.name) + + # Monkeypatch set_all_target_values + def fake_set_all_target_values(adjs, values): + for adj, v in zip(adjs, values): + adj.set_target_value(v) + return [DummyTask(f"task_{a.ID}") for a in adjs] + + monkeypatch.setattr("slic.core.scanner.scanbackend.set_all_target_values", fake_set_all_target_values) + + # Execute + tasks = set_all_target_values(adjs, [10, 20]) + wait_for_all(tasks) + + # Verify values were updated + values = [a.get_current_value() for a in adjs] + assert values == [10, 20] + + # Verify all tasks were waited for + assert called == ["task_1", "task_2"] + + # Verify returned objects have wait method + assert all(hasattr(t, "wait") for t in tasks) + + +def test_wait_for_all_calls_wait_on_all_tasks(monkeypatch): + # Test wait functionality on all tasks + called = [] + + class DummyTask: + def __init__(self, name): + self.name = name + def wait(self): + called.append(self.name) + + tasks = [DummyTask("t1"), DummyTask("t2"), DummyTask("t3")] + + wait_for_all(tasks) + + assert called == ["t1", "t2", "t3"] + + +def test_stop_all_calls_stop_and_handles_exceptions(capsys): + # Test stop functionality with exception handling + called = [] + + class WorkingTask: + def __init__(self, name): + self.name = name + self.stopped = False + def stop(self): + self.stopped = True + called.append(self.name) + + class FailingTask: + def stop(self): + raise RuntimeError("boom") + + tasks = [WorkingTask("T1"), FailingTask(), WorkingTask("T2")] + + stop_all(tasks) + + # Verify normal tasks were stopped + assert all(t.stopped for t in tasks if isinstance(t, WorkingTask)) + + # Verify task names recorded + assert called == ["T1", "T2"] + + # Verify error was handled and printed + out = capsys.readouterr().out + assert "Stopping caused" in out + assert "boom" in out \ No newline at end of file