From 77e3cafc4789bae7dcdbb03e52ac3062b39b5ce1 Mon Sep 17 00:00:00 2001 From: tligui_y Date: Wed, 21 Jan 2026 19:14:18 +0100 Subject: [PATCH] Delete tests/test_core_scanbackend.py --- tests/test_core_scanbackend.py | 852 --------------------------------- 1 file changed, 852 deletions(-) delete mode 100644 tests/test_core_scanbackend.py diff --git a/tests/test_core_scanbackend.py b/tests/test_core_scanbackend.py deleted file mode 100644 index e7901f5a..00000000 --- a/tests/test_core_scanbackend.py +++ /dev/null @@ -1,852 +0,0 @@ -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