fix: make fact conditions explicit and reject bare fact classes

docs/using-operations.md

@@ -242,6 +242,31 @@ server.shell(commands=["..."], _if=all_changed(create_user, create_otheruser))
 !!! important
     `_if` must be a callable, or a list of callables. Passing a value directly (e.g. `_if=host.get_fact(MyFact)`) does not gate the operation: most non-callable values raise `ArgumentTypeError` at prepare time, and `None` is treated as "no condition" so the operation always runs. Wrap the value in a lambda to gate on it: `_if=lambda: bool(host.get_fact(MyFact))`.
 
+Use `any_fact` (OR) or `all_fact` (AND) to gate an operation on the truthiness of facts that take no arguments:
+
+```python
+from pyinfra import host
+from pyinfra.facts.server import HasGui, LinuxName, Which
+from pyinfra.operations.util import any_fact, all_fact
+
+server.shell(commands=["..."], _if=any_fact(HasGui, LinuxName))
+server.shell(commands=["..."], _if=all_fact(HasGui, LinuxName))
+# Lists keep their AND semantics and can mix fact and operation conditions.
+server.shell(commands=["..."], _if=[any_fact(HasGui, LinuxName), create_user.did_change])
+
+# Facts that need arguments use an explicit lambda.
+server.shell(commands=["..."], _if=lambda: bool(host.get_fact(Which, "git")))
+```
+
+These helpers collect facts lazily when the condition runs, using the current executing host.
+They short-circuit: `any_fact` stops at the first truthy value, and `all_fact` stops at the
+first falsey value. Each invocation reads facts through the existing fact API.
+With no facts, `any_fact()` returns a condition that is always false and `all_fact()` returns
+one that is always true. An empty `_if` list also allows the operation to run.
+
+Passing a bare fact class, such as `_if=HasGui` or `_if=[HasGui]`, raises `ArgumentTypeError`
+during prepare. Wrap it with `any_fact` or `all_fact`, or use a lambda as above.
+
 ## Output & Callbacks
 
 pyinfra doesn't immediately execute operations, meaning output is not available right away. It is possible to access this output at runtime by providing a callback function using the [`python.call`](operations/python.md#python-call) operation. Callback functions may also call other operations which will be immediately executed.

src/pyinfra/api/operation.py

@@ -23,7 +23,8 @@
 from .arguments import EXECUTION_KWARG_KEYS, AllArguments, pop_global_arguments
 from .arguments_typed import PyinfraOperation
 from .command import PyinfraCommand, StringCommand
-from .exceptions import NestedOperationError, OperationValueError, PyinfraError
+from .exceptions import ArgumentTypeError, NestedOperationError, OperationValueError, PyinfraError
+from .facts import FactBase, ShortFactBase
 from .host import Host
 from .operations import run_host_op
 from .state import State, StateOperationHostData, StateOperationMeta, StateStage
@@ -298,6 +299,15 @@ def decorated_func(*args: P.args, **kwargs: P.kwargs) -> OperationMeta:
         # Get the meta kwargs (globals that apply to all hosts)
         global_arguments, global_argument_keys = pop_global_arguments(state, host, kwargs)
 
+        conditions = global_arguments.get("_if")
+        for condition in conditions if isinstance(conditions, list) else (conditions,):
+            if isinstance(condition, type) and issubclass(condition, (FactBase, ShortFactBase)):
+                raise ArgumentTypeError(
+                    "Invalid argument `_if`: bare fact classes are not conditions. "
+                    "Use any_fact(MyFact), all_fact(MyFact), or "
+                    "lambda: bool(host.get_fact(MyFact, ...)) for facts with arguments."
+                )
+
         names, add_args = generate_operation_name(func, host, kwargs, global_arguments)
         op_order, op_hash = solve_operation_consistency(names, args, kwargs, state, host)
 

src/pyinfra/operations/util/__init__.py

@@ -1,7 +1,10 @@
 from typing import TYPE_CHECKING
 from collections.abc import Callable
 
+from pyinfra.context import host
+
 if TYPE_CHECKING:
+    from pyinfra.api.facts import FactBase, ShortFactBase
     from pyinfra.api.operation import OperationMeta
 
 
@@ -11,3 +14,13 @@ def any_changed(*args: "OperationMeta") -> Callable[[], bool]:
 
 def all_changed(*args: "OperationMeta") -> Callable[[], bool]:
     return lambda: all(meta.did_change() for meta in args)
+
+
+def any_fact(*facts: type["FactBase"] | type["ShortFactBase"]) -> Callable[[], bool]:
+    """Return a condition that is true when any zero-argument fact is truthy."""
+    return lambda: any(host.get_fact(fact) for fact in facts)
+
+
+def all_fact(*facts: type["FactBase"] | type["ShortFactBase"]) -> Callable[[], bool]:
+    """Return a condition that is true when all zero-argument facts are truthy."""
+    return lambda: all(host.get_fact(fact) for fact in facts)

tests/test_api/test_api_operations.py

@@ -2,7 +2,7 @@
 from io import StringIO
 from os import path
 from unittest import TestCase
-from unittest.mock import mock_open, patch
+from unittest.mock import MagicMock, mock_open, patch
 import time
 
 import pyinfra
@@ -17,13 +17,15 @@
     StringCommand,
 )
 from pyinfra.api.connect import connect_all, disconnect_all
-from pyinfra.api.exceptions import PyinfraError
+from pyinfra.api.exceptions import ArgumentTypeError, PyinfraError
 from pyinfra.api.operation import OperationMeta, add_op
 from pyinfra.api.operations import run_ops
 from pyinfra.api.state import StateOperationMeta, StateStage
 from pyinfra.connectors.util import CommandOutput, OutputLine
 from pyinfra.context import ctx_host, ctx_state
+from pyinfra.facts.server import Hostname, LinuxName, Which
 from pyinfra.operations import files, python, server
+from pyinfra.operations.util import all_fact, any_fact
 
 from ..paramiko_util import FakeBuffer, FakeChannel, PatchSSHTestCase
 from ..util import make_inventory
@@ -40,6 +42,93 @@ def test_operation_meta_repr_changes(self):
 
 
 class TestOperationsApi(PatchSSHTestCase):
+    def test_callable_conditions_remain_supported(self):
+        inventory = make_inventory(hosts=("somehost",))
+        state = State(inventory, Config())
+        state.current_stage = StateStage.Prepare
+        connect_all(state)
+        host = inventory.get_host("somehost")
+        with (
+            patch.object(host, "get_fact", return_value="/bin/tool") as get_fact,
+            patch("pyinfra.connectors.ssh.SSHConnector.run_shell_command") as run_command,
+        ):
+            run_command.return_value = (True, FakeBuffer("", FakeChannel(0)))
+            first = add_op(state, server.shell, "echo first")[host]
+            for condition in (
+                None,
+                [],
+                lambda: True,
+                MagicMock(return_value=True),
+                first.did_change,
+                lambda: bool(pyinfra.host.get_fact(Which, "tool")),
+                [lambda: True, lambda: False],
+            ):
+                add_op(state, server.shell, "echo test", _if=condition)
+            get_fact.assert_not_called()
+            run_ops(state)
+        assert run_command.call_count == 7
+        get_fact.assert_called_once_with(Which, "tool")
+
+    def test_bare_fact_conditions_rejected_before_registration(self):
+        for fact in (Hostname, LinuxName):
+            for condition in (fact, [fact], [lambda: True, fact]):
+                for source in ("operation", "deploy", "host"):
+                    with self.subTest(fact=fact, condition=condition, source=source):
+                        inventory = make_inventory(hosts=("somehost",))
+                        state = State(inventory, Config())
+                        state.current_stage = StateStage.Prepare
+                        host = inventory.get_host("somehost")
+                        kwargs = {}
+                        if source == "operation":
+                            kwargs["_if"] = condition
+                        elif source == "deploy":
+                            host.current_deploy_kwargs = {"_if": condition}
+                        else:
+                            host.data._if = condition
+                        with patch.object(host, "run_shell_command") as run_command:
+                            with self.assertRaisesRegex(
+                                ArgumentTypeError, "any_fact.*all_fact.*lambda"
+                            ):
+                                add_op(state, server.shell, "echo test", host=host, **kwargs)
+                        assert not state.op_meta
+                        assert not state.ops[host]
+                        run_command.assert_not_called()
+
+    def test_fact_conditions_execute_with_current_host(self):
+        for factory in (any_fact, all_fact):
+            for as_list in (False, True):
+                for check_for_changes in (False, True):
+                    with self.subTest(factory=factory, as_list=as_list, check=check_for_changes):
+                        condition = factory(Hostname, LinuxName)
+                        inventory = make_inventory()
+                        state = State(inventory, Config(), check_for_changes=check_for_changes)
+                        state.current_stage = StateStage.Prepare
+                        connect_all(state)
+                        somehost = inventory.get_host("somehost")
+                        anotherhost = inventory.get_host("anotherhost")
+                        with (
+                            patch.object(somehost, "get_fact", return_value=None) as false_fact,
+                            patch.object(
+                                anotherhost, "get_fact", return_value="Linux"
+                            ) as true_fact,
+                            patch(
+                                "pyinfra.connectors.ssh.SSHConnector.run_shell_command"
+                            ) as run_command,
+                        ):
+                            run_command.return_value = (True, FakeBuffer("", FakeChannel(0)))
+                            results = add_op(
+                                state,
+                                server.shell,
+                                "echo test",
+                                _if=[condition] if as_list else condition,
+                            )
+                            false_fact.assert_not_called()
+                            true_fact.assert_not_called()
+                            run_ops(state)
+                        assert run_command.call_count == 1
+                        assert results[somehost].did_change() is False
+                        assert results[anotherhost].did_change() is True
+
     def test_op(self):
         inventory = make_inventory()
         somehost = inventory.get_host("somehost")

tests/test_operations_utils.py

@@ -3,12 +3,39 @@
 
 import pytest
 
+from pyinfra.context import ctx_host
+from pyinfra.facts.server import Hostname, LinuxName
 from pyinfra.facts.util.packages import PackageInfo, PackageStatus
+from pyinfra.operations.util import all_fact, any_fact
 from pyinfra.operations.util.docker import parse_image_reference, parse_registry
 from pyinfra.operations.util.files import ensure_mode_int, unix_path_join
 from pyinfra.operations.util.packaging import ensure_packages
 
 
+@pytest.mark.parametrize("factory", [any_fact, all_fact])
+def test_fact_conditions(factory):
+    condition = factory(Hostname, LinuxName)
+    for values in [(None, []), ([], "found"), ({"found": True}, 0), (1, ["found"])]:
+        host = MagicMock()
+        host.get_fact.side_effect = values
+        host.get_fact.assert_not_called()
+        expected = any(values) if factory is any_fact else all(values)
+        with ctx_host.use(host):
+            assert condition() is expected
+        short_circuited = bool(values[0]) if factory is any_fact else not values[0]
+        assert host.get_fact.call_count == (1 if short_circuited else 2)
+        host.get_fact.assert_any_call(Hostname)
+        host.get_fact.reset_mock(side_effect=True)
+        host.get_fact.side_effect = values
+        with ctx_host.use(host):
+            assert condition() is expected
+    assert factory()() is (factory is all_fact)
+    with ctx_host.use(host):
+        host.get_fact.side_effect = ValueError("fact failed")
+        with pytest.raises(ValueError, match="fact failed"):
+            condition()
+
+
 class TestUnixPathJoin(TestCase):
     def test_simple_path(self):
         assert unix_path_join("home", "pyinfra") == "home/pyinfra"