# SPDX-FileCopyrightText: 2022-present deepset GmbH # # SPDX-License-Identifier: Apache-2.0 import warnings import pytest from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState def test_pipeline_state_no_warning_on_init(): with warnings.catch_warnings(): warnings.simplefilter("error", Warning) PipelineState(inputs={}, component_visits={}, pipeline_outputs={}) def test_pipeline_state_warn_on_inplace_mutation(): state = PipelineState(inputs={}, component_visits={}, pipeline_outputs={}) with pytest.warns(Warning, match="dataclasses.replace"): state.inputs = {"new": "value"} def test_pipeline_snapshot_no_warning_on_init(): state = PipelineState(inputs={}, component_visits={"comp": 1}, pipeline_outputs={}) bp = Breakpoint(component_name="comp") with warnings.catch_warnings(): warnings.simplefilter("error", Warning) PipelineSnapshot(original_input_data={}, ordered_component_names=["comp"], pipeline_state=state, break_point=bp) def test_pipeline_snapshot_warn_on_inplace_mutation(): state = PipelineState(inputs={}, component_visits={"comp": 1}, pipeline_outputs={}) bp = Breakpoint(component_name="comp") snap = PipelineSnapshot( original_input_data={}, ordered_component_names=["comp"], pipeline_state=state, break_point=bp ) with pytest.warns(Warning, match="dataclasses.replace"): snap.original_input_data = {"new": "data"}