diff --git a/src/pyinfra/api/operation.py b/src/pyinfra/api/operation.py index 57ebe2804..8bbe1e65f 100644 --- a/src/pyinfra/api/operation.py +++ b/src/pyinfra/api/operation.py @@ -293,7 +293,7 @@ def decorated_func(*args: P.args, **kwargs: P.kwargs) -> OperationMeta: global_arguments, global_argument_keys = pop_global_arguments(state, host, kwargs) names, add_args = generate_operation_name(func, host, kwargs, global_arguments) - op_order, op_hash = solve_operation_consistency(names, state, host) + op_order, op_hash = solve_operation_consistency(names, args, kwargs, state, host) # Ensure shared (between servers) operation meta, mutates state op_meta = ensure_shared_op_meta(state, op_hash, op_order, global_arguments, names) @@ -410,7 +410,7 @@ def generate_operation_name(func, host, kwargs, global_arguments): return names, add_args -def solve_operation_consistency(names, state, host): +def solve_operation_consistency(names, args, kwargs, state, host): # Operation order is used to tie-break available nodes in the operation DAG, in CLI mode # we use stack call order so this matches as defined by the user deploy code. if pyinfra.is_cli: @@ -422,8 +422,18 @@ def solve_operation_consistency(names, state, host): if host.loop_position: op_order.extend(host.loop_position) - # Make a hash from the call stack lines - op_hash = make_hash(op_order) + # Make a hash from the call stack lines, operation names and arguments. Names and + # arguments are included so that different operations at the same position (eg calls + # in a plain loop with per-host data) do not collide into one hash, while identical + # operations across hosts still share a hash. + op_hash = make_hash( + ( + op_order, + sorted(names), + [_get_arg_value(arg) for arg in args], + {key: _get_arg_value(value) for key, value in kwargs.items()}, + ) + ) # Avoid adding duplicates! This happens if an operation is called within # a loop - such that the filename/lineno/code _are_ the same, but the diff --git a/tests/test_api/test_api_operations.py b/tests/test_api/test_api_operations.py index cb6f526f5..0a0c8d8cb 100644 --- a/tests/test_api/test_api_operations.py +++ b/tests/test_api/test_api_operations.py @@ -118,6 +118,119 @@ def test_op(self): disconnect_all(state) + def test_op_per_host_data_no_hash_collision(self): + inventory = make_inventory( + hosts=( + ("somehost", {"users": ["user-1"]}), + ("anotherhost", {"users": ["user-5"]}), + ), + ) + somehost = inventory.get_host("somehost") + anotherhost = inventory.get_host("anotherhost") + + state = State(inventory, Config()) + state.current_stage = StateStage.Prepare + + # Call the "same" operation (identical position) with per-host names and + # arguments - these must not collapse into a single operation hash (#1370). + for host in inventory: + for user in host.data.users: + add_op( + state, + server.shell, + commands=[f"echo {user}"], + name=f"Create {user} user", + host=host, + ) + + op_order = state.get_op_order() + assert len(op_order) == 2 + + names_by_host = { + host_name: { + name for op_hash in state.ops[host] for name in state.op_meta[op_hash].names + } + for host_name, host in (("somehost", somehost), ("anotherhost", anotherhost)) + } + assert names_by_host == { + "somehost": {"Create user-1 user"}, + "anotherhost": {"Create user-5 user"}, + } + + # Identical operations on both hosts must still share a single hash + add_op(state, server.shell, commands=["echo same"], host=[somehost, anotherhost]) + + shared_op_hash = state.get_op_order()[-1] + assert shared_op_hash in state.ops[somehost] + assert shared_op_hash in state.ops[anotherhost] + + def test_op_per_host_args_no_name_no_hash_collision(self): + inventory = make_inventory( + hosts=( + ("somehost", {"users": ["user-1"]}), + ("anotherhost", {"users": ["user-5"]}), + ), + ) + somehost = inventory.get_host("somehost") + anotherhost = inventory.get_host("anotherhost") + + state = State(inventory, Config()) + state.current_stage = StateStage.Prepare + + # Same operation and position with per-host arguments but no explicit + # name - these must not collapse into a single operation hash (#1370). + for host in inventory: + for user in host.data.users: + add_op(state, server.shell, commands=[f"echo {user}"], host=host) + + op_order = state.get_op_order() + assert len(op_order) == 2 + + somehost_op_hash = next(iter(state.ops[somehost])) + anotherhost_op_hash = next(iter(state.ops[anotherhost])) + assert somehost_op_hash != anotherhost_op_hash + + # Both auto-named the same, but the attached arguments are kept separate + assert state.op_meta[somehost_op_hash].names == {"server.shell"} + assert state.op_meta[anotherhost_op_hash].names == {"server.shell"} + assert state.op_meta[somehost_op_hash].args == ["commands=['echo user-1']"] + assert state.op_meta[anotherhost_op_hash].args == ["commands=['echo user-5']"] + + def test_op_per_host_data_different_execution_kwargs(self): + inventory = make_inventory() + somehost = inventory.get_host("somehost") + anotherhost = inventory.get_host("anotherhost") + + state = State(inventory, Config()) + state.current_stage = StateStage.Prepare + + # Distinct operations at the same position with different execution kwargs + # must not raise - they only shared a hash by accident before #1370. + add_op( + state, + server.shell, + commands=["echo somehost"], + name="somehost op", + _parallel=1, + host=somehost, + ) + add_op( + state, + server.shell, + commands=["echo anotherhost"], + name="anotherhost op", + _parallel=2, + host=anotherhost, + ) + + op_order = state.get_op_order() + assert len(op_order) == 2 + + somehost_op_hash = next(iter(state.ops[somehost])) + anotherhost_op_hash = next(iter(state.ops[anotherhost])) + assert state.op_meta[somehost_op_hash].global_arguments["_parallel"] == 1 + assert state.op_meta[anotherhost_op_hash].global_arguments["_parallel"] == 2 + @patch("pyinfra.api.util.open", mock_open(read_data="test!"), create=True) @patch("pyinfra.operations.files.Path.is_file", lambda *args, **kwargs: True) def test_file_upload_op(self): @@ -762,6 +875,84 @@ def test_cli_op_line_numbers(self): assert len(state.ops[inventory.get_host("somehost")]) == 2 assert len(state.ops[inventory.get_host("anotherhost")]) == 2 + # Operations called in a plain loop share the same stack position, so their + # hash must come from the name & arguments - otherwise per-host data makes + # different operations collapse into one (#1370). + def test_cli_op_loop_per_host_data_no_hash_collision(self): + inventory = make_inventory( + hosts=( + ("somehost", {"users": ["user-1", "user-2"]}), + ("anotherhost", {"users": ["user-5"]}), + ), + ) + somehost = inventory.get_host("somehost") + anotherhost = inventory.get_host("anotherhost") + + state = State(inventory, Config()) + state.current_stage = StateStage.Prepare + connect_all(state) + + state.current_deploy_filename = __file__ + + pyinfra.is_cli = True + + try: + with ctx_state.use(state): + for name in ("somehost", "anotherhost"): + host = inventory.get_host(name) + with ctx_host.use(host): + for user in host.data.users: + server.shell( # called on *the same line* for every host/user + name=f"Create {user} user", + commands=[f"echo {user}"], + ) + finally: + ctx_state.reset() + ctx_host.reset() + pyinfra.is_cli = False + + op_order = state.get_op_order() + assert len(op_order) == 3 + + names_by_host = { + host_name: { + name for op_hash in state.ops[host] for name in state.op_meta[op_hash].names + } + for host_name, host in (("somehost", somehost), ("anotherhost", anotherhost)) + } + assert names_by_host == { + "somehost": {"Create user-1 user", "Create user-2 user"}, + "anotherhost": {"Create user-5 user"}, + } + + # Identical operations called in a loop (same position, same name & arguments) + # must still be deduplicated by appending a counter to the hash. + def test_cli_op_loop_identical_ops_dedupe(self): + inventory = make_inventory(hosts=("somehost",)) + + state = State(inventory, Config()) + state.current_stage = StateStage.Prepare + connect_all(state) + + state.current_deploy_filename = __file__ + + pyinfra.is_cli = True + + try: + with ctx_state.use(state): + with ctx_host.use(inventory.get_host("somehost")): + for _ in range(3): + server.shell(commands=["echo same"]) # called on *the same line* + finally: + ctx_state.reset() + ctx_host.reset() + pyinfra.is_cli = False + + op_order = state.get_op_order() + assert len(op_order) == 3 + assert op_order[1] == f"{op_order[0]}-0" + assert op_order[2] == f"{op_order[0]}-0-1" + # In API mode, pyinfra *overrides* the line numbers such that whenever an # operation or deploy is added it is simply appended. This makes sense as # the user writing the API calls has full control over execution order.