115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
import pytest
|
|
|
|
from SYS import pipeline as ctx
|
|
|
|
|
|
def test_new_pipeline_state_isolation():
|
|
# Ensure we start with a clean baseline
|
|
ctx.reset()
|
|
ctx.set_last_items(["global"])
|
|
assert ctx.get_last_items() == ["global"]
|
|
|
|
with ctx.new_pipeline_state():
|
|
# New state should be fresh
|
|
assert ctx.get_last_items() == []
|
|
ctx.set_last_items(["inner1"])
|
|
assert ctx.get_last_items() == ["inner1"]
|
|
|
|
with ctx.new_pipeline_state():
|
|
# Nested fresh state
|
|
assert ctx.get_last_items() == []
|
|
ctx.set_last_items(["inner2"])
|
|
assert ctx.get_last_items() == ["inner2"]
|
|
|
|
# After exiting inner, outer new state should still have its values
|
|
assert ctx.get_last_items() == ["inner1"]
|
|
|
|
# After all contexts exit, global state remains unchanged
|
|
assert ctx.get_last_items() == ["global"]
|
|
|
|
|
|
def test_display_overlay_priority_and_restore():
|
|
ctx.reset()
|
|
|
|
# Set a base last result table and items
|
|
ctx.set_last_result_table("table-a", ["a1", "a2"], subject="sub-a")
|
|
assert ctx.get_last_result_items() == ["a1", "a2"]
|
|
|
|
# Overlay should take precedence for selection
|
|
ctx.set_last_result_table_overlay("table-b", ["b1"], subject="sub-b")
|
|
assert ctx.get_last_result_items() == ["b1"]
|
|
assert ctx.get_last_result_subject() == "sub-b"
|
|
|
|
# Restoring previous should revert to base
|
|
assert ctx.restore_previous_result_table() is True
|
|
assert ctx.get_last_result_items() == ["a1", "a2"]
|
|
assert ctx.get_last_result_subject() == "sub-a"
|
|
|
|
|
|
def test_pipeline_runner_snapshot_restore():
|
|
from TUI.pipeline_runner import PipelineRunner
|
|
|
|
runner = PipelineRunner()
|
|
ctx.reset()
|
|
|
|
# Seed state with a few values
|
|
ctx.set_last_result_table("table-x", ["x1", "x2"], subject="sx")
|
|
ctx.set_last_items(["L1"])
|
|
ctx.set_current_cmdlet_name("cmd-name")
|
|
ctx.set_pending_pipeline_tail([["stage1"], ["stage2"]], source_command="source")
|
|
|
|
snap = runner._snapshot_ctx_state()
|
|
|
|
# Mutate state
|
|
ctx.clear_last_result()
|
|
ctx.set_last_items([])
|
|
ctx.set_current_cmdlet_name("other")
|
|
ctx.clear_pending_pipeline_tail()
|
|
|
|
# Restore and verify
|
|
runner._restore_ctx_state(snap)
|
|
assert ctx.get_last_result_items() == ["x1", "x2"]
|
|
assert ctx.get_last_items() == ["L1"]
|
|
assert ctx.get_current_cmdlet_name() == "cmd-name"
|
|
assert ctx.get_pending_pipeline_tail() == [["stage1"], ["stage2"]]
|
|
|
|
|
|
def test_preserve_history_no_push():
|
|
ctx.reset()
|
|
|
|
ctx.set_last_result_table("t1", ["a"], subject="s1")
|
|
# Push second table
|
|
ctx.set_last_result_table("t2", ["b"], subject="s2")
|
|
state = ctx.get_pipeline_state()
|
|
assert len(state.result_table_history) == 1
|
|
|
|
# Preserve history update should NOT push a new history entry
|
|
ctx.set_last_result_table_preserve_history("t2b", ["c"], subject="s2b")
|
|
assert len(state.result_table_history) == 1
|
|
assert ctx.get_last_result_items() == ["c"]
|
|
|
|
|
|
def test_forward_and_restore():
|
|
ctx.reset()
|
|
|
|
ctx.set_last_result_table("t1", ["a"], subject="s1")
|
|
ctx.set_last_result_table("t2", ["b"], subject="s2")
|
|
|
|
assert ctx.restore_previous_result_table() is True
|
|
assert ctx.get_last_result_items() == ["a"]
|
|
|
|
assert ctx.restore_next_result_table() is True
|
|
assert ctx.get_last_result_items() == ["b"]
|
|
|
|
|
|
def test_display_items_priority():
|
|
ctx.reset()
|
|
|
|
ctx.set_last_result_table("base", ["x1", "x2"])
|
|
ctx.set_last_result_items_only(["d1", "d2"])
|
|
|
|
# display items take priority for selection
|
|
assert ctx.get_last_result_items() == ["d1", "d2"]
|
|
# but selectable items should ignore display-only items
|
|
assert ctx.get_last_selectable_result_items() == ["x1", "x2"]
|