How to use the pyinfra.api.operation.add_op function in pyinfra

def test_op_hosts_limit(self):
        inventory = make_inventory()
        state = State(inventory, Config())
        connect_all(state)

        # Add op to both hosts
        add_op(state, server.shell, 'echo "hi"')

        # Add op to just the first host
        add_op(
            state, server.user,
            'somehost_user',
            hosts=inventory['somehost'],
        )

        # Ensure there are two ops
        self.assertEqual(len(state.get_op_order()), 2)

        # Ensure somehost has two ops and anotherhost only has the one
        self.assertEqual(len(state.ops[inventory.get_host('somehost')]), 2)
        self.assertEqual(len(state.ops[inventory.get_host('anotherhost')]), 1)

def test_op_line_numbers(self):
        inventory = make_inventory()
        state = State(inventory, Config())
        connect_all(state)

        # Add op to both hosts
        add_op(state, server.shell, 'echo "hi"')

        # Add op to just the second host - using the pseudo modules such that
        # it replicates a deploy file.
        pseudo_state.set(state)
        pseudo_host.set(inventory['anotherhost'])
        first_pseudo_hash = server.user('anotherhost_user').hash
        first_pseudo_call_line = getframeinfo(currentframe()).lineno - 1

        # Add op to just the first host - using the pseudo modules such that
        # it replicates a deploy file.
        pseudo_state.set(state)
        pseudo_host.set(inventory['somehost'])
        second_pseudo_hash = server.user('somehost_user').hash
        second_pseudo_call_line = getframeinfo(currentframe()).lineno - 1

        pseudo_state.reset()

def test_run_once_serial_op(self):
        inventory = make_inventory()
        state = State(inventory, Config())
        connect_all(state)

        # Add a run once op
        add_op(state, server.shell, 'echo "hi"', run_once=True, serial=True)

        # Ensure it's added to op_order
        self.assertEqual(len(state.get_op_order()), 1)

        somehost = inventory.get_host('somehost')
        anotherhost = inventory.get_host('anotherhost')

        # Ensure between the two hosts we only run the one op
        self.assertEqual(len(state.ops[somehost]) + len(state.ops[anotherhost]), 1)

        # Check run works
        run_ops(state)

        self.assertEqual((
            state.results[somehost]['success_ops']
            + state.results[anotherhost]['success_ops']

def test_full_op_fail(self):
        inventory = make_inventory()
        state = State(inventory, Config())
        connect_all(state)

        add_op(state, server.shell, 'echo "hi"')

        with patch('pyinfra.api.connectors.ssh.run_shell_command') as fake_run_command:
            fake_channel = FakeChannel(1)
            fake_run_command.return_value = (
                False,
                FakeBuffer('', fake_channel),
            )

            with self.assertRaises(PyinfraError) as e:
                run_ops(state)

            self.assertEqual(e.exception.args[0], 'No hosts remaining!')

            somehost = inventory.get_host('somehost')

            # Ensure the op was not flagged as success

def test_op(self):
        inventory = make_inventory()
        somehost = inventory.get_host('somehost')
        anotherhost = inventory.get_host('anotherhost')

        state = State(inventory, Config())

        # Enable printing on this test to catch any exceptions in the formatting
        state.print_output = True
        state.print_fact_info = True
        state.print_fact_output = True

        connect_all(state)

        add_op(
            state, files.file,
            '/var/log/pyinfra.log',
            user='pyinfra',
            group='pyinfra',
            mode='644',
            sudo=True,
            sudo_user='test_sudo',
            su_user='test_su',
            ignore_errors=True,
            env={
                'TEST': 'what',
            },
        )

        op_order = state.get_op_order()

def test_function_call_op(self):
        inventory = make_inventory()
        state = State(inventory, Config())
        connect_all(state)

        is_called = []

        def mocked_function(*args, **kwargs):
            is_called.append(True)
            return None

        # Add op to both hosts
        add_op(state, python.call, mocked_function)

        # Ensure there is one op
        self.assertEqual(len(state.get_op_order()), 1)

        run_ops(state)

        assert is_called

state, name,
                args=args,
            )

        print_facts(fact_data)
        _exit()

    # Prepare the deploy!
    #

    # Execute a raw command with server.shell
    if command == 'exec':
        # Print the output of the command
        state.print_output = True

        add_op(
            state, server.shell,
            ' '.join(operations),
        )

    # Deploy files(s)
    elif command == 'deploy':
        print()
        print('--> Preparing operations...')

        # Number of "steps" to make = number of files * number of hosts
        for i, filename in enumerate(operations):
            logger.info('Loading: {0}'.format(click.style(filename, bold=True)))
            state.current_op_file = i
            load_deploy_file(state, filename)

    # Operation w/optional args

FAIL_PERCENT=81,
    CONNECT_TIMEOUT=5,
)

# Setup the pyinfra state for this deploy
state = State(inventory, config)


# Connect to all the hosts
print('Connecting...')
connect_all(state)


# Start adding operations
print('Generating operations...')
add_op(
    state, server.user,
    'pyinfra',
    home='/home/pyinfra',
    shell='/bin/bash',
    sudo=True,
)

add_op(
    state, server.group,
    {'Ensure pyinfra2 group exists'},  # set as the first arg names the operation
    'pyinfra2',
    sudo=True,
    # Add an op only to a subset of hosts
    # (in this case, the inventory.centos group)
    hosts=inventory.get_group('centos'),
)

print('--> Preparing operations...')

        # Number of "steps" to make = number of files * number of hosts
        for i, filename in enumerate(operations):
            logger.info('Loading: {0}'.format(click.style(filename, bold=True)))
            state.current_op_file = i
            load_deploy_file(state, filename)

    # Operation w/optional args
    elif command == 'op':
        print()
        print('--> Preparing operation...')

        op, args = operations

        add_op(
            state, op,
            *args[0], **args[1]
        )

    # Always show meta output
    print()
    print('--> Proposed changes:')
    print_meta(state)

    # If --debug-facts or --debug-operations, print and exit
    if debug_facts or debug_operations:
        if debug_facts:
            print_state_facts(state)

        if debug_operations:
            print_state_operations(state)

mode='644',
    sudo=True,
)

# Ensure the state of directories
add_op(
    state, files.directory,
    '/tmp/email',
    user='pyinfra',
    group='pyinfra',
    mode='755',
    sudo=True,
)

# Copy local files to remote host
add_op(
    state, files.put,
    'files/file.txt', '/home/vagrant/file.txt',
)


# And finally we run the ops
run_ops(state)


# We can also get facts for all the hosts
facts = get_facts(state, 'os')
print(jsonify(facts, indent=4))