diff options
Diffstat (limited to 'test/lib/ansible_test/_internal/inventory.py')
-rw-r--r-- | test/lib/ansible_test/_internal/inventory.py | 175 |
1 files changed, 175 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/inventory.py b/test/lib/ansible_test/_internal/inventory.py new file mode 100644 index 0000000..6abf9ed --- /dev/null +++ b/test/lib/ansible_test/_internal/inventory.py @@ -0,0 +1,175 @@ +"""Inventory creation from host profiles.""" +from __future__ import annotations + +import shutil +import typing as t + +from .config import ( + EnvironmentConfig, +) + +from .util import ( + sanitize_host_name, + exclude_none_values, +) + +from .host_profiles import ( + ControllerHostProfile, + ControllerProfile, + HostProfile, + Inventory, + NetworkInventoryProfile, + NetworkRemoteProfile, + SshTargetHostProfile, + WindowsInventoryProfile, + WindowsRemoteProfile, +) + +from .ssh import ( + ssh_options_to_str, +) + + +def create_controller_inventory(args: EnvironmentConfig, path: str, controller_host: ControllerHostProfile) -> None: + """Create and return inventory for use in controller-only integration tests.""" + inventory = Inventory( + host_groups=dict( + testgroup=dict( + testhost=dict( + ansible_connection='local', + ansible_pipelining='yes', + ansible_python_interpreter=controller_host.python.path, + ), + ), + ), + ) + + inventory.write(args, path) + + +def create_windows_inventory(args: EnvironmentConfig, path: str, target_hosts: list[HostProfile]) -> None: + """Create and return inventory for use in target Windows integration tests.""" + first = target_hosts[0] + + if isinstance(first, WindowsInventoryProfile): + if args.explain: + return + + try: + shutil.copyfile(first.config.path, path) + except shutil.SameFileError: + pass + + return + + target_hosts = t.cast(list[WindowsRemoteProfile], target_hosts) + hosts = [(target_host, target_host.wait_for_instance().connection) for target_host in target_hosts] + windows_hosts = {sanitize_host_name(host.config.name): host.get_inventory_variables() for host, connection in hosts} + + inventory = Inventory( + host_groups=dict( + windows=windows_hosts, + ), + # The `testhost` group is needed to support the `binary_modules_winrm` integration test. + # The test should be updated to remove the need for this. + extra_groups={ + 'testhost:children': [ + 'windows', + ], + }, + ) + + inventory.write(args, path) + + +def create_network_inventory(args: EnvironmentConfig, path: str, target_hosts: list[HostProfile]) -> None: + """Create and return inventory for use in target network integration tests.""" + first = target_hosts[0] + + if isinstance(first, NetworkInventoryProfile): + if args.explain: + return + + try: + shutil.copyfile(first.config.path, path) + except shutil.SameFileError: + pass + + return + + target_hosts = t.cast(list[NetworkRemoteProfile], target_hosts) + host_groups: dict[str, dict[str, dict[str, t.Union[str, int]]]] = {target_host.config.platform: {} for target_host in target_hosts} + + for target_host in target_hosts: + host_groups[target_host.config.platform][sanitize_host_name(target_host.config.name)] = target_host.get_inventory_variables() + + inventory = Inventory( + host_groups=host_groups, + # The `net` group was added to support platform agnostic testing. It may not longer be needed. + # see: https://github.com/ansible/ansible/pull/34661 + # see: https://github.com/ansible/ansible/pull/34707 + extra_groups={ + 'net:children': sorted(host_groups), + }, + ) + + inventory.write(args, path) + + +def create_posix_inventory(args: EnvironmentConfig, path: str, target_hosts: list[HostProfile], needs_ssh: bool = False) -> None: + """Create and return inventory for use in POSIX integration tests.""" + target_hosts = t.cast(list[SshTargetHostProfile], target_hosts) + + if len(target_hosts) != 1: + raise Exception() + + target_host = target_hosts[0] + + if isinstance(target_host, ControllerProfile) and not needs_ssh: + inventory = Inventory( + host_groups=dict( + testgroup=dict( + testhost=dict( + ansible_connection='local', + ansible_pipelining='yes', + ansible_python_interpreter=target_host.python.path, + ), + ), + ), + ) + else: + connections = target_host.get_controller_target_connections() + + if len(connections) != 1: + raise Exception() + + ssh = connections[0] + + testhost: dict[str, t.Optional[t.Union[str, int]]] = dict( + ansible_connection='ssh', + ansible_pipelining='yes', + ansible_python_interpreter=ssh.settings.python_interpreter, + ansible_host=ssh.settings.host, + ansible_port=ssh.settings.port, + ansible_user=ssh.settings.user, + ansible_ssh_private_key_file=ssh.settings.identity_file, + ansible_ssh_extra_args=ssh_options_to_str(ssh.settings.options), + ) + + if ssh.become: + testhost.update( + ansible_become='yes', + ansible_become_method=ssh.become.method, + ) + + testhost = exclude_none_values(testhost) + + inventory = Inventory( + host_groups=dict( + testgroup=dict( + testhost=testhost, + ), + ), + ) + + inventory.write(args, path) |