test_host.py

"""Test class for Hosts UI

:Requirement: Host

:CaseAutomation: Automated

:CaseComponent: Hosts

:Team: Proton

:CaseImportance: High

"""

from contextlib import contextmanager
import copy
import csv
from datetime import UTC, datetime, timedelta
import json
import re
import time

from airgun.exceptions import DisabledWidgetError, NoSuchElementException
from box import Box
import pytest
from wait_for import wait_for
import yaml

from robottelo.config import settings
from robottelo.constants import (
    ANY_CONTEXT,
    DEFAULT_ARCHITECTURE,
    DEFAULT_CV,
    DEFAULT_LOC,
    DEFAULT_ORG,
    ENVIRONMENT,
    FAKE_1_CUSTOM_PACKAGE,
    FAKE_7_CUSTOM_PACKAGE,
    FAKE_8_CUSTOM_PACKAGE,
    FAKE_8_CUSTOM_PACKAGE_NAME,
    FOREMAN_PROVIDERS,
    OSCAP_PERIOD,
    OSCAP_WEEKDAY,
    REPO_TYPE,
    REPOS,
    ROLES,
)
from robottelo.constants.repos import CUSTOM_FILE_REPO
from robottelo.exceptions import APIResponseError
from robottelo.logging import logger
from robottelo.utils.datafactory import gen_string
from tests.foreman.api.test_errata import cv_publish_promote


def _get_set_from_list_of_dict(value):
    """Returns a set of tuples representation of each dict sorted by keys

    :param list value: a list of simple dict.
    """
    return {tuple(sorted(list(global_param.items()), key=lambda t: t[0])) for global_param in value}


@contextmanager
def mock_service_as_rebootable(contenthost, service_name):
    """Context manager to temporarily make a service require reboot instead of restart.

    Modifies the katello tracer's STATIC_SERVICES list on the content host to include
    the specified service, causing tracer to classify it as type='static' (reboot required)
    instead of type='daemon' (restart required).

    This is useful for testing reboot-required trace scenarios without needing to
    actually downgrade kernel or systemd packages.

    :param contenthost: ContentHost instance to modify
    :param service_name: Name of the service to mark as reboot-required

    Usage:
        with mock_service_as_rebootable(rhel_contenthost, 'robottelo-mock-service'):
            rhel_contenthost.execute('yum -y downgrade robottelo-mock-service')

    """
    # Dynamically find the katello tracer dnf.py file (Python version agnostic)
    find_cmd = 'python3 -c "import katello.tracer.dnf; print(katello.tracer.dnf.__file__)"'
    find_result = contenthost.execute(find_cmd)

    if find_result.status != 0:
        raise RuntimeError(
            f'Failed to locate katello tracer module on {contenthost.hostname}: {find_result.stderr}'
        )

    tracer_file = find_result.stdout.strip()
    backup_path = '/tmp/katello_tracer_dnf.py.backup'

    # Create backup
    backup_result = contenthost.execute(f'cp {tracer_file} {backup_path}')
    if backup_result.status != 0:
        logger.warning(f'Failed to backup tracer file: {backup_result.stderr}')

    # Add service to STATIC_SERVICES list
    # In the ../tracer/dnf.py it looks like this:
    #   STATIC_SERVICES = [
    #      "systemd",
    #      "dbus",
    #    ]
    add_cmd = f'''python3 << 'EOFPYTHON'
import re

with open('{tracer_file}', 'r') as f:
    content = f.read()

# Find STATIC_SERVICES list and add service before the closing bracket
pattern = r'(STATIC_SERVICES = \\[.*?)(\\])'
replacement = r'\\1    "{service_name}",\\n\\2'
modified_content = re.sub(pattern, replacement, content, flags=re.DOTALL)

with open('{tracer_file}', 'w') as f:
    f.write(modified_content)

print("Service added successfully")
EOFPYTHON
'''
    add_result = contenthost.execute(add_cmd)

    if add_result.status != 0:
        logger.error(f'Failed to add service to STATIC_SERVICES: {add_result.stderr}')
        # Restore backup
        contenthost.execute(f'mv {backup_path} {tracer_file}')
        raise RuntimeError(f'Failed to modify tracer configuration: {add_result.stderr}')

    try:
        # Verify the modification
        verify_result = contenthost.execute(f'grep "{service_name}" {tracer_file}')
        if verify_result.status == 0:
            logger.info(
                f'Successfully added {service_name} to STATIC_SERVICES on {contenthost.hostname}'
            )
        yield
    finally:
        # Cleanup: Restore original file
        restore_result = contenthost.execute(f'mv {backup_path} {tracer_file}')
        if restore_result.status == 0:
            logger.info(f'Restored original tracer configuration on {contenthost.hostname}')
        else:
            logger.warning(
                f'Failed to restore tracer file on {contenthost.hostname}: {restore_result.stderr}'
            )


# this fixture inherits the fixture called ui_user in confest.py, method name has to be same
@pytest.fixture(scope='module')
def ui_user(ui_user, smart_proxy_location, module_target_sat):
    module_target_sat.api.User(
        id=ui_user.id,
        default_location=smart_proxy_location,
    ).update(['default_location'])
    return ui_user


@pytest.fixture
def ui_admin_user(target_sat):
    """Admin user."""
    admin_user = target_sat.api.User().search(
        query={'search': f'login={settings.server.admin_username}'}
    )[0]
    admin_user.password = settings.server.admin_password

    return admin_user


@pytest.fixture
def host_ui_default(target_sat):
    settings_object = target_sat.api.Setting().search(query={'search': 'name=host_details_ui'})[0]
    settings_object.value = 'No'
    settings_object.update({'value'})
    yield
    settings_object.value = 'Yes'
    settings_object.update({'value'})


@pytest.fixture
def ui_view_hosts_user(target_sat, current_sat_org, current_sat_location, expected_permissions):
    """User with View hosts role."""
    role = target_sat.api.Role(organization=[current_sat_org]).create()
    target_sat.api_factory.create_role_permissions(
        role,
        {
            'Host': ['view_hosts'],
            'Organization': expected_permissions['Organization'],
            'Location': expected_permissions['Location'],
        },
    )
    password = gen_string('alphanumeric')
    user = target_sat.api.User(
        admin=False,
        location=[current_sat_location],
        organization=[current_sat_org],
        role=[role],
        password=password,
    ).create()
    user.password = password

    yield user

    user.delete()


@pytest.fixture(params=['ui_admin_user', 'ui_view_hosts_user'])
def ui_hosts_columns_user(request):
    """Parametrized fixture returning defined users for the UI session."""
    return request.getfixturevalue(request.param)


@pytest.fixture
def scap_policy(scap_content, target_sat):
    return target_sat.cli_factory.make_scap_policy(
        {
            'name': gen_string('alpha'),
            'deploy-by': 'ansible',
            'scap-content-id': scap_content["scap_id"],
            'scap-content-profile-id': scap_content["scap_profile_id"],
            'period': OSCAP_PERIOD['weekly'].lower(),
            'weekday': OSCAP_WEEKDAY['friday'].lower(),
        }
    )


second_scap_policy = scap_policy


@pytest.fixture(scope='module')
def module_global_params(module_target_sat):
    """Create 3 global parameters and clean up at teardown"""
    global_parameters = []
    for _ in range(3):
        global_parameter = module_target_sat.api.CommonParameter(
            name=gen_string('alpha'), value=gen_string('alphanumeric')
        ).create()
        global_parameters.append(global_parameter)
    yield global_parameters
    # cleanup global parameters
    for global_parameter in global_parameters:
        global_parameter.delete()


@pytest.fixture
def tracer_install_host(rex_contenthost, target_sat):
    """This fixture automatically configures IPv6 support based on the host's network type and creates
    version-appropriate repositories.

    :param rex_contenthost: Remote execution enabled content host
    :param target_sat: Target Satellite server
    :return: ContentHost with tracer tools installed and configured
    """

    # add IPv6 proxy for IPv6 communication based on network type
    if not rex_contenthost.network_type.has_ipv4:
        rex_contenthost.enable_ipv6_dnf_and_rhsm_proxy()
        rex_contenthost.enable_ipv6_system_proxy()

    # create a custom, rhel version-specific OS repo
    rhelver = rex_contenthost.os_version.major

    if rhelver > 7:
        # RHEL 8, 9 and 10 use the same repository structure
        rex_contenthost.create_custom_repos(**settings.repos[f'rhel{rhelver}_os'])
    else:
        # RHEL 7 has different repository structure
        rex_contenthost.create_custom_repos(
            **{f'rhel{rhelver}_os': settings.repos[f'rhel{rhelver}_os']}
        )
    return rex_contenthost


@pytest.fixture
def tracer_hosts(rex_contenthosts, target_sat):
    """Fixture that provides two tracer hosts with mock service installed.
    Similar to the tracer_host and tracer_install_host fixtures but provides multiple hosts for bulk operations testing.
    """
    for host in rex_contenthosts:
        # add IPv6 proxy for IPv6 communication based on network type
        if not host.network_type.has_ipv4:
            host.enable_ipv6_dnf_and_rhsm_proxy()
            host.enable_ipv6_system_proxy()

        # create a custom, rhel version-specific OS repo
        rhelver = host.os_version.major
        if rhelver > 7:
            # RHEL 8, 9 and 10 use the same repository structure
            host.create_custom_repos(**settings.repos[f'rhel{rhelver}_os'])
        else:
            # RHEL 7 has different repository structure
            host.create_custom_repos(**{f'rhel{rhelver}_os': settings.repos[f'rhel{rhelver}_os']})

        # Install tracer
        host.install_tracer()

        # Install mock service repository and package
        host.create_custom_repos(
            **{f'mock_service_rhel{rhelver}': settings.repos['MOCK_SERVICE_REPO'][f'rhel{rhelver}']}
        )
        assert host.execute(f'yum -y install {settings.repos["MOCK_SERVICE_RPM"]}').status == 0
        assert host.execute(f'rpm -q {settings.repos["MOCK_SERVICE_RPM"]}').status == 0
        host.execute(f'systemctl start {settings.repos["MOCK_SERVICE_RPM"]}')

    return rex_contenthosts


@pytest.mark.e2e
def test_positive_end_to_end(module_global_params, target_sat, host_ui_options, request, ui_user):
    """Create a new Host with parameters, config group. Check host presence on
        the dashboard. Update name with 'new' prefix and delete.

    :id: d2f86309-1a6d-42dc-a865-9e607cd25ae5

    :expectedresults: Host is created with parameters, config group. Updated
        and deleted.

    :BZ: 1419161
    """
    api_values, host_name = host_ui_options
    global_params = [
        global_param.to_json_dict(lambda attr, field: attr in ['name', 'value'])
        for global_param in module_global_params
    ]
    host_parameters = []
    for _ in range(2):
        host_parameters.append(dict(name=gen_string('alpha'), value=gen_string('alphanumeric')))
    expected_host_parameters = copy.deepcopy(host_parameters)
    # override the first global parameter
    overridden_global_parameter = {'name': global_params[0]['name'], 'value': gen_string('alpha')}
    expected_host_parameters.append(overridden_global_parameter)
    expected_global_parameters = copy.deepcopy(global_params)
    for global_param in expected_global_parameters:
        # update with overridden expected value
        if global_param['name'] == overridden_global_parameter['name']:
            global_param['overridden'] = True
        else:
            global_param['overridden'] = False
    new_name = f"new{gen_string('alpha').lower()}"
    new_host_name = f"{new_name}.{api_values['interfaces.interface.domain']}"

    stripped_headers = None

    @request.addfinalizer
    def _finalize():
        # Get table to original state
        with target_sat.ui_session(user=ui_user.login, password=ui_user.password) as session:
            session.organization.select(api_values['host.organization'])
            session.location.select(api_values['host.location'])
            session.all_hosts.manage_table_columns({header: True for header in stripped_headers})

    with target_sat.ui_session(user=ui_user.login, password=ui_user.password) as session:
        session.organization.select(api_values['host.organization'])
        session.location.select(api_values['host.location'])
        api_values.update(
            {
                'parameters.host_params': host_parameters,
                'parameters.global_params': [overridden_global_parameter],
            }
        )
        session.host.create(api_values)
        assert session.host.search(host_name)[0]['Name'] == host_name
        values = session.host.read(host_name, widget_names=['parameters'])
        assert _get_set_from_list_of_dict(
            values['parameters']['host_params']
        ) == _get_set_from_list_of_dict(expected_host_parameters)
        assert _get_set_from_list_of_dict(expected_global_parameters).issubset(
            _get_set_from_list_of_dict(values['parameters']['global_params'])
        )

        # check host presence on the dashboard
        dashboard_values = session.dashboard.read('NewHosts')['hosts']
        displayed_host = [row for row in dashboard_values if row['Host'] == host_name][0]
        assert api_values['operating_system.operating_system'] in displayed_host['Operating System']
        assert displayed_host['Installed'] == 'N/A'
        # update
        session.host.update(host_name, {'host.name': new_name})
        assert session.host.search(host_name)[0][0] == 'No Results'
        assert session.host.search(new_host_name)[0]['Name'] == new_host_name
        # delete
        headers = session.all_hosts.get_displayed_table_headers()
        stripped_headers = tuple(
            header for header in headers if header is not None and header != 'Name'
        )
        wait_for(lambda: session.browser.refresh(), timeout=5)
        # Make sure there is only Name column displayed
        session.all_hosts.manage_table_columns({header: False for header in stripped_headers})
        assert session.all_hosts.delete(new_host_name)
        assert not target_sat.api.Host().search(query={'search': f'name="{new_host_name}"'})


def test_positive_read_from_details_page(target_sat, module_host_template, ui_user):
    """Create new Host and read all its content through details page

    :id: ffba5d40-918c-440e-afbb-6b910db3a8fb

    :expectedresults: Host is created and has expected content
    """

    template = module_host_template
    template.name = gen_string('alpha').lower()
    host = template.create()
    os_name = f'{template.operatingsystem.name} {template.operatingsystem.major}'
    host_name = host.name
    with target_sat.ui_session(user=ui_user.login, password=ui_user.password) as session:
        assert session.host_new.search(host_name)[0]['Name'] == host_name
        values = session.host_new.get_details(host_name)
        assert values['overview']['host_status']['status'] == 'All statuses OK'
        assert (
            values['details']['system_properties']['sys_properties']['domain']
            == template.domain.name
        )
        assert values['overview']['details']['details']['mac_address'] == host.mac
        assert values['details']['operating_system']['architecture'] == template.architecture.name
        assert values['details']['operating_system']['os'] == os_name
        assert (
            values['details']['system_properties']['sys_properties']['location']
            == template.location.name
        )
        assert (
            values['details']['system_properties']['sys_properties']['organization']
            == template.organization.name
        )
        assert (
            'Admin User' in values['details']['system_properties']['sys_properties']['host_owner']
        )


def test_read_host_with_ics_domain(
    session, module_host_template, smart_proxy_location, module_org, module_target_sat
):
    """Create new Host with ics domain name and verify that it can be read

    :id: 54e3db92-16c2-412b-bf68-44d479c5987b

    :steps:
        1. Create a host with a domain ending in .ics
        2. Read the host's details through the UI

    :expectedresults: Host ending with ics domain name can be accessed through Host UI

    :customerscenario: true

    :Verifies: SAT-26202
    """
    template = module_host_template
    template.name = gen_string('alpha').lower()
    ics_domain = module_target_sat.api.Domain(
        location=[smart_proxy_location],
        organization=[module_org],
        name=gen_string('alpha').lower() + '.ics',
    ).create()
    template.domain = ics_domain
    host = template.create()
    host_name = host.name
    with module_target_sat.ui_session() as session:
        session.organization.select(module_org.name)
        session.location.select(smart_proxy_location.name)
        values = session.host_new.get_details(host_name, widget_names='details')
        assert (
            values['details']['system_properties']['sys_properties']['domain']
            == template.domain.name
        )
        assert values['details']['system_properties']['sys_properties']['name'] == host_name


def test_positive_read_from_edit_page(target_sat, ui_user, host_ui_options):
    """Create new Host and read all its content through edit page

    :id: 758fcab3-b363-4bfc-8f5d-173098a7e72d

    :expectedresults: Host is created and has expected content
    """
    api_values, host_name = host_ui_options
    with target_sat.ui_session(user=ui_user.login, password=ui_user.password) as session:
        session.organization.select(api_values['host.organization'])
        session.location.select(api_values['host.location'])
        session.host.create(api_values)
        assert session.host.search(host_name)[0]['Name'] == host_name
        values = session.host.read(host_name)
        assert values['host']['name'] == host_name.partition('.')[0]
        assert values['host']['organization'] == api_values['host.organization']
        assert (
            values['operating_system']['architecture']
            == api_values['operating_system.architecture']
        )
        assert (
            values['operating_system']['operating_system']
            == api_values['operating_system.operating_system']
        )
        assert values['operating_system']['media_type'] == 'All Media'
        assert values['operating_system']['media'] == api_values['operating_system.media']
        assert values['operating_system']['ptable'] == api_values['operating_system.ptable']
        assert (
            values['interfaces']['interfaces_list'][0]['Identifier']
            == api_values['interfaces.interface.device_identifier']
        )
        assert values['interfaces']['interfaces_list'][0]['Type'] == 'Interface physical'
        assert (
            values['interfaces']['interfaces_list'][0]['MAC Address']
            == api_values['interfaces.interface.mac']
        )
        assert values['interfaces']['interfaces_list'][0]['FQDN'] == host_name
        assert session._user in values['additional_information']['owned_by']
        assert values['additional_information']['enabled'] is True


def test_positive_assign_taxonomies(
    module_org,
    smart_proxy_location,
    target_sat,
    function_org,
    function_location_with_org,
    ui_user,
    host_ui_options,
):
    """Ensure Host organization and Location can be assigned.

    :id: 52466df5-6f56-4faa-b0f8-42b63731f494

    :expectedresults: Host Assign Organization and Location actions are
        working as expected.
    """

    host = target_sat.api.Host(organization=module_org, location=smart_proxy_location).create()
    with target_sat.ui_session(user=ui_user.login, password=ui_user.password) as session:
        session.organization.select(host_ui_options[0]['host.organization'])
        session.location.select(host_ui_options[0]['host.location'])
        assert session.all_hosts.search(host.name)[0]['Name'] == host.name
        session.all_hosts.change_associations_organization(
            host_names=[host.name],
            new_organization=function_org.name,
        )
        assert not target_sat.api.Host(organization=module_org).search(
            query={'search': f'name="{host.name}"'}
        )
        assert (
            len(
                target_sat.api.Host(organization=function_org).search(
                    query={'search': f'name="{host.name}"'}
                )
            )
            == 1
        )
        session.organization.select(org_name=function_org.name)
        assert session.all_hosts.search(host.name)[0]['Name'] == host.name
        session.all_hosts.change_associations_location(
            host_names=[host.name],
            new_location=function_location_with_org.name,
        )
        assert not target_sat.api.Host(location=smart_proxy_location).search(
            query={'search': f'name="{host.name}"'}
        )
        assert (
            len(
                target_sat.api.Host(location=function_location_with_org).search(
                    query={'search': f'name="{host.name}"'}
                )
            )
            == 1
        )
        session.location.select(loc_name=function_location_with_org.name)
        assert session.all_hosts.search(host.name)[0]['Name'] == host.name
        values = session.host_new.get_details(host.name)
        assert (
            values['details']['system_properties']['sys_properties']['location']
            == function_location_with_org.name
        )
        assert (
            values['details']['system_properties']['sys_properties']['organization']
            == function_org.name
        )


@pytest.mark.skipif(
    (settings.ui.webdriver != 'chrome'), reason='Currently only chrome is supported'
)
def test_positive_export_selected_columns(request, target_sat, current_sat_location):
    """Select certain columns in the hosts table and check that they are exported in the CSV file.

    :id: 2b65c1d6-0b94-11ef-a4b7-000c2989e153

    :steps:
        1. Select different columns to be displayed in the hosts table.
        2. Export the hosts into CSV file.

    :expectedresults: All columns selected in the UI table should be exported in the CSV file.

    :BZ: 2167146

    :Verifies: SAT-38427

    :customerscenario: true

    """

    columns = (
        Box(ui='Power', csv='Power Status', displayed=True),
        Box(ui='Name', csv='Name', displayed=True),
        Box(ui='OS', csv='Operatingsystem', displayed=True),
        Box(ui='Owner', csv='Owner', displayed=True),
        Box(ui='Host group', csv='Hostgroup', displayed=True),
        Box(ui='Boot time', csv='Reported Data - Boot Time', displayed=True),
        Box(ui='Last report', csv='Last Report', displayed=True),
        Box(ui='Comment', csv='Comment', displayed=True),
        Box(ui='IPv4', csv='Ip', displayed=True),
        Box(ui='IPv6', csv='Ip6', displayed=True),
        Box(ui='MAC', csv='Mac', displayed=True),
        Box(ui='Model', csv='Compute Resource Or Model', displayed=True),
        Box(ui='Sockets', csv='Reported Data - Sockets', displayed=True),
        Box(ui='Cores', csv='Reported Data - Cores', displayed=True),
        Box(ui='RAM', csv='Reported Data - Ram', displayed=True),
        Box(ui='Virtual', csv='Virtual', displayed=True),
        Box(ui='Total disk space', csv='Reported Data - Disks Total', displayed=True),
        Box(ui='Kernel version', csv='Reported Data - Kernel Version', displayed=True),
        Box(ui='BIOS vendor', csv='Reported Data - Bios Vendor', displayed=True),
        Box(ui='BIOS release date', csv='Reported Data - Bios Release Date', displayed=True),
        Box(ui='BIOS version', csv='Reported Data - Bios Version', displayed=True),
        Box(ui='RHEL Lifecycle status', csv='Rhel Lifecycle Status', displayed=True),
        Box(ui='Installable updates', csv='Installable updates', displayed=False),
        Box(ui='Last seen', csv='Last Checkin', displayed=True),
        Box(ui='Lifecycle environment', csv='Lifecycle Environment', displayed=True),
        Box(ui='Content view', csv='Content View', displayed=True),
        Box(ui='Registered at', csv='Registered', displayed=True),
        Box(ui='Recommendations', csv='Recommendations', displayed=True),
    )

    with target_sat.ui_session() as session:
        session.location.select(loc_name=current_sat_location.name)
        # Save original column settings
        original_headers = session.all_hosts.get_displayed_table_headers()
        original_columns = {header: True for header in original_headers if header is not None}

        def restore_columns():
            """Restore original column settings after test"""
            with target_sat.ui_session() as restore_session:
                restore_session.location.select(loc_name=current_sat_location.name)
                wait_for(lambda: restore_session.browser.refresh(), timeout=5)
                all_possible_columns = {column.ui: False for column in columns}
                all_possible_columns.update(original_columns)
                restore_session.all_hosts.manage_table_columns(all_possible_columns)

        request.addfinalizer(restore_columns)

        # Set test-specific columns
        session.all_hosts.manage_table_columns({column.ui: column.displayed for column in columns})
        file_path = session.all_hosts.export()
        with open(file_path, newline='') as fh:
            csvfile = csv.DictReader(fh)
            assert set(csvfile.fieldnames) == set(
                [column.csv for column in columns if column.displayed]
            )


def test_positive_create_with_inherited_params(
    target_sat, function_org, function_location_with_org
):
    """Create a new Host in organization and location with parameters

    :BZ: 1287223

    :id: 628122f2-bda9-4aa1-8833-55debbd99072

    :expectedresults: Host has inherited parameters from organization and
        location

    :CaseImportance: High
    """
    org_param = dict(name=gen_string('alphanumeric'), value=gen_string('alphanumeric'))
    loc_param = dict(name=gen_string('alphanumeric'), value=gen_string('alphanumeric'))
    host_template = target_sat.api.Host(
        organization=function_org, location=function_location_with_org
    )
    host_template.create_missing()
    host = host_template.create()
    host_name = host.name
    with target_sat.ui_session() as session:
        session.organization.select(org_name=function_org.name)
        session.location.select(loc_name=function_location_with_org.name)
        session.organization.update(function_org.name, {'parameters.resources': org_param})
        session.location.update(
            function_location_with_org.name, {'parameters.resources': loc_param}
        )
        session.organization.select(org_name=function_org.name)
        session.location.select(loc_name=function_location_with_org.name)
        values = session.host.read(host_name, 'parameters')
        expected_params = {
            (org_param['name'], org_param['value']),
            (loc_param['name'], loc_param['value']),
        }
        assert expected_params.issubset(
            {(param['name'], param['value']) for param in values['parameters']['global_params']}
        )


def test_negative_delete_primary_interface(module_target_sat, host_ui_options, ui_user):
    """Attempt to delete primary interface of a host

    :id: bc747e2c-38d9-4920-b4ae-6010851f704e

    :customerscenario: true

    :BZ: 1417119

    :expectedresults: Interface was not deleted
    """
    values, host_name = host_ui_options
    interface_id = values['interfaces.interface.device_identifier']
    with module_target_sat.ui_session(user=ui_user.login, password=ui_user.password) as session:
        session.location.select(values['host.location'])
        session.host.create(values)
        with pytest.raises(DisabledWidgetError) as context:
            session.host.delete_interface(host_name, interface_id)
        assert 'Interface Delete button is disabled' in str(context.value)


def test_positive_view_hosts_with_non_admin_user(
    test_name, module_org, smart_proxy_location, target_sat, host_ui_default
):
    """View hosts and content hosts as a non-admin user with only view_hosts, edit_hosts
    and view_organization permissions

    :BZ: 1642076, 1801630

    :customerscenario: true

    :id: 19a07026-0550-11ea-bfdc-98fa9b6ecd5a

    :expectedresults: user with only view_hosts, edit_hosts and view_organization permissions
        is able to read content hosts and hosts
    """
    user_password = gen_string('alpha')
    role = target_sat.api.Role(organization=[module_org]).create()
    target_sat.api_factory.create_role_permissions(
        role, {'Organization': ['view_organizations'], 'Host': ['view_hosts']}
    )
    user = target_sat.api.User(
        role=[role],
        admin=False,
        password=user_password,
        organization=[module_org],
        location=[smart_proxy_location],
        default_organization=module_org,
        default_location=smart_proxy_location,
    ).create()
    created_host = target_sat.api.Host(
        location=smart_proxy_location, organization=module_org
    ).create()
    with target_sat.ui_session(test_name, user=user.login, password=user_password) as session:
        host = session.host_new.get_details(created_host.name, widget_names='breadcrumb')
        assert host['breadcrumb'] == created_host.name


def test_positive_remove_parameter_non_admin_user(
    test_name, module_org, smart_proxy_location, target_sat, host_ui_default, expected_permissions
):
    """Remove a host parameter as a non-admin user with enough permissions

    :BZ: 1996035

    :id: 598111c1-fdb6-42e9-8c28-fae999b5d112

    :expectedresults: user with sufficient permissions may remove host
        parameter
    """
    user_password = gen_string('alpha')
    parameter = {'name': gen_string('alpha'), 'value': gen_string('alpha')}
    role = target_sat.api.Role(organization=[module_org]).create()
    target_sat.api_factory.create_role_permissions(
        role,
        {
            'Parameter': expected_permissions['Parameter'],
            'Host': expected_permissions['Host'],
            'Operatingsystem': ['view_operatingsystems'],
            'Organization': expected_permissions['Organization'],
            'Location': expected_permissions['Location'],
        },
    )
    user = target_sat.api.User(
        role=[role],
        admin=False,
        password=user_password,
        organization=[module_org],
        location=[smart_proxy_location],
        default_organization=module_org,
        default_location=smart_proxy_location,
    ).create()
    host = target_sat.api.Host(
        location=smart_proxy_location,
        organization=module_org,
        host_parameters_attributes=[parameter],
    ).create()
    with target_sat.ui_session(
        testname=test_name, user=user.login, password=user_password
    ) as session:
        read_parameters = session.host_new.get_parameters(host.name)
        parameters_table = read_parameters.get('parameters_table', [])
        assert any(
            row.get('Name') == parameter['name'] and row.get('Value') == parameter['value']
            for row in parameters_table
        ), f'Parameter {parameter["name"]} not found in parameters table.'
        session.host_new.delete_parameter(host.name, parameter['name'])
        read_parameters = session.host_new.get_parameters(host.name)
        parameters_table = read_parameters.get('parameters_table', [])
        assert all(row.get('Name') != parameter['name'] for row in parameters_table), (
            f'Parameter {parameter["name"]} still present after delete.'
        )


def test_negative_remove_parameter_non_admin_user(
    test_name, module_org, smart_proxy_location, module_target_sat, expected_permissions
):
    """Attempt to remove host parameter as a non-admin user with
    insufficient permissions

    :BZ: 1317868

    :id: 78fd230e-2ec4-4158-823b-ddbadd5e232f

    :customerscenario: true

    :expectedresults: user with insufficient permissions is unable to
        remove host parameter, 'Remove' link is not visible for him
    """

    user_password = gen_string('alpha')
    parameter = {'name': gen_string('alpha'), 'value': gen_string('alpha')}
    role = module_target_sat.api.Role(organization=[module_org]).create()
    module_target_sat.api_factory.create_role_permissions(
        role,
        {
            'Parameter': ['view_params'],
            'Host': ['view_hosts'],
            'Operatingsystem': ['view_operatingsystems'],
            'Organization': expected_permissions['Organization'],
            'Location': expected_permissions['Location'],
        },
    )
    user = module_target_sat.api.User(
        role=[role],
        admin=False,
        password=user_password,
        organization=[module_org],
        location=[smart_proxy_location],
        default_organization=module_org,
        default_location=smart_proxy_location,
    ).create()
    host = module_target_sat.api.Host(
        content_facet_attributes={
            'content_view_id': module_org.default_content_view.id,
            'lifecycle_environment_id': module_org.library.id,
        },
        location=smart_proxy_location,
        organization=module_org,
        host_parameters_attributes=[parameter],
    ).create()
    with module_target_sat.ui_session(
        testname=test_name, user=user.login, password=user_password
    ) as session:
        read_parameters = session.host_new.get_parameters(host.name)
        parameters_table = read_parameters.get('parameters_table', [])
        assert any(
            row.get('Name') == parameter['name'] and row.get('Value') == parameter['value']
            for row in parameters_table
        ), f'Parameter {parameter["name"]} not found in parameters table.'
        with pytest.raises(NoSuchElementException):
            session.host_new.delete_parameter(host.name, parameter['name'])


def test_positive_check_permissions_affect_create_procedure(
    test_name, smart_proxy_location, target_sat, function_org, function_role, expected_permissions
):
    """Verify whether user permissions affect what entities can be selected
    when host is created

    :id: 4502f99d-86fb-4655-a9dc-b2612cf849c6

    :customerscenario: true

    :expectedresults: user with specific permissions can choose only
        entities for create host procedure that he has access to

    :BZ: 1293716
    """
    # Create two lifecycle environments
    lc_env = target_sat.api.LifecycleEnvironment(organization=function_org).create()
    filter_lc_env = target_sat.api.LifecycleEnvironment(organization=function_org).create()
    # Create two content views and promote them to one lifecycle
    # environment which will be used in filter
    cv = target_sat.api.ContentView(organization=function_org).create()
    filter_cv = target_sat.api.ContentView(organization=function_org).create()
    for content_view in [cv, filter_cv]:
        content_view.publish()
        content_view = content_view.read()
        content_view.version[0].promote(data={'environment_ids': filter_lc_env.id})
    # Create two host groups
    hg = target_sat.api.HostGroup(
        organization=[function_org], location=[smart_proxy_location]
    ).create()
    filter_hg = target_sat.api.HostGroup(
        organization=[function_org], location=[smart_proxy_location]
    ).create()
    # Create lifecycle environment permissions and select one specific
    # environment user will have access to
    target_sat.api_factory.create_role_permissions(
        function_role,
        {
            'Katello::KTEnvironment': [
                'promote_or_remove_content_views_to_environments',
                'view_lifecycle_environments',
            ]
        },
        # allow access only to the mentioned here environment
        search=f'name = {filter_lc_env.name}',
    )
    # Add necessary permissions for content view as we did for lce
    target_sat.api_factory.create_role_permissions(
        function_role,
        {
            'Katello::ContentView': [
                'promote_or_remove_content_views',
                'view_content_views',
                'publish_content_views',
            ]
        },
        # allow access only to the mentioned here cv
        search=f'name = {filter_cv.name}',
    )
    # Add necessary permissions for hosts as we did for lce
    target_sat.api_factory.create_role_permissions(
        function_role,
        {'Host': ['create_hosts', 'view_hosts']},
        # allow access only to the mentioned here host group
        search=f'hostgroup_fullname = {filter_hg.name}',
    )
    # Add necessary permissions for host groups as we did for lce
    target_sat.api_factory.create_role_permissions(
        function_role,
        {'Hostgroup': ['view_hostgroups']},
        # allow access only to the mentioned here host group
        search=f'name = {filter_hg.name}',
    )
    # Add permissions for Organization and Location
    target_sat.api_factory.create_role_permissions(
        function_role,
        {
            'Organization': expected_permissions['Organization'],
            'Location': expected_permissions['Location'],
        },
    )
    # Create new user with a configured role
    user_password = gen_string('alpha')
    user = target_sat.api.User(
        role=[function_role],
        admin=False,
        password=user_password,
        organization=[function_org],
        location=[smart_proxy_location],
        default_organization=function_org,
        default_location=smart_proxy_location,
    ).create()
    host_fields = [
        {'name': 'host.hostgroup', 'unexpected_value': hg.name, 'expected_value': filter_hg.name},
        {
            'name': 'host.lce',
            'unexpected_value': lc_env.name,
            'expected_value': filter_lc_env.name,
        },
        {
            'name': 'host.content_view',
            'unexpected_value': cv.name,
            'expected_value': filter_cv.name,
            # content view selection needs the right lce to be selected
            'other_fields_values': {'host.lce': filter_lc_env.name},
        },
    ]
    with target_sat.ui_session(test_name, user=user.login, password=user_password) as session:
        for host_field in host_fields:
            values = {host_field['name']: host_field['unexpected_value']}
            values.update(host_field.get('other_fields_values', {}))
            with pytest.raises(NoSuchElementException) as context:
                session.host.helper.read_create_view(values)
            error_message = str(context.value)
            assert host_field['unexpected_value'] in error_message
            # After the NoSuchElementException from FilteredDropdown, airgun is not able to
            # navigate to other locations, Note in normal situation we should send Escape key to
            # browser.
            session.browser.refresh()
            values = {host_field['name']: host_field['expected_value']}
            values.update(host_field.get('other_fields_values', {}))
            create_values = session.host.helper.read_create_view(values, host_field['name'])
            tab_name, field_name = host_field['name'].split('.')
            assert create_values[tab_name][field_name] == host_field['expected_value']


def test_positive_search_by_parameter(session, module_org, smart_proxy_location, target_sat):
    """Search for the host by global parameter assigned to it

    :id: 8e61127c-d0a0-4a46-a3c6-22d3b2c5457c

    :expectedresults: Only one specific host is returned by search

    :BZ: 1725686
    """
    param_name = gen_string('alpha')
    param_value = gen_string('alpha')
    parameters = [{'name': param_name, 'value': param_value}]
    param_host = target_sat.api.Host(
        organization=module_org,
        location=smart_proxy_location,
        host_parameters_attributes=parameters,
    ).create()
    additional_host = target_sat.api.Host(
        organization=module_org, location=smart_proxy_location
    ).create()
    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=smart_proxy_location.name)
        # Check that hosts present in the system
        for host in [param_host, additional_host]:
            assert session.host_new.search(host.name)[0]['Name'] == host.name
        # Check that search by parameter returns only one host in the list
        values = session.host_new.search(f'params.{param_name} = {param_value}')
        assert len(values) == 1
        assert values[0]['Name'] == param_host.name


@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
@pytest.mark.no_containers
def test_positive_search_by_reported_data(
    target_sat, rhel_contenthost, module_org, module_ak_with_cv
):
    """
    Search for host by reported data.
    For example, search by `reported.bios_vendor = SeaBIOS`.

    :id: 54341d00-34bc-11ef-a8a6-000c29a0e355

    :expectedresults: Return only hosts matching the reported data.

    :Verifies: SAT-9132, SAT-38761

    :customerscenario: true
    """
    result = rhel_contenthost.register(module_org, None, module_ak_with_cv.name, target_sat)
    assert result.status == 0, f'Failed to register host: {result.stderr}'
    client = rhel_contenthost

    reported_data_params = [
        'bios_release_date',
        'bios_vendor',
        'bios_version',
    ]
    source_host = client.nailgun_host
    assert source_host.reported_data, f'Source host {client.hostname} does not report any data.'
    source_reported_data = source_host.reported_data

    with target_sat.ui_session() as session:
        session.organization.select(org_name=ANY_CONTEXT['org'])
        session.location.select(loc_name=ANY_CONTEXT['location'])

        for param_name in reported_data_params:
            param_value = source_reported_data[param_name]
            search_string = f'reported.{param_name} = {param_value}'
            api_hosts = target_sat.api.Host().search(query={'search': search_string})
            ui_hosts = session.host.search(search_string)
            assert set([host.name for host in api_hosts]) == set(
                [host['Name'] for host in ui_hosts]
            )


@pytest.mark.usefixtures('function_host')
def test_positive_search_by_configuration_status_alias(target_sat):
    """
    Search for host by new alias `configuration_status`.

    :id: 6af725d2-44fe-4b04-a8d6-ce8321a673d4

    :expectedresults: Searching hosts by original `status` and new alias `configuration_status`
        returns the same results.

    :Verifies: SAT-9132

    :customerscenario: true
    """
    status_search_term = Box(name='status', alias='configuration_status')
    search_params = [
        'enabled = true',
        'failed >= 0',
        'failed_restarts >= 0',
        'applied >= 0',
        'pending >= 0',
    ]

    with target_sat.ui_session() as session:
        session.organization.select(org_name=ANY_CONTEXT['org'])
        session.location.select(loc_name=ANY_CONTEXT['location'])

        for search_param in search_params:
            results = {}
            for search_term in status_search_term.values():
                search_string = f'{search_term}.{search_param}'
                results[search_term] = [host['Name'] for host in session.host.search(search_string)]
            assert results[status_search_term.name] == results[status_search_term.alias], (
                f'Different search results were found: {results}'
            )


def test_positive_search_by_parameter_with_different_values(
    session, module_org, smart_proxy_location, target_sat
):
    """Search for the host by global parameter assigned to it by its value

    :id: c3a4551e-d759-4a9d-ba90-8db4cab3db2c

    :expectedresults: Only one specific host is returned by search

    :BZ: 1725686
    """
    param_name = gen_string('alpha')
    param_values = [gen_string('alpha'), gen_string('alphanumeric')]
    hosts = [
        target_sat.api.Host(
            organization=module_org,
            location=smart_proxy_location,
            host_parameters_attributes=[{'name': param_name, 'value': param_value}],
        ).create()
        for param_value in param_values
    ]
    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=smart_proxy_location.name)
        # Check that hosts present in the system
        for host in hosts:
            assert session.host_new.search(host.name)[0]['Name'] == host.name
        # Check that search by parameter returns only one host in the list
        for param_value, host in zip(param_values, hosts, strict=True):
            values = session.host_new.search(f'params.{param_name} = {param_value}')
            assert len(values) == 1
            assert values[0]['Name'] == host.name


def test_positive_search_by_parameter_with_prefix(
    session, smart_proxy_location, target_sat, function_org
):
    """Search by global parameter assigned to host using prefix 'not' and
    any random string as parameter value to make sure that all hosts will
    be present in the list

    :id: a4affb90-1222-4d9a-94be-213f9e5be573

    :expectedresults: All assigned hosts to organization are returned by
        search
    """
    param_name = gen_string('alpha')
    param_value = gen_string('alpha')
    search_param_value = gen_string('alphanumeric')
    parameters = [{'name': param_name, 'value': param_value}]
    param_host = target_sat.api.Host(
        organization=function_org,
        location=smart_proxy_location,
        host_parameters_attributes=parameters,
    ).create()
    additional_host = target_sat.api.Host(
        organization=function_org, location=smart_proxy_location
    ).create()
    with target_sat.ui_session() as session:
        session.organization.select(org_name=function_org.name)
        session.location.select(loc_name=smart_proxy_location.name)
        # Check that the hosts are present
        for host in [param_host, additional_host]:
            assert session.host.search(host.name)[0]['Name'] == host.name
        # Check that search by parameter with 'not' prefix returns both hosts
        values = session.host.search(f'not params.{param_name} = {search_param_value}')
        assert {value['Name'] for value in values} == {param_host.name, additional_host.name}


def test_positive_search_by_parameter_with_operator(
    session, smart_proxy_location, target_sat, function_org
):
    """Search by global parameter assigned to host using operator '<>' and
    any random string as parameter value to make sure that all hosts will
    be present in the list

    :id: 264065b7-0d04-467d-887a-0aba0d871b7c

    :expectedresults: All assigned hosts to organization are returned by
        search

    :BZ: 1463806
    """
    param_name = gen_string('alpha')
    param_value = gen_string('alpha')
    param_global_value = gen_string('numeric')
    search_param_value = gen_string('alphanumeric')
    target_sat.api.CommonParameter(name=param_name, value=param_global_value).create()
    parameters = [{'name': param_name, 'value': param_value}]
    param_host = target_sat.api.Host(
        organization=function_org,
        location=smart_proxy_location,
        host_parameters_attributes=parameters,
    ).create()
    additional_host = target_sat.api.Host(
        organization=function_org, location=smart_proxy_location
    ).create()
    with target_sat.ui_session() as session:
        session.organization.select(org_name=function_org.name)
        session.location.select(loc_name=smart_proxy_location.name)
        # Check that the hosts are present
        for host in [param_host, additional_host]:
            assert session.host.search(host.name)[0]['Name'] == host.name
        # Check that search by parameter with '<>' operator returns both hosts
        values = session.host.search(f'params.{param_name} <> {search_param_value}')
        assert {value['Name'] for value in values} == {param_host.name, additional_host.name}


def test_positive_search_with_org_and_loc_context(
    session, target_sat, function_org, function_location
):
    """Perform usual search for host, but organization and location used
    for host create procedure should have 'All capsules' checkbox selected

    :id: 2ce50df0-2b30-42cc-a40b-0e1f4fde3c6f

    :expectedresults: Search functionality works as expected and correct
        result is returned

    :BZ: 1405496

    :customerscenario: true
    """
    host = target_sat.api.Host(organization=function_org, location=function_location).create()
    with target_sat.ui_session() as session:
        session.organization.update(function_org.name, {'capsules.all_capsules': True})
        session.location.update(function_location.name, {'capsules.all_capsules': True})
        session.organization.select(org_name=function_org.name)
        session.location.select(loc_name=function_location.name)
        assert session.host.search(f'name = "{host.name}"')[0]['Name'] == host.name
        assert session.host.search(host.name)[0]['Name'] == host.name


def test_positive_search_by_org(session, smart_proxy_location, target_sat):
    """Search for host by specifying host's organization name

    :id: a3bb5bc5-cb9c-4b56-b383-f3e4d3d4d222

    :customerscenario: true

    :expectedresults: Search functionality works as expected and correct
        result is returned

    :BZ: 1447958
    """
    host = target_sat.api.Host(location=smart_proxy_location).create()
    org = host.organization.read()
    with target_sat.ui_session() as session:
        session.organization.select(org_name=ANY_CONTEXT['org'])
        session.location.select(loc_name=ANY_CONTEXT['location'])
        assert session.host.search(f'organization = "{org.name}"')[0]['Name'] == host.name


def test_positive_validate_inherited_cv_lce_ansiblerole(session, target_sat, module_host_template):
    """Create a host with hostgroup specified via CLI. Make sure host
    inherited hostgroup's lifecycle environment, content view and both
    fields are properly reflected via WebUI. Also host should be searchable by the
    inherited ansible role.

    :id: c83f6819-2649-4a8b-bb1d-ce93b2243765

    :expectedresults: Host's lifecycle environment, content view and ansible role match
       the ones specified in hostgroup.

    :customerscenario: true

    :BZ: 1391656, 2094912
    """
    SELECTED_ROLE = 'RedHatInsights.insights-client'
    cv_name = gen_string('alpha')
    lce_name = gen_string('alphanumeric')
    cv = target_sat.api_factory.cv_publish_promote(
        name=cv_name, env_name=lce_name, org_id=module_host_template.organization.id
    )
    lce = (
        target_sat.api.LifecycleEnvironment()
        .search(
            query={
                'search': f'name={lce_name} '
                f'and organization_id={module_host_template.organization.id}'
            }
        )[0]
        .read()
    )
    target_sat.cli.Ansible.roles_sync(
        {'role-names': SELECTED_ROLE, 'proxy-id': target_sat.nailgun_smart_proxy.id}
    )
    hostgroup = target_sat.cli_factory.hostgroup(
        {
            'content-view-id': cv.id,
            'lifecycle-environment-id': lce.id,
            'organization-ids': module_host_template.organization.id,
        }
    )
    result = target_sat.cli.HostGroup.ansible_roles_assign(
        {'name': hostgroup.name, 'ansible-roles': SELECTED_ROLE}
    )
    assert 'Ansible roles were assigned to the hostgroup' in result[0]['message']
    result = target_sat.cli.HostGroup.ansible_roles_add(
        {'name': hostgroup.name, 'ansible-role': SELECTED_ROLE}
    )
    assert 'Ansible role has been associated.' in result[0]['message']
    host = target_sat.cli_factory.make_host(
        {
            'architecture-id': module_host_template.architecture.id,
            'domain-id': module_host_template.domain.id,
            'hostgroup-id': hostgroup['id'],
            'location-id': module_host_template.location.id,
            'medium-id': module_host_template.medium.id,
            'operatingsystem-id': module_host_template.operatingsystem.id,
            'organization-id': module_host_template.organization.id,
            'partition-table-id': module_host_template.ptable.id,
        }
    )
    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_host_template.organization.name)
        session.location.select(loc_name=module_host_template.location.name)
        values = session.host_new.read(host['name'], ['host.lce', 'host.content_view'])
        assert values['host']['lce'] == lce.name
        assert values['host']['content_view'] == cv.name
        matching_hosts = target_sat.api.Host().search(
            query={'search': f'ansible_role="{SELECTED_ROLE}"'}
        )
        assert len(matching_hosts), 'Host not found by inherited ansible role'
        assert host.name in [host.name for host in matching_hosts]


# ------------------------------ NEW HOST UI DETAILS ----------------------------
def test_positive_read_details_page_from_new_ui(target_sat, host_ui_options):
    """Create new Host and read all its content through details page

    :id: ef0c5942-9049-11ec-8029-98fa9b6ecd5a

    :expectedresults: Host is created and has expected content
    """
    with target_sat.ui_session() as session:
        api_values, host_name = host_ui_options
        session.organization.select(api_values['host.organization'])
        session.location.select(api_values['host.location'])
        session.host_new.create(api_values)
        assert session.host_new.search(host_name)[0]['Name'] == host_name
        values = session.host_new.get_details(host_name, widget_names='overview')
        assert values['overview']['host_status']['status'] == 'All statuses OK'
        assert (
            values['overview']['details']['details']['mac_address']
            == api_values['interfaces.interface.mac']
        )
        user = session.host_new.get_details(host_name, widget_names='current_user')['current_user']
        assert values['overview']['details']['details']['host_owner'] == user
        assert values['overview']['details']['details']['comment'] == 'Host with fake data'


def test_positive_manage_table_columns(
    target_sat, test_name, ui_hosts_columns_user, current_sat_org, current_sat_location
):
    """Set custom columns of the hosts table.

    :id: e5e18982-cc43-11ed-8562-000c2989e153

    :steps:
        1. Navigate to the Hosts page.
        2. Switch to default organization and location, where is at least one host (Satellite).
        3. Set custom columns for the hosts table via the 'Manage columns' dialog.

    :expectedresults: Check if the custom columns were set properly, i.e., are displayed
        or not displayed in the table.

    :BZ: 1813274, 2212499

    :customerscenario: true
    """
    columns = {
        'Host group': False,
        'Last report': False,
        'Comment': False,
        'Installable updates': True,
        'RHEL Lifecycle status': False,
        'Registered at': True,
        'Last seen': True,
        'IPv4': True,
        'MAC': True,
        'Sockets': True,
        'Cores': True,
        'RAM': True,
        'Boot time': True,
        'Recommendations': False,
    }
    with target_sat.ui_session(
        test_name, ui_hosts_columns_user.login, ui_hosts_columns_user.password
    ) as session:
        session.organization.select(org_name=current_sat_org.name)
        session.location.select(loc_name=current_sat_location.name)
        session.all_hosts.manage_table_columns(columns)
        displayed_columns = session.host.get_displayed_table_headers()
        for column, is_displayed in columns.items():
            assert (column in displayed_columns) is is_displayed


def test_all_hosts_manage_columns(target_sat):
    """Verify that the manage columns widget changes the columns appropriately

    :id: 5e13267a-68d2-451a-ae00-6502dd5db7f4

    :expectedresults: Through the widget you can change the columns on the All Hosts page

    :CaseComponent: Hosts

    :Team: Proton

    :Verifies: SAT-19064
    """
    columns = {
        'Host group': True,
        'Last report': True,
        'Comment': True,
        'IPv4': True,
        'MAC': True,
        'Sockets': True,
        'Cores': True,
        'RAM': True,
        'Boot time': True,
    }
    with target_sat.ui_session() as session:
        # Small workaround for an existing bug, reloads the page
        session.all_hosts.get_displayed_table_headers()
        wait_for(lambda: session.browser.refresh(), timeout=5)
        session.all_hosts.manage_table_columns(columns)
        displayed_columns = session.all_hosts.get_displayed_table_headers()
        for column, is_displayed in columns.items():
            assert (column in displayed_columns) is is_displayed


def test_positive_host_details_read_templates(
    session, target_sat, current_sat_org, current_sat_location
):
    """Check if all assigned host provisioning templates are correctly reported
    in host detail / Details tab / Provisioning templates card.

    :id: 43ca722e-d28a-11ed-8970-000c2989e153

    :steps:
        1. Go to Hosts page and select the Satellite host machine.
        2. Go to the Details tab.
        3. Gather all names from the `Provisioning templates` card.
        4. Compare them with the host provisioning templates obtained via API.

    :expectedresults: Provisioning templates reported via API and in UI should match.

    :BZ: 2128038

    :customerscenario: true
    """
    host = target_sat.api.Host().search(query={'search': f'name={target_sat.hostname}'})[0]
    api_templates = [template['name'] for template in host.list_provisioning_templates()]
    with target_sat.ui_session() as session:
        session.organization.select(org_name=current_sat_org.name)
        session.location.select(loc_name=current_sat_location.name)
        host_detail = session.host_new.get_details(target_sat.hostname, widget_names='details')
        ui_templates = [
            row['column1'].strip()
            for row in host_detail['details']['provisioning_templates']['templates_table']
        ]
    assert set(api_templates) == set(ui_templates)


@pytest.mark.rhel_ver_match('N-1')
@pytest.mark.no_containers
@pytest.mark.parametrize(
    'module_repos_collection_with_setup',
    [{'YumRepository': {'url': settings.repos.yum_3.url}}],
    ids=['yum_3'],
    indirect=True,
)
def test_positive_update_delete_package(
    session,
    target_sat,
    rhel_contenthost,
    module_repos_collection_with_setup,
    module_org,
):
    """Update a package on a host using the new Content tab

    :id: ffc19a40-85f4-4894-a18b-f6d88b2ce377

    :steps:
        1. Navigate to the Content tab.
        2. Disable repository set
        3. Package from repository cannot be installed
        4. Enable repository set
        5. Install a package on a registered host.
        6. Downgrade package version
        7. Check if the package is in an upgradable state.
        8. Select package and upgrade via rex.
        9. Delete the package

    :expectedresults: The package is updated and deleted
    """
    client = rhel_contenthost
    client.add_rex_key(target_sat)
    module_repos_collection_with_setup.setup_virtual_machine(client, enable_custom_repos=True)
    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=DEFAULT_LOC)
        product_name = module_repos_collection_with_setup.custom_product.name

        session.host_new.override_repo_sets(client.hostname, product_name, "Override to disabled")
        repos = session.host_new.get_repo_sets(client.hostname, product_name)
        assert repos[0]['Status'] == 'Disabled'
        result = client.run(f'yum install -y {FAKE_7_CUSTOM_PACKAGE}')
        assert result.status != 0
        session.host_new.override_repo_sets(client.hostname, product_name, "Override to enabled")
        repos = session.host_new.get_repo_sets(client.hostname, product_name)
        assert repos[0]['Status'] == 'Enabled'
        # refresh repos on system
        client.run('subscription-manager repos')
        # install package
        session.host_new.install_package(client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME)
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Install package(s) on {client.hostname}'),
            search_rate=4,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        session.browser.refresh()
        packages = session.host_new.get_packages(client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME)
        assert len(packages) == 1
        assert packages[0]['Package'] == FAKE_8_CUSTOM_PACKAGE_NAME
        assert 'Up-to date' in packages[0]['Status']
        result = client.run(f'rpm -q {FAKE_8_CUSTOM_PACKAGE}')
        assert result.status == 0

        # downgrade package version
        client.run(f'yum -y downgrade {FAKE_8_CUSTOM_PACKAGE_NAME}')
        result = client.run(f'rpm -q {FAKE_8_CUSTOM_PACKAGE_NAME}')
        assert result.status == 0

        # this should reload page to update packages table
        session.host_new.get_details(client.hostname, widget_names='overview')
        session.browser.refresh()
        # filter packages
        packages = session.host_new.get_packages(client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME)
        assert len(packages) == 1
        assert packages[0]['Package'] == FAKE_8_CUSTOM_PACKAGE_NAME
        assert 'Upgradable' in packages[0]['Status']

        # update package
        session.host_new.apply_package_action(
            client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME, "Upgrade via remote execution"
        )
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Update package(s) {FAKE_8_CUSTOM_PACKAGE_NAME} on {client.hostname}'),
            search_rate=2,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        # this should reload page to update packages table
        session.host_new.get_details(client.hostname, widget_names='overview')
        session.browser.refresh()
        packages = session.host_new.get_packages(client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME)
        assert 'Up-to date' in packages[0]['Status']
        # remove package
        session.host_new.apply_package_action(client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME, "Remove")
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Remove package(s) {FAKE_8_CUSTOM_PACKAGE_NAME} on {client.hostname}'),
            search_rate=2,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        # this should reload page to update packages table
        session.host_new.get_details(client.hostname, widget_names='overview')
        session.browser.refresh()
        packages = session.host_new.get_packages(client.hostname, FAKE_8_CUSTOM_PACKAGE_NAME)
        assert not packages
        result = client.run(f'rpm -q {FAKE_8_CUSTOM_PACKAGE}')
        assert result.status != 0


@pytest.mark.rhel_ver_match('N-1')
@pytest.mark.no_containers
@pytest.mark.parametrize(
    'module_repos_collection_with_setup',
    [{'YumRepository': {'url': settings.repos.yum_3.url}}],
    ids=['yum3'],
    indirect=True,
)
def test_positive_apply_erratum(
    session,
    target_sat,
    rhel_contenthost,
    module_repos_collection_with_setup,
    module_org,
):
    """Apply an erratum on a host using the new Errata tab

    :id: 328e629a-f261-4dc1-ad6f-def27e2fcf07

    :setup:
        1. Valid yum repo with an applicable erratum.

    :steps:
        1. Install a package on a registered host.
        2. Check the Errata card on the Overview tab
        3. Navigate to the Errata tab.
        4. Check for applicable errata.
        5. Select errata and apply via rex.

    :expectedresults: The erratum is applied
    """
    # install package
    client = rhel_contenthost
    client.add_rex_key(target_sat)
    module_repos_collection_with_setup.setup_virtual_machine(client, enable_custom_repos=True)
    errata_id = settings.repos.yum_3.errata[25]
    client.run(f'yum install -y {FAKE_7_CUSTOM_PACKAGE}')
    result = client.run(f'rpm -q {FAKE_7_CUSTOM_PACKAGE}')
    assert result.status == 0
    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=DEFAULT_LOC)
        assert session.host_new.search(client.hostname)[0]['Name'] == client.hostname
        # read widget on overview page
        values = session.host_new.get_details(client.hostname, widget_names='overview')['overview']
        assert values['installable_errata']['security_advisory'] == '1 security advisory'
        assert values['installable_errata']['enhancements'] == '1 enhancement'
        # read errata tab
        values = session.host_new.get_details(client.hostname, widget_names='content.errata')
        assert len(values['content']['errata']['table']) == 2
        # filter just security erratum
        erratas = session.host_new.get_errata_by_type(client.hostname, 'Security')
        assert len(erratas['content']['errata']['table']) == 1
        assert erratas['content']['errata']['table'][0]['Errata'] == errata_id
        # apply errata
        session.host_new.apply_erratas(client.hostname, f"errata_id == {errata_id}")
        task_result = target_sat.wait_for_tasks(
            search_query=(
                f'"Install errata errata_id == {errata_id.lower()} '
                f'and type=security on {client.hostname}"'
            ),
            search_rate=2,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        # verify
        session.browser.refresh()
        values = session.host_new.get_details(client.hostname, widget_names='content.errata')
        assert 'table' not in values['content']['errata']
        result = client.run(
            'yum update --assumeno --security | grep "No packages needed for security"'
        )
        assert result.status == 1


@pytest.mark.e2e
@pytest.mark.upgrade
@pytest.mark.rhel_ver_match('N-1')
@pytest.mark.no_containers
@pytest.mark.parametrize(
    'module_repos_collection_with_setup',
    [{'YumRepository': {'url': settings.repos.module_stream_1.url}}],
    ids=['module_stream_1'],
    indirect=True,
)
def test_positive_crud_module_streams(
    session,
    target_sat,
    rhel_contenthost,
    module_repos_collection_with_setup,
    module_org,
):
    """CRUD test for the Module streams new UI tab

    :id: 9800a006-49cc-4c0a-aed8-6a32c4bf0eab

    :setup:
        1. Valid yum repo with Module Streams.

    :steps:
        1. Create Yum Repository which contains module-streams as URL
        2. Enable Module stream
        3. Install Module stream
        4. Delete the Module stream
        5. Reset the Module stream

    :expectedresults: Module streams can be enabled, installed, removed and reset using the new UI.
    """
    module_name = 'duck'
    client = rhel_contenthost
    client.add_rex_key(target_sat)
    module_repos_collection_with_setup.setup_virtual_machine(client, enable_custom_repos=True)
    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=DEFAULT_LOC)
        streams = session.host_new.get_module_streams(client.hostname, module_name)
        assert streams[0]['Name'] == module_name
        assert streams[0]['State'] == 'Default'
        # enable module stream
        session.host_new.apply_module_streams_action(client.hostname, module_name, "Enable")
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Module enable {module_name} on {client.hostname}'),
            search_rate=5,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        session.browser.refresh()
        streams = session.host_new.get_module_streams(client.hostname, module_name)
        assert streams[0]['State'] == 'Enabled'
        assert streams[0]['Installation status'] == 'Not installed'

        # install
        session.host_new.apply_module_streams_action(client.hostname, module_name, "Install")
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Module install {module_name} on {client.hostname}'),
            search_rate=5,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        session.browser.refresh()
        streams = session.host_new.get_module_streams(client.hostname, module_name)
        assert streams[0]['Installation status'] == 'Up-to-date'

        # remove
        # No space in 'RemoveUninstall and reset' is intentional,
        # to match the button that has small text under itself in the DropDownItem
        session.host_new.apply_module_streams_action(
            client.hostname, module_name, 'RemoveUninstall and reset'
        )
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Module remove {module_name} on {client.hostname}'),
            search_rate=5,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        session.browser.refresh()
        streams = session.host_new.get_module_streams(client.hostname, module_name)
        assert streams[0]['State'] == 'Enabled'
        assert streams[0]['Installation status'] == 'Not installed'

        # No space in 'ResetReset to the default state' is intentional,
        # to match the button that has small text under itself in the DropDownItem
        session.host_new.apply_module_streams_action(
            client.hostname, module_name, "ResetReset to the default state"
        )
        task_result = target_sat.wait_for_tasks(
            search_query=(f'Module reset {module_name} on {client.hostname}'),
            search_rate=5,
            max_tries=60,
        )
        task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
        assert task_status['result'] == 'success'
        session.browser.refresh()
        streams = session.host_new.get_module_streams(client.hostname, module_name)
        assert streams[0]['State'] == 'Default'
        assert streams[0]['Installation status'] == 'Not installed'


# ------------------------------ PUPPET ENABLED SAT TESTS ----------------------------
@pytest.fixture(scope='module')
def module_puppet_enabled_proxy_with_loc(
    session_puppet_enabled_sat, module_puppet_loc, session_puppet_enabled_proxy
):
    session_puppet_enabled_proxy.location.append(
        session_puppet_enabled_sat.api.Location(id=module_puppet_loc.id)
    )
    session_puppet_enabled_proxy.update(['location'])


@pytest.mark.skipif((not settings.robottelo.REPOS_HOSTING_URL), reason='Missing repos_hosting_url')
@pytest.mark.usefixtures('module_puppet_enabled_proxy_with_loc')
def test_positive_create_with_puppet_class(
    session_puppet_enabled_sat,
    module_puppet_loc,
    module_puppet_org,
    module_env_search,
    module_import_puppet_module,
    module_puppet_enabled_proxy_with_loc,
):
    """Create new Host with puppet class assigned to it

    :id: d883f169-1105-435c-8422-a7160055734a

    :expectedresults: Host is created and contains correct puppet class
    """

    host_template = session_puppet_enabled_sat.api.Host(
        organization=module_puppet_org, location=module_puppet_loc
    )
    host_template.create_missing()
    host_name = f'{host_template.name}.{host_template.domain.name}'
    os_name = f'{host_template.operatingsystem.name} {host_template.operatingsystem.major}'
    values = {
        'host.name': host_template.name,
        'host.organization': host_template.organization.name,
        'host.location': host_template.location.name,
        'host.lce': ENVIRONMENT,
        'host.content_view': DEFAULT_CV,
        'operating_system.architecture': host_template.architecture.name,
        'operating_system.operating_system': os_name,
        'operating_system.media_type': 'All Media',
        'operating_system.media': host_template.medium.name,
        'operating_system.ptable': host_template.ptable.name,
        'operating_system.root_password': host_template.root_pass,
        'interfaces.interface.interface_type': 'Interface',
        'interfaces.interface.device_identifier': gen_string('alpha'),
        'interfaces.interface.mac': host_template.mac,
        'interfaces.interface.domain': host_template.domain.name,
        'interfaces.interface.primary': True,
        'interfaces.interface.interface_additional_data.virtual_nic': False,
        'parameters.global_params': None,
        'parameters.host_params': None,
        'additional_information.comment': 'Host with fake data',
        'host.puppet_environment': module_env_search.name,
        'puppet_enc.classes.assigned': [module_import_puppet_module['puppet_class']],
    }
    with session_puppet_enabled_sat.ui_session() as session:
        session.organization.select(org_name=module_puppet_org.name)
        session.location.select(loc_name=ANY_CONTEXT['location'])
        session.host.create(values)
        assert session.host.search(host_name)[0]['Name'] == host_name
        values = session.host.read(host_name, widget_names='puppet_enc')
        assert len(values['puppet_enc']['classes']['assigned']) == 1
        assert (
            values['puppet_enc']['classes']['assigned'][0]
            == module_import_puppet_module['puppet_class']
        )


def test_positive_inherit_puppet_env_from_host_group_when_create(
    session_puppet_enabled_sat, module_env_search, module_puppet_org, module_puppet_loc
):
    """Host group puppet environment is inherited to host in create
    procedure

    :id: 05831ecc-3132-4eb7-ad90-155470f331b6

    :customerscenario: true

    :expectedresults: Expected puppet environment is inherited to the form

    :BZ: 1414914
    """

    hg_name = gen_string('alpha')
    with session_puppet_enabled_sat.ui_session() as session:
        session.organization.select(org_name=module_puppet_org.name)
        session.location.select(loc_name=module_puppet_loc.name)
        session.hostgroup.create(
            {'host_group.name': hg_name, 'host_group.puppet_environment': module_env_search.name}
        )
        assert session.hostgroup.search(hg_name)[0]['Name'] == hg_name
        values = session.host.helper.read_create_view(
            {}, ['host.puppet_environment', 'host.inherit_puppet_environment']
        )
        assert not values['host']['puppet_environment']
        assert values['host']['inherit_puppet_environment'] is False
        values = session.host.helper.read_create_view(
            {'host.hostgroup': hg_name},
            ['host.puppet_environment', 'host.inherit_puppet_environment'],
        )
        assert values['host']['puppet_environment'] == module_env_search.name
        assert values['host']['inherit_puppet_environment'] is True
        values = session.host.helper.read_create_view(
            {'host.inherit_puppet_environment': False},
            ['host.puppet_environment', 'host.inherit_puppet_environment'],
        )
        assert values['host']['puppet_environment'] == module_env_search.name
        assert values['host']['inherit_puppet_environment'] is False


@pytest.mark.usefixtures('module_puppet_enabled_proxy_with_loc')
def test_positive_set_multi_line_and_with_spaces_parameter_value(
    session_puppet_enabled_sat,
    module_puppet_org,
    module_puppet_loc,
    module_puppet_published_cv,
    module_puppet_lce_library,
):
    """Check that host parameter value with multi-line and spaces is
    correctly represented in yaml format

    :id: d72b481d-2279-4478-ab2d-128f92c76d9c

    :customerscenario: true

    :expectedresults:
        1. parameter is correctly represented in yaml format without
           line break (special chars should be escaped)
        2. host parameter value is the same when restored from yaml format

    :BZ: 1315282
    """
    host_template = session_puppet_enabled_sat.api.Host(
        organization=module_puppet_org, location=module_puppet_loc
    )
    host_template.create_missing()

    param_name = gen_string('alpha').lower()
    # long string that should be escaped and affected by line break with
    # yaml dump by default
    param_value = (
        'auth                          include              '
        'password-auth\r\n'
        'account     include                  password-auth'
    )
    host = session_puppet_enabled_sat.api.Host(
        organization=host_template.organization,
        architecture=host_template.architecture,
        domain=host_template.domain,
        location=host_template.location,
        mac=host_template.mac,
        medium=host_template.medium,
        operatingsystem=host_template.operatingsystem,
        ptable=host_template.ptable,
        root_pass=host_template.root_pass,
        content_facet_attributes={
            'content_view_id': module_puppet_published_cv.id,
            'lifecycle_environment_id': module_puppet_lce_library.id,
        },
    ).create()
    with session_puppet_enabled_sat.ui_session() as session:
        session.organization.select(org_name=module_puppet_org.name)
        session.location.select(loc_name=module_puppet_loc.name)
        session.host.update(
            host.name, {'parameters.host_params': [dict(name=param_name, value=param_value)]}
        )
        yaml_text = session.host.read_yaml_output(host.name)
        # ensure parameter value is represented in yaml format without
        # line break (special chars should be escaped)
        assert param_value.encode('unicode_escape') in bytes(yaml_text, 'utf-8')
        # host parameter value is the same when restored from yaml format
        yaml_content = yaml.load(yaml_text, yaml.SafeLoader)
        host_parameters = yaml_content.get('parameters')
        assert host_parameters
        assert param_name in host_parameters
        assert host_parameters[param_name] == param_value


@pytest.mark.pit_client
@pytest.mark.rhel_ver_match('[7,8,9]')
def test_positive_tracer_enable_reload(tracer_install_host, target_sat):
    """Using the new Host UI,enable tracer and verify that the page reloads

    :id: c9ebd4a8-6db3-4d0e-92a2-14951c26769b

    :CaseComponent: katello-tracer

    :Team: Endeavour

    :steps:
        1. Register a RHEL host to Satellite.
        2. Prepare katello-tracer to be installed
        3. Navigate to the Traces tab in New Host UI
        4. Enable tracer using the popup

    :expectedresults: The Tracer tab message updates accordingly during the process, and displays
        the state the correct Title
    """
    host = (
        target_sat.api.Host().search(query={'search': tracer_install_host.hostname})[0].read_json()
    )
    with target_sat.ui_session() as session:
        session.organization.select(host['organization_name'])
        tracer_title = session.host_new.get_tracer_tab_title(tracer_install_host.hostname)
        assert tracer_title == "Traces are not enabled"
        session.host_new.enable_tracer(tracer_install_host.hostname)
        timestamp = (datetime.now(UTC) - timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M')
        target_sat.wait_for_tasks(
            search_query='action = "Run hosts job: Install package(s) katello-host-tools-tracer"'
            f' and started_at >= "{timestamp}"',
            search_rate=15,
            max_tries=10,
        )
        tracer_title = session.host_new.get_tracer_tab_title(tracer_install_host.hostname)
        assert tracer_title == "No applications to restart"


def test_all_hosts_delete(target_sat, function_org, function_location):
    """Create a host and delete it through All Hosts UI

    :id: 42b4560c-bb57-4c58-928e-e5fd5046b93f

    :expectedresults: Successful deletion of a host through the table dropdown

    :CaseComponent:Hosts

    :Team: Proton
    """
    host = target_sat.api.Host(organization=function_org, location=function_location).create()
    with target_sat.ui_session() as session:
        session.organization.select(function_org.name)
        session.location.select(function_location.name)
        # Get current headers
        headers = session.all_hosts.get_displayed_table_headers()
        stripped_headers = tuple(
            header for header in headers if header is not None and header != 'Name'
        )
        wait_for(lambda: session.browser.refresh(), timeout=5)
        # Make sure there is only Name column displayed
        session.all_hosts.manage_table_columns({header: False for header in stripped_headers})
        assert session.all_hosts.delete(host.name)
        # Get table to original state
        wait_for(lambda: session.browser.refresh(), timeout=5)
        session.all_hosts.manage_table_columns({header: True for header in stripped_headers})


def test_all_hosts_bulk_delete(target_sat, function_org, function_location):
    """Create several hosts, and delete them via Bulk Actions in All Hosts UI

    :id: af1b4a66-dd83-47c3-904b-e8627119cc53

    :expectedresults: Successful deletion of multiple hosts at once through Bulk Action

    :CaseComponent:Hosts

    :Team: Proton
    """
    for _ in range(10):
        target_sat.api.Host(organization=function_org, location=function_location).create()
    with target_sat.ui_session() as session:
        session.organization.select(function_org.name)
        session.location.select(function_location.name)
        assert session.all_hosts.bulk_delete_all()


def test_all_hosts_bulk_cve_reassign(
    target_sat, module_org, module_location, module_lce, module_cv
):
    """Create several hosts, and bulk assigns them a new CVE via All Hosts UI

    :id: 9acb3cf3-c042-4cc7-abdf-31f9a7f1fff0

    :steps:
        1.Create 2 Hosts, and register them to the first LCE.
        2.Create a second LCE and promote the CV from the first LCE to this one.
        3.Using the All Hosts UI, reassign both hosts from the first LCE to the second LCE.

    :expectedresults: Both hosts are successfully assigned to a new LCE and CV

    :CaseComponent: Hosts

    :Team: Proton
    """
    lce2 = target_sat.api.LifecycleEnvironment(organization=module_org).create()
    module_cv = target_sat.api.ContentView(id=module_cv.id).read()
    module_cv = cv_publish_promote(target_sat, module_org, module_cv, module_lce)['content-view']
    module_cv = cv_publish_promote(target_sat, module_org, module_cv, lce2)['content-view']
    for _ in range(3):
        target_sat.api.Host(
            organization=module_org,
            location=module_location,
            content_facet_attributes={
                'content_view_id': module_cv.id,
                'lifecycle_environment_id': module_lce.id,
            },
        ).create()
    with target_sat.ui_session() as session:
        session.organization.select(module_org.name)
        session.location.select(module_location.name)
        try:
            headers = session.all_hosts.get_displayed_table_headers()
            if 'Lifecycle environment' not in headers:
                session.all_hosts.manage_table_columns(
                    {
                        'Lifecycle environment': True,
                    }
                )
            pre_table = session.all_hosts.read_table()
            for row in pre_table:
                assert row['Lifecycle environment'] == module_lce.name
            session.all_hosts.manage_cve(lce_name=lce2.name, cv_name=module_cv.name)
            wait_for(lambda: session.browser.refresh(), timeout=5)
            post_table = session.all_hosts.read_table()
            for row in post_table:
                assert row['Lifecycle environment'] == lce2.name
        finally:
            session.all_hosts.manage_table_columns(
                {
                    'Lifecycle environment': False,
                }
            )


def test_all_hosts_redirect_button(target_sat):
    """Verify that the New UI button on the old Host page correctly redirects
    to the All Hosts UI

    :id: 7256f6d0-3ad9-471c-9e3e-bd41cc00a217

    :expectedresults: New UI Button redirects to All Hosts page

    :CaseComponent: Hosts

    :Team: Proton
    """
    with target_sat.ui_session() as session:
        url = session.host.new_ui_button()
        assert "/new/hosts" in url


def test_all_hosts_bulk_build_management(target_sat, function_org, function_location):
    """Create several hosts, and manage them via Build Management in All Host UI

    :id: fff71945-6534-45cf-88a6-16b25c060f0a

    :expectedresults: Build Management dropdown in All Hosts UI works properly.

    :CaseComponent:Hosts

    :Team: Proton
    """
    for _ in range(3):
        target_sat.api.Host(organization=function_org, location=function_location).create()
    with target_sat.ui_session() as session:
        session.organization.select(function_org.name)
        session.location.select(function_location.name)
        assert 'Success alert: Built 3 hosts' in session.all_hosts.build_management()
        assert (
            'Success alert: 3 hosts set to build and rebooting.'
            in session.all_hosts.build_management(reboot=True)
        )
        assert 'Rebuilt configuration for 3 hosts' in session.all_hosts.build_management(
            rebuild=True
        )


@pytest.fixture(scope='module')
def change_content_source_prep(
    module_target_sat,
    module_sca_manifest_org,
    module_capsule_configured,
    module_location,
):
    """
    This fixture sets up all the necessary entities for tests
    exercising the Change of the hosts's content source.

     It creates a new product in the organization,
     creates a new repository in the product,
     creates a new lce,
     creates a new CV in the organization, adds the repository to the CV,
     publishes the CV, and promotes the published version to the lifecycle environment,
     creates a new activation key for the CV in the lce,
     registers the RHEL content host with the activation key,
     updates the capsule's taxonomies
     adds the lifecycle environment to the capsule's content.

     Fixture returns module_target_sat, org, lce, capsule, content_view, loc, ak
    """
    product_name, lce_name = (gen_string('alpha') for _ in range(2))

    org = module_sca_manifest_org
    loc = module_location

    product = module_target_sat.api.Product(
        name=product_name,
        organization=org.id,
    ).create()

    repository = module_target_sat.api.Repository(
        product=product,
        content_type=REPO_TYPE['file'],
        url=CUSTOM_FILE_REPO,
    ).create()

    lce = module_target_sat.cli_factory.make_lifecycle_environment(
        {'name': lce_name, 'organization-id': org.id}
    )

    # Create CV
    content_view = module_target_sat.api.ContentView(organization=org.id).create()
    # Add repos to CV
    content_view.repository = [repository]
    content_view = content_view.update(['repository'])
    # Publish that CV and promote it
    content_view.publish()
    content_view.read().version[0].promote(data={'environment_ids': lce.id})

    ak = module_target_sat.api.ActivationKey(
        content_view=content_view, organization=org.id, environment=lce.id
    ).create()

    # Edit capsule's taxonomies
    capsule = module_target_sat.cli.Capsule.update(
        {
            'name': module_capsule_configured.hostname,
            'organization-ids': org.id,
            'location-ids': loc.id,
        }
    )

    module_target_sat.cli.Capsule.content_add_lifecycle_environment(
        {
            'id': module_capsule_configured.nailgun_capsule.id,
            'organization-id': org.id,
            'lifecycle-environment': lce.name,
        }
    )

    return module_target_sat, org, lce, capsule, content_view, loc, ak


@pytest.mark.no_containers
@pytest.mark.rhel_ver_match('[789]')
def test_change_content_source(session, change_content_source_prep, rhel_contenthost):
    """
    This test exercises different ways to change host's content source

    :id: 5add68c3-16b1-496d-9b24-f5388013351d

    :expectedresults: Job invocation page should be correctly generated
        by the change content source action, generated script should also be correct

    :CaseComponent:Hosts

    :Team: Proton

    :BlockedBy: SAT-41505
    """

    module_target_sat, org, lce, capsule, content_view, loc, ak = change_content_source_prep

    rhel_contenthost.register(org, loc, ak.name, module_target_sat)

    with module_target_sat.ui_session() as session:
        session.organization.select(org_name=org.name)
        session.location.select(loc_name=ANY_CONTEXT['location'])

        # STEP 1: Test the part where you use "Update hosts manually" button
        # Set the content source to the checked-out capsule
        # Check that generated script contains correct name of new content source
        rhel_contenthost_pre_values = rhel_contenthost.nailgun_host.content_facet_attributes
        generated_script = session.host.change_content_source_get_script(
            entities_list=[
                rhel_contenthost.hostname,
            ],
            content_source=capsule[0]['name'],
            lce=lce.name,
            content_view=content_view.name,
        )
        rhel_contenthost_post_values = rhel_contenthost.nailgun_host.content_facet_attributes
        content_source_from_script = re.search(r'--server.hostname=\"(.*?)\"', generated_script)

        assert content_source_from_script.group(1) == capsule[0]['name']
        assert rhel_contenthost_post_values['content_source']['name'] == capsule[0]['name']
        assert rhel_contenthost_post_values['content_view']['name'] == content_view.name
        assert rhel_contenthost_post_values['lifecycle_environment']['name'] == lce.name

        session.browser.refresh()

        # Step 2: Test the part where you use "Run job invocation" button
        # Change the rhel_contenthost's content source back to what it was before STEP 1
        # Check the prefilled job invocation page
        session.host.change_content_source(
            entities_list=[
                rhel_contenthost.hostname,
            ],
            content_source=rhel_contenthost_pre_values['content_source']['name'],
            lce=rhel_contenthost_pre_values['lifecycle_environment']['name'],
            content_view=rhel_contenthost_pre_values['content_view']['name'],
            run_job_invocation=True,
        )
        # Getting the data from the prefilled job invocation form
        selected_category_and_template = session.jobinvocation.get_job_category_and_template()
        selected_targeted_hosts = session.jobinvocation.get_targeted_hosts()

        assert selected_category_and_template['job_category'] == 'Katello'
        assert (
            selected_category_and_template['job_template']
            == 'Configure host for new content source'
        )
        assert selected_targeted_hosts['selected_hosts'] == [rhel_contenthost.hostname]

        session.jobinvocation.submit_prefilled_view()
        rhel_contenthost_post_values = rhel_contenthost.nailgun_host.content_facet_attributes
        assert (
            rhel_contenthost_post_values['content_source']['name']
            == rhel_contenthost_pre_values['content_source']['name']
        )
        assert (
            rhel_contenthost_post_values['content_view']['name']
            == rhel_contenthost_post_values['content_view']['name']
        )
        assert (
            rhel_contenthost_post_values['lifecycle_environment']['name']
            == rhel_contenthost_post_values['lifecycle_environment']['name']
        )


@pytest.mark.rhel_ver_match('8')
def test_positive_page_redirect_after_update(target_sat, rhel_contenthost, setup_content):
    """Check that page redirects correctly after editing a host without making any changes.

    :id: 29c3397e-0010-11ef-bca4-000c2989e153

    :steps:
        1. Go to All Hosts page.
        2. Edit a host with FQDN.
        3. Submit the host edit dialog without making any changes.

    :expectedresults: The page should be redirected to the host details page.

    :BZ: 2166303
    """
    ak, org, _ = setup_content
    rhel_contenthost.register(org, None, ak.name, target_sat)
    column = {'Name': True, 'Host group': True, 'OS': True, 'Owner': True, 'Last report': True}
    with target_sat.ui_session() as session:
        session.organization.select(org.name)
        headers = session.all_hosts.get_displayed_table_headers()
        columns = {**column, **{h: False for h in headers if h is not None and h not in column}}
        session.all_hosts.manage_table_columns(columns)
        session.host_new.update(rhel_contenthost.hostname, {})
        assert 'page-not-found' not in session.browser.url
        assert rhel_contenthost.hostname in session.browser.url


@pytest.mark.no_containers
@pytest.mark.rhel_ver_match([settings.content_host.default_rhel_version])
def test_host_status_honors_taxonomies(
    target_sat,
    test_name,
    rhel_contenthost,
    function_ak_with_cv,
    function_org,
    default_location,
    default_org,
    default_org_lce,
):
    """Check that host status counts in Monitor -> Host Statuses show only hosts that the user has permissions to

    :id: 2c4e6df7-c17e-4074-b691-4d8e2efda062
    :steps:
        1. In a non-default organization, create a user
        2. As that user, check that host count is 0 in Monitor -> Host Statuses
        3. Add a host to the non-default org
        4. As that user, check that host count is 1 in Monitor -> Host Statuses

    :expectedresults: First, the user can't see any host, then they can see one host
    """
    lce = default_org_lce
    # Create content view environment for the default org
    content_view = target_sat.api.ContentView(organization=default_org).create()
    content_view.publish()
    published_cv = content_view.read()
    content_view_version = published_cv.version[0]
    content_view_version.promote(data={'environment_ids': lce.id})

    # default_org != function_org
    default_org_ak_name = gen_string('alpha')
    target_sat.cli_factory.make_activation_key(
        {
            'name': default_org_ak_name,
            'organization-id': default_org.id,
            'lifecycle-environment-id': lce.id,
            'content-view-id': published_cv.id,
        }
    )['name']
    # register the host to default_org
    assert (
        rhel_contenthost.register(
            default_org, default_location, default_org_ak_name, target_sat
        ).status
        == 0
    )
    host_id = target_sat.cli.Host.info({'name': rhel_contenthost.hostname})['id']
    password = gen_string('alpha')
    login = gen_string('alpha')
    # the user is in function_org
    target_sat.cli.User.create(
        {
            'organization-id': function_org.id,
            'location-id': default_location.id,
            'auth-source': 'Internal',
            'password': password,
            'mail': 'root@localhost',
            'login': login,
            'roles': ROLES,
        }
    )
    with target_sat.ui_session(test_name, user=login, password=password) as session:
        statuses = session.host.host_statuses()
    assert all(int(status['count'].split(': ')[1]) == 0 for status in statuses)
    # register the host to function_org
    assert rhel_contenthost.unregister().status == 0
    target_sat.cli.Host.delete({'id': host_id})
    assert (
        rhel_contenthost.register(
            function_org, default_location, function_ak_with_cv.name, target_sat
        ).status
        == 0
    )
    with target_sat.ui_session(test_name, user=login, password=password) as session:
        statuses = session.host.host_statuses()
    assert len([status for status in statuses if int(status['count'].split(': ')[1]) != 0]) == 1


@pytest.mark.parametrize(
    'module_repos_collection_with_setup',
    [
        {
            'distro': 'rhel8',
            'YumRepository': {'url': settings.repos.yum_3.url},
        }
    ],
    ids=['yum3'],
    indirect=True,
)
@pytest.mark.parametrize('finish_via', ['rex', 'custom_rex'])
@pytest.mark.parametrize(
    'package_management_action',
    [
        'install_1_pckg',
        'install_2_pckgs',
        'upgrade_1_pckg',
        'upgrade_all_pckgs',
        'remove_1_pckg',
        'remove_2_pckgs',
    ],
)
@pytest.mark.parametrize('number_of_hosts', [1, 2], ids=['1_host', '2_hosts'])
@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
@pytest.mark.no_containers
def test_positive_manage_packages(
    request,
    module_target_sat,
    mod_content_hosts,
    module_repos_collection_with_setup,
    number_of_hosts,
    package_management_action,
    finish_via,
):
    """
    This test is testing the new Satellite feature - Managing packages on hosts via the new All hosts UI.
    It is highly parametrized so it can test various package management actions on various hosts.
    Test cases are defined by the combination of the following parameters:
    - number_of_hosts: 1 or 2
    - package_management_action: install_1_pckg, install_2_pckgs, upgrade_1_pckg, upgrade_all_pckgs, remove_1_pckg, remove_2_pckgs
    - finish_via: rex or custom_rex
    All this leads to 2 * 6 * 2 = 24 test cases in total.

    :id: 1d6760ca-9c7e-4267-9a4b-3f91d50c8eb1

    :steps:
        1. Setup hosts on Satellite, override reposets to enabled and refresh applicability so the package profile is updated
        2. Setup all the control flags for airgun entity
        3. Setup packages on the selected hosts so they can be managed in the next steps
        4. Run the selected package management action on the selected hosts
        5. Wait for the specific tasks to finish
        6. Assert the results

    :expectedresults: Various package management actions should run successfully on various hosts

    :CaseComponent: Hosts

    :parametrized: yes

    :Team: Proton
    """

    packages = ['panda', 'seal']

    for host in mod_content_hosts:
        host.add_rex_key(module_target_sat)
        module_repos_collection_with_setup.setup_virtual_machine(host)

    product_name = module_repos_collection_with_setup.custom_product.name

    with module_target_sat.ui_session() as session:
        session.organization.select(module_repos_collection_with_setup.organization['name'])

        for host in mod_content_hosts:
            if (
                session.host_new.get_repo_sets(host.hostname, product_name)[0]['Status']
                == 'Disabled'
            ):
                session.host_new.override_repo_sets(
                    host.hostname, product_name, "Override to enabled"
                )
                session.host_new.refresh_applicability(host.hostname)
                latest_refresh_applicability_id = int(
                    module_target_sat.cli.JobInvocation().list()[0]['id']
                )
                module_target_sat.wait_for_tasks(
                    search_query=(
                        f'action: "Upload package profile for a host" and resource_id = {latest_refresh_applicability_id}'
                    )
                )

        # Define hosts to test based on the number of hosts
        hosts_to_test = []
        match number_of_hosts:
            case 1:
                hosts_to_test = [mod_content_hosts[0]]
            case 2:
                hosts_to_test = mod_content_hosts

        # Set default value to management action flags
        upgrade_all_packages_flag = upgrade_packages_flag = install_packages_flag = (
            remove_packages_flag
        ) = False

        # Set flags according to current parametrization
        match package_management_action:
            case 'install_1_pckg':
                install_packages_flag = True
                packages_to_install = [packages[0]]
                packages_to_upgrade = None
                packages_to_remove = None

            case 'install_2_pckgs':
                install_packages_flag = True
                packages_to_install = packages
                packages_to_upgrade = None
                packages_to_remove = None

            case 'upgrade_1_pckg':
                upgrade_packages_flag = True
                packages_to_install = None
                packages_to_upgrade = [packages[0]]
                packages_to_remove = None

            case 'upgrade_all_pckgs':
                upgrade_packages_flag = True
                upgrade_all_packages_flag = True
                packages_to_install = None
                packages_to_upgrade = packages
                packages_to_remove = None

            case 'remove_1_pckg':
                remove_packages_flag = True
                packages_to_install = None
                packages_to_upgrade = None
                packages_to_remove = [packages[0]]

            case 'remove_2_pckgs':
                remove_packages_flag = True
                packages_to_install = None
                packages_to_upgrade = None
                packages_to_remove = packages

        # Set flags based on current type of the finish
        match finish_via:
            case 'rex':
                manage_by_customized_rex_flag = False
            case 'custom_rex':
                manage_by_customized_rex_flag = True

        # Get the latest versions of wanted packages on the tested hosts
        tested_hosts_packages_latest_version_dicts = {}
        for host in hosts_to_test:
            # Check the latests versions of wanted packages
            result = host.run(f'dnf list {" ".join(packages)}').stdout
            # Cropping dnf output so it shows only packages
            packages_list = [line for line in result.split('\n')[3:-1]]
            packages_latest_version_dict = {}
            # Create dict containing {'package_name': 'version', ...} for further checks
            for item in packages_list:
                parts = item.split()
                # Remove architecture part from package name
                package_name = parts[0].split('.')[0]
                version = parts[1]  # parts look like ['package_name', 'version', 'repo']
                packages_latest_version_dict[package_name] = version

            tested_hosts_packages_latest_version_dicts[host.hostname] = (
                packages_latest_version_dict  # {'example.com':{'package_name': 'version', ...}, ...}
            )

        used_action = package_management_action.split('_')[0]

        if used_action == 'install':
            # Checking if wanted packages are available so they can be installed in the next step
            for host in hosts_to_test:
                assert host.run(f'rpm -q {" ".join(packages_to_install)}').status == len(
                    packages_to_install
                ), 'Some of the packages are already installed!'

        elif used_action == 'upgrade':
            # Installing and downgrading packages so they can be upgraded in the next step
            for host in hosts_to_test:
                assert (
                    host.run(
                        f'dnf list available | grep -E "{"|".join(packages_to_upgrade)}"'
                    ).status
                    == 0
                ), 'Wanted packages are not available!'
                assert host.run(f'dnf install {" ".join(packages_to_upgrade)} -y').status == 0, (
                    'Could not install wanted packages!'
                )
                assert host.run(f'dnf downgrade {" ".join(packages_to_upgrade)} -y').status == 0, (
                    'Packages were not downgraded!'
                )

        elif used_action == 'remove':
            # Installing packages so they can be removed in the next step
            for host in hosts_to_test:
                assert host.run(f'dnf install {" ".join(packages_to_remove)} -y').status == 0, (
                    'Could not install wanted packages!'
                )

        # Run airgun entity which performs Package management action based on the flags set above
        session.all_hosts.manage_packages(
            host_names=[host.hostname for host in hosts_to_test],
            upgrade_all_packages=upgrade_all_packages_flag,
            upgrade_packages=upgrade_packages_flag,
            install_packages=install_packages_flag,
            remove_packages=remove_packages_flag,
            packages_to_upgrade=packages_to_upgrade,
            packages_to_install=packages_to_install,
            packages_to_remove=packages_to_remove,
            manage_by_customized_rex=manage_by_customized_rex_flag,
        )

        # Wait till the job launched by the management action is finished
        job_id = int(module_target_sat.cli.JobInvocation().list()[0]['id'])
        if install_packages_flag:
            module_target_sat.wait_for_tasks(
                search_query=(
                    f'action: "Install package(s) name ^ ({",".join(packages_to_install)})" and resource_id = {job_id}'
                ),
            )

        elif upgrade_packages_flag and (not upgrade_all_packages_flag):
            module_target_sat.wait_for_tasks(
                f'action:  "Update package(s) name ^ ({",".join(packages_to_upgrade)})" and resource_id = {job_id}'
            )

        elif upgrade_all_packages_flag:
            if not manage_by_customized_rex_flag:
                module_target_sat.wait_for_tasks(
                    f'action: "Upgrade all packages" and resource_id = {job_id}'
                )
            else:
                module_target_sat.wait_for_tasks(
                    f'action: "Update package(s)" and resource_id = {job_id}'
                )

        elif remove_packages_flag:
            module_target_sat.wait_for_tasks(
                f'action: "Remove packages name ^ ({",".join(packages_to_remove)})" and resource_id = {job_id}'
            )

        # MAKE ASSERTS AFTER INSTALLING PACKAGES
        if used_action == 'install':
            for host in hosts_to_test:
                # Check that all the wanted packages are installed
                assert host.run(f'rpm -q {" ".join(packages_to_install)}').status == 0, (
                    'Some of the wanted packages is not installed!'
                )

                # Get versions of installed packages
                installed_packages = host.run(
                    f'rpm -qa {" ".join(packages_to_install)}'
                ).stdout.split('\n')[:-1]
                installed_packages_version_dict = {}
                # Create dict containing {'installed_package_name': 'version', ...} for further checks
                for package in installed_packages:
                    package_name, version = package.split('-')[0], '-'.join(package.split('-')[1:])
                    installed_packages_version_dict[package_name] = version[
                        : version.rfind('.')
                    ].strip()

                # Check that the latest version of packages is installed
                for package in packages_to_install:
                    assert (
                        installed_packages_version_dict[package]
                        == tested_hosts_packages_latest_version_dicts[host.hostname][package]
                    ), f'Package "{package}" is not installed in the latest version!'

        # MAKE ASSERTS AFTER UPGRADING PACKAGES
        elif used_action == 'upgrade':
            for host in hosts_to_test:
                # Get versions of installed packages
                installed_packages = host.run(
                    f'rpm -qa {" ".join(packages_to_upgrade)}'
                ).stdout.split('\n')[:-1]
                installed_packages_version_dict = {}
                # Create dict containing {'installed_package_name': 'version', ...} for further checks
                for package in installed_packages:
                    package_name, version = package.split('-')[0], '-'.join(package.split('-')[1:])
                    installed_packages_version_dict[package_name] = version[
                        : version.rfind('.')
                    ].strip()

                # Check that the package was upgraded to the latest version
                for package in packages_to_upgrade:
                    assert (
                        installed_packages_version_dict[package]
                        == tested_hosts_packages_latest_version_dicts[host.hostname][package]
                    ), f'Package "{package}" is not upgraded to the latest version!'

        # MAKE ASSERTS AFTER REMOVING PACKAGES
        elif used_action == 'remove':
            for host in hosts_to_test:
                # Assert that all the wanted packages were removed
                assert host.run(f'rpm -qa {" ".join(packages_to_remove)}').stdout == ''

    @request.addfinalizer
    def _cleanup():
        for host in hosts_to_test:
            time.sleep(5)
            assert host.run(f'dnf remove {" ".join(packages)} -y').status == 0, (
                'Could not remove installed packages in a finalizer!'
            )


@pytest.mark.parametrize('errata_to_install', ['1', '2'])
@pytest.mark.parametrize('manage_by_custom_rex', [True, False])
@pytest.mark.parametrize(
    'function_repos_collection_with_manifest',
    [
        {
            'distro': 'rhel8',
            'YumRepository': [
                {'url': settings.repos.yum_3.url},
                {'url': settings.repos.yum_1.url},
            ],
        }
    ],
    indirect=True,
)
@pytest.mark.no_containers
@pytest.mark.rhel_ver_match('8')
def test_all_hosts_manage_errata(
    session,
    module_target_sat,
    function_sca_manifest_org,
    content_hosts,
    function_repos_collection_with_manifest,
    manage_by_custom_rex,
    errata_to_install,
):
    """Apply an errata on multiple hosts through bulk errata wizard in All Hosts page.

    :id: c14c87a3-fdeb-4ad2-af36-65aa17fc7d41

    :expectedresults: Errata can be bulk applied to hosts through the All Hosts page.

    :CaseComponent: Hosts

    :Team: Proton
    """
    if errata_to_install == '1':
        errata_ids = settings.repos.yum_3.errata[25]
    if errata_to_install == '2':
        errata_ids = [settings.repos.yum_3.errata[25], settings.repos.yum_1.errata[1]]
    for host in content_hosts:
        host.add_rex_key(module_target_sat)
        function_repos_collection_with_manifest.setup_virtual_machine(
            host, enable_custom_repos=True
        )
        host.run(f'yum install -y {FAKE_7_CUSTOM_PACKAGE}')
        result = host.run(f'rpm -q {FAKE_7_CUSTOM_PACKAGE}')
        assert result.status == 0
        if errata_to_install == '2':
            host.run(f'yum install -y {FAKE_1_CUSTOM_PACKAGE}')
            result = host.run(f'rpm -q {FAKE_1_CUSTOM_PACKAGE}')
            assert result.status == 0
    with module_target_sat.ui_session() as session:
        session.organization.select(function_sca_manifest_org.name)
        session.location.select(loc_name=DEFAULT_LOC)
        session.all_hosts.manage_errata(
            host_names=[content_hosts[0].hostname, content_hosts[1].hostname],
            erratas_to_apply_by_id=errata_ids,
            manage_by_customized_rex=manage_by_custom_rex,
        )
        if errata_to_install == '2':
            errata_ids = f'{errata_ids[0]},{errata_ids[1]}'
        for host in content_hosts:
            task_result = module_target_sat.wait_for_tasks(
                search_query=(f'"Install errata errata_id ^ ({errata_ids}) on {host.hostname}"'),
                search_rate=2,
                max_tries=60,
            )
            task_status = module_target_sat.api.ForemanTask(id=task_result[0].id).poll()
            assert task_status['result'] == 'success'


@pytest.mark.rhel_ver_match([settings.content_host.default_rhel_version])
def test_positive_manage_repository_sets(
    module_target_sat,
    module_sca_manifest_org,
    module_lce,
    module_ak,
    content_hosts,
):
    """
    Change one or more repository status on multiple hosts through Manage content wizard

    :id: 4c9913d8-ce0d-4b50-901a-024aca207fc5

    :expectedresults: Repository status can be changed via All Hosts page > Manage content wizard.

    :Team: Proton
    """
    rhel_ver = content_hosts[0].os_version.major
    repo_key = f'rhel{rhel_ver}_bos'
    all_repo_names = []
    host_names = []
    status_to_be_changed = {
        0: 'Override to disabled',
        1: 'Override to enabled',
        2: 'Reset to default',
    }

    # Create content view and enable the BOS repo for the host RHEL version
    content_view = module_target_sat.api.ContentView(organization=module_sca_manifest_org).create()
    rh_repo_id = module_target_sat.api_factory.enable_rhrepo_and_fetchid(
        basearch=DEFAULT_ARCHITECTURE,
        org_id=module_sca_manifest_org.id,
        product=REPOS[repo_key]['product'],
        repo=REPOS[repo_key]['name'],
        reposet=REPOS[repo_key]['reposet'],
        releasever=REPOS[repo_key]['releasever'],
    )
    module_target_sat.wait_for_tasks(
        search_query='Actions::Katello::Repository::MetadataGenerate',
        max_tries=5,
        search_rate=10,
    )
    rh_repo = module_target_sat.api.Repository(id=rh_repo_id).read()
    content_view.repository = [rh_repo]
    content_view = module_target_sat.api.ContentView(
        id=content_view.id, repository=content_view.repository
    ).update(['repository'])
    content_view.publish()
    content_view = content_view.read()
    cv_version = content_view.version[0]
    cv_version.promote(data={'environment_ids': module_lce.id})

    # Update activation key
    module_ak = module_target_sat.api.ActivationKey(
        id=module_ak.id,
        organization=module_sca_manifest_org,
        content_view=content_view,
        environment=module_lce,
    ).update()

    # Register hosts and collect repo names
    for host in content_hosts:
        assert (
            host.register(module_sca_manifest_org, None, module_ak.name, module_target_sat).status
            == 0
        )
        raw_output = host.execute('subscription-manager repos --list').stdout
        data_list = raw_output.split('\n')
        for line in data_list:
            if 'Repo Name' in line:
                repository = (line.split(':')[1]).lstrip()
                if repository not in all_repo_names:
                    all_repo_names.append(repository)
        if 'Enabled:   0' in raw_output:
            rep_status = host.execute(r'subscription-manager repos --enable \*').stdout
            assert 'enabled for this system' in rep_status
        host_names.append(host.hostname)

    # Change repository status to disabled on multiple hosts
    override_to_disabled = status_to_be_changed[0]
    with module_target_sat.ui_session() as session:
        session.organization.select(module_sca_manifest_org.name)
        session.all_hosts.manage_repository_sets(
            host_names=host_names,
            select_all_hosts=False,
            repository_names=all_repo_names,
            status_to_change=override_to_disabled,
        )
        for host in content_hosts:
            output = host.execute('subscription-manager repos --list').stdout
            assert 'Enabled:   0' in output, 'repository status not changed'

        # Change repository status to enabled
        override_to_enabled = status_to_be_changed[1]
        session.all_hosts.manage_repository_sets(
            host_names=host_names,
            select_all_hosts=False,
            repository_names=all_repo_names,
            status_to_change=override_to_enabled,
        )
        for host in content_hosts:
            output = host.execute('subscription-manager repos --list').stdout
            assert 'Enabled:   1' in output, 'repository status not changed'


def test_disassociate_multiple_hosts(
    request,
    target_sat,
    module_location,
    module_org,
    vmware,
    default_location,
):
    """
    Import multiple VMs from a VMware compute resource, disassociate them via the UI,
    and verify via API that their uuid and compute_resource_id are cleared.

    :id: e5af21c7-62ef-4cc7-a72a-ab6c26090b68

    :steps:
        1. Create all required entities (domain, subnet, hostgroup, etc.)
        2. Import 2 VMs from VMware into Satellite
        3. Disassociate the VMs via the All Hosts UI
        4. Verify via API that uuid and compute_resource_id are None

    :expectedresults: VMs are disassociated and their compute resource info is cleared.

    :CaseComponent: Hosts

    :Team: Proton
    """

    cr_name = gen_string('alpha')

    # create entities for hostgroup
    target_sat.api.SmartProxy(
        id=target_sat.nailgun_smart_proxy.id, location=[default_location.id, module_location.id]
    ).update()
    domain = target_sat.api.Domain(
        organization=[module_org.id], location=[module_location]
    ).create()
    subnet = target_sat.api.Subnet(
        organization=[module_org.id], location=[module_location], domain=[domain]
    ).create()
    architecture = target_sat.api.Architecture().create()
    ptable = target_sat.api.PartitionTable(
        organization=[module_org.id], location=[module_location]
    ).create()
    operatingsystem = target_sat.api.OperatingSystem(
        architecture=[architecture], ptable=[ptable]
    ).create()
    medium = target_sat.api.Media(
        organization=[module_org.id], location=[module_location], operatingsystem=[operatingsystem]
    ).create()
    lce = (
        target_sat.api.LifecycleEnvironment(name="Library", organization=module_org.id)
        .search()[0]
        .read()
        .id
    )
    cv = target_sat.api.ContentView(organization=module_org).create()
    cv.publish()

    # create hostgroup
    hostgroup_name = gen_string('alpha')
    target_sat.api.HostGroup(
        name=hostgroup_name,
        architecture=architecture,
        domain=domain,
        subnet=subnet,
        location=[module_location.id],
        medium=medium,
        operatingsystem=operatingsystem,
        organization=[module_org],
        ptable=ptable,
        lifecycle_environment=lce,
        content_view=cv,
        content_source=target_sat.nailgun_smart_proxy.id,
    ).create()

    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=module_location.name)
        session.computeresource.create(
            {
                'name': cr_name,
                'provider': FOREMAN_PROVIDERS['vmware'],
                'provider_content.vcenter': vmware.hostname,
                'provider_content.user': settings.vmware.username,
                'provider_content.password': settings.vmware.password,
                'provider_content.datacenter.value': settings.vmware.datacenter,
                'locations.resources.assigned': [module_location.name],
                'organizations.resources.assigned': [module_org.name],
            }
        )
        session.hostgroup.update(
            hostgroup_name, {'host_group.deploy': f'{cr_name} ({FOREMAN_PROVIDERS["vmware"]})'}
        )

        cr_vm_names = [settings.vmware.vm_name, 'proton-testing-guest-rhel-8']
        vm_names_with_domains = [f'{name.replace(".", "")}.{domain.name}' for name in cr_vm_names]

        # Import VMs from VMware compute resource
        for cr_vm_name, vm_name_with_domain in zip(
            cr_vm_names, vm_names_with_domains, strict=False
        ):
            session.computeresource.vm_import(
                cr_name,
                cr_vm_name,
                hostgroup_name,
                module_location.name,
                name=cr_vm_name.replace('.', ''),
            )
            assert session.all_hosts.search(vm_name_with_domain)

        @request.addfinalizer
        def _cleanup():
            for vm_name in vm_names_with_domains:
                try:
                    target_sat.api.Host().search(query={"search": f'name={vm_name}'})[0].delete()
                except APIResponseError as e:
                    print(f"Failed to delete VM {vm_name}: {e}")

        for vm_name in vm_names_with_domains:
            # Get info about host from API
            host = target_sat.api.Host().search(query={"search": f'name={vm_name}'})[0]
            # Check that uuid and compute_resource_id are set
            assert host.uuid is not None, f"UUID for {vm_name} is not set"
            assert host.compute_resource.id is not None, (
                f"Compute resource ID for {vm_name} is not set"
            )

        session.all_hosts.disassociate_hosts(host_names=vm_names_with_domains)

        for vm_name in vm_names_with_domains:
            # Get info about host from API
            host = target_sat.api.Host().search(query={"search": f'name={vm_name}'})[0]
            # Check that uuid and compute_resource_id are set to None
            assert host.uuid is None, f"UUID for {vm_name} is not None after disassociation"
            assert host.compute_resource is None, (
                f"Compute resource ID for {vm_name} is not None after disassociation"
            )


def test_positive_change_power_state(
    request,
    target_sat,
    module_location,
    module_org,
    vmware,
    default_location,
):
    """
    Import multiple VMs from a VMware compute resource, Power Off them via the UI,
    and verify Hosts are Power Off .

    :id: e23bcd69-4c57-4967-aeec-7b4fbf6eaff3

    :steps:
        1. Create all required entities (domain, subnet, hostgroup, etc.)
        2. Import 2 VMs from VMware into Satellite
        3. Power Off the VMs via the All Hosts UI
        4. Verify via UI that Hosts are Power off
        5. Powers them back on and verifies "On" state
        6. Stops them twice and verifies final "Off" state

    :expectedresults: VMs are Power off and Power icon should change to red.

    """

    cr_name = gen_string('alpha')

    # create entities for hostgroup
    target_sat.api.SmartProxy(
        id=target_sat.nailgun_smart_proxy.id, location=[default_location.id, module_location.id]
    ).update()
    domain = target_sat.api.Domain(
        organization=[module_org.id], location=[module_location]
    ).create()
    subnet = target_sat.api.Subnet(
        organization=[module_org.id], location=[module_location], domain=[domain]
    ).create()
    architecture = target_sat.api.Architecture().create()
    ptable = target_sat.api.PartitionTable(
        organization=[module_org.id], location=[module_location]
    ).create()
    operatingsystem = target_sat.api.OperatingSystem(
        architecture=[architecture], ptable=[ptable]
    ).create()
    medium = target_sat.api.Media(
        organization=[module_org.id], location=[module_location], operatingsystem=[operatingsystem]
    ).create()
    lce = target_sat.api.LifecycleEnvironment(id=module_org.library.id).read()
    cv = target_sat.api.ContentView(organization=module_org).create()
    cv.publish()

    # create hostgroup
    hostgroup_name = gen_string('alpha')
    target_sat.api.HostGroup(
        name=hostgroup_name,
        architecture=architecture,
        domain=domain,
        subnet=subnet,
        location=[module_location.id],
        medium=medium,
        operatingsystem=operatingsystem,
        organization=[module_org],
        ptable=ptable,
        lifecycle_environment=lce.id,
        content_view=cv,
        content_source=target_sat.nailgun_smart_proxy.id,
    ).create()
    cr_vm_names = [settings.vmware.vm_name, 'proton-testing-guest-rhel-8']
    vm_names_with_domains = [f'{name.replace(".", "")}.{domain.name}' for name in cr_vm_names]

    # Create compute resource via API
    compute_resource = target_sat.api.VMWareComputeResource(
        name=cr_name,
        provider='Vmware',
        url=vmware.hostname,
        user=settings.vmware.username,
        password=settings.vmware.password,
        datacenter=settings.vmware.datacenter,
        organization=[module_org],
        location=[module_location],
    ).create()

    @request.addfinalizer
    def _cleanup():
        # Clean up each host: power on, disassociate, and delete via API
        for vm_name in vm_names_with_domains:
            try:
                hosts = target_sat.api.Host().search(query={"search": f'name={vm_name}'})
                if hosts:
                    host = hosts[0]
                    # Power on (in case test failed during power-off state)
                    try:
                        host.power(data={'power_action': 'start'})
                    except Exception as e:
                        print(f"Failed to power on VM {vm_name}: {e}")

                    # Disassociate from compute resource
                    try:
                        host.disassociate()
                    except Exception as e:
                        print(f"Failed to disassociate VM {vm_name}: {e}")

                    # Delete host
                    try:
                        host.delete()
                    except APIResponseError as e:
                        print(f"Failed to delete VM {vm_name}: {e}")
            except Exception as e:
                print(f"Failed to find host {vm_name}: {e}")

        # Clean up compute resource
        try:
            compute_resource.delete()
        except Exception as e:
            print(f"Failed to delete compute resource: {e}")

    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=module_location.name)

        try:
            session.hostgroup.update(
                hostgroup_name, {'host_group.deploy': f'{cr_name} ({FOREMAN_PROVIDERS["vmware"]})'}
            )

            # Import VMs from VMware compute resource
            for cr_vm_name, _vm_name_with_domain in zip(
                cr_vm_names, vm_names_with_domains, strict=False
            ):
                session.computeresource.vm_import(
                    cr_name,
                    cr_vm_name,
                    hostgroup_name,
                    module_location.name,
                    name=cr_vm_name.replace('.', ''),
                )
            for vm_name in cr_vm_names:
                power_status = session.computeresource.vm_status(cr_name, vm_name)
                if not power_status:
                    session.computeresource.vm_poweron(cr_name, vm_name)

            # Ensure the 'Power' column is displayed in the All Hosts table
            headers = session.all_hosts.get_displayed_table_headers()
            if "Power" not in headers:
                session.all_hosts.manage_table_columns(
                    {
                        'Power': True,
                    }
                )
            session.all_hosts.change_power_state(state='Power Off', select_all_hosts=True)
            # Wait for the modal to close and table to refresh after power state change
            session.browser.plugin.ensure_page_safe(timeout='20s')
            # UI check
            for host in vm_names_with_domains:
                state = session.all_hosts.read_power_state_icon(host_name=host)
                assert state['state'] == 'Off', f"Host {host} state didn't change to Off"
            session.all_hosts.change_power_state(state='Start', select_all_hosts=True)
            # Wait for the modal to close and table to refresh after power state change
            session.browser.plugin.ensure_page_safe(timeout='20s')
            for host in vm_names_with_domains:
                state = session.all_hosts.read_power_state_icon(host_name=host)
                assert state['state'] == 'On', f"Host {host} state didn't change to On"
            # Stop sends a shutdown command to the server or vm and that can take anywhere
            # from a few minutes to 15-20 so if we send Stop twice then it shutdown immediately
            session.all_hosts.change_power_state(state='Stop', select_all_hosts=True)
            session.all_hosts.change_power_state(state='Stop', select_all_hosts=True)
            # Wait for the modal to close and table to refresh after power state change
            session.browser.plugin.ensure_page_safe(timeout='20s')
            for host in vm_names_with_domains:
                state = session.all_hosts.read_power_state_icon(host_name=host)
                assert state['state'] == 'Off', f"Host {host} state didn't change to Off"

        finally:
            # Reset Power column state before session closes
            session.all_hosts.manage_table_columns(
                {
                    'Power': False,
                }
            )


def assert_hosts_owner_helper(target_sat, session, hosts, expected_owner, owner_type='user'):
    """
    Assert that all hosts have the expected owner both via API and UI.

    :param target_sat: Satellite object for API access
    :param session: UI session object
    :param hosts: list of host objects
    :param expected_owner: expected owner login or group name
    :param owner_type: 'user' or 'usergroup'
    """
    # API check
    if owner_type == 'user':
        new_hosts_owner_logins = [
            target_sat.api.User(
                id=target_sat.api.Host().search(query={"search": f'name={host.name}'})[0].owner.id
            )
            .read()
            .login
            for host in hosts
        ]
        assert all(owner == expected_owner for owner in new_hosts_owner_logins), (
            f'Hosts owner was not changed to {expected_owner}'
        )
    elif owner_type == 'usergroup':
        user_group_id = (
            target_sat.api.UserGroup().search(query={'search': f'name={expected_owner}'})[0].id
        )
        new_hosts_owner_ids = [
            target_sat.api.Host().search(query={"search": f'name={host.name}'})[0].owner.id
            for host in hosts
        ]
        assert all(owner_id == user_group_id for owner_id in new_hosts_owner_ids), (
            f'Hosts owner was not changed to user group id {user_group_id}'
        )

    # UI check
    search_string = ' or '.join([f'name="{host.name}"' for host in hosts])
    read_hosts_vals = session.all_hosts.search(search_string)
    for host in read_hosts_vals:
        assert host['Owner'] == expected_owner, (
            f'Host {host["Name"]} owner was not changed to {expected_owner}'
        )


def test_positive_change_hosts_owner(module_org, module_location, target_sat):
    """
    This test changes the owner of multiple hosts using the new All Hosts UI bulk actions.

    :id: 51c9a368-8512-46ce-9c55-bd7d41ab2b9d

    :steps:
        1. Have 2 hosts
        2. Create new user
        3. Get the owner of the selected hosts
        4. Change the owner of the selected hosts to new owner
        5. Verify the owner of the selected hosts has been changed
        6. Create a new user group and add the new user to it
        7. Change the owner of the selected hosts to user group created in step 6
        8. Verify the owner of the selected hosts has been changed to user group

    :expectedresults: The owner of the selected hosts should be changed successfully.

    :CaseComponent: Hosts

    :Team: Proton
    """
    new_user_login = gen_string('alpha')
    new_user_password = gen_string('alpha')
    user_group_name = gen_string('alpha')

    # Create 2 hosts
    hosts = []
    for _ in range(2):
        host = target_sat.cli_factory.make_fake_host(
            {
                'organization': module_org.name,
                'location': module_location.name,
            }
        )
        hosts.append(host)

    new_user = target_sat.api.User(
        admin=False,
        location=[module_location],
        organization=[module_org],
        login=new_user_login,
        password=new_user_password,
    ).create()

    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)
        session.location.select(loc_name=module_location.name)

        # Change the hosts' owner to the new user
        session.all_hosts.change_hosts_owner(
            host_names=[host.name for host in hosts], new_owner_name=new_user_login
        )

        # Ensure the 'Owner' column is displayed in the All Hosts table
        headers = session.all_hosts.get_displayed_table_headers()
        if "Owner" not in headers:
            session.all_hosts.manage_table_columns(
                {
                    'Owner': True,
                }
            )

        assert_hosts_owner_helper(target_sat, session, hosts, new_user_login, owner_type='user')

        target_sat.api.UserGroup(name=user_group_name, user=[new_user.id]).create()
        session.browser.refresh()

        # Change the hosts' owner to the user group
        session.all_hosts.change_hosts_owner(
            host_names=[host.name for host in hosts], new_owner_name=user_group_name
        )

        assert_hosts_owner_helper(
            target_sat, session, hosts, user_group_name, owner_type='usergroup'
        )


def test_positive_change_hosts_org_loc(
    module_target_sat,
    module_org,
    module_location,
    request,
):
    """
    This test changes organization and location of multiple hosts via bulk action in All Hosts page.

    :id: 7d53c0ec-a392-4003-ba75-948340f19f37

    :steps:
        1. Create two organizations and two locations
        2. Create/Register multiple content hosts in the first organization/location
        3. Navigate to All Hosts page
        4. Select multiple hosts
        5. Use bulk action to change organization and location
        6. Verify hosts are now in the new organization/location

    :expectedresults:
        1. Multiple hosts can be selected in All Hosts page
        2. Bulk organization/location change operation succeeds
        3. Hosts are successfully moved to new organization/location
        4. Host properties reflect the new taxonomies
    """
    # Create second organization and location for the test
    new_org = module_target_sat.api.Organization().create()
    new_location = module_target_sat.api.Location(organization=[new_org]).create()

    @request.addfinalizer
    def cleanup():
        new_org.delete()
        new_location.delete()

    host_names = []
    for _ in range(2):
        host = module_target_sat.cli_factory.make_fake_host(
            {
                'organization': module_org.name,
                'location': module_location.name,
            }
        )
        host_names.append(host.name)

    # Verify hosts are initially in the first org/location
    for host_name in host_names:
        host_entity = module_target_sat.api.Host().search(query={'search': f'name={host_name}'})[0]
        assert host_entity.organization.id == module_org.id
        assert host_entity.location.id == module_location.id

    # Perform bulk organization/location change through UI
    with module_target_sat.ui_session() as session:
        # Select the first organization to see the hosts
        session.organization.select(module_org.name)
        session.location.select(module_location.name)
        # Scenario 1 - Change organization with option "Fix in mismatch"
        session.all_hosts.change_associations_organization(
            host_names=host_names,
            new_organization=new_org.name,
        )
        # Switch to second organization to verify the change
        session.organization.select(new_org.name)

        # Scenario 2 - Change location with option "Fix in mismatch"
        session.all_hosts.change_associations_location(
            host_names=host_names,
            new_location=new_location.name,
        )
        # Switch to second location to verify the change
        session.location.select(new_location.name)

        # Verify hosts appear in the new organization and new location
        for host_name in host_names:
            host_values = session.all_hosts.search(host_name)
            assert host_values is not None, f"Host {host_name} not found in new organization"

        # API verification - verify hosts are now in the second org/location
        for host_name in host_names:
            host_entity = module_target_sat.api.Host().search(
                query={'search': f'name={host_name}'}
            )[0]
            assert host_entity.organization.id == new_org.id, (
                f'Host {host_name} not moved to second organization'
            )
            assert host_entity.location.id == new_location.id, (
                f'Host {host_name} not moved to second location'
            )


def test_positive_bulk_change_location_select_all_respects_context(
    module_target_sat,
    request,
):
    """Verify bulk 'Change Location' via 'Select All' only affects hosts
    in the current location context, not all hosts across the Satellite.

    :id: 235fa529-0a35-41a3-98f2-73d2ba3de5ef

    :Verifies: SAT-44548

    :steps:
        1. Create an organization with three locations (X, Y, Z)
        2. Create three fake hosts in location X and two in location Y
        3. Navigate to All Hosts page with location X selected
        4. Select all hosts using the 'Select All' checkbox
        5. Use bulk action to change location to Z
        6. Verify hosts originally in location X are now in location Z
        7. Verify hosts in location Y remain unaffected

    :expectedresults:
        1. Bulk 'Change Location' with 'Select All' respects the current
           location context and only moves hosts from that location
        2. Hosts in other locations are not affected by the bulk action
    """
    org = module_target_sat.api.Organization().create()
    loc_x, loc_y, loc_z = [
        module_target_sat.api.Location(organization=[org]).create() for _ in range(3)
    ]

    hosts_x_names = []
    for _ in range(3):
        host = module_target_sat.cli_factory.make_fake_host(
            {'organization': org.name, 'location': loc_x.name}
        )
        hosts_x_names.append(host['name'])

    hosts_y_names = []
    for _ in range(2):
        host = module_target_sat.cli_factory.make_fake_host(
            {'organization': org.name, 'location': loc_y.name}
        )
        hosts_y_names.append(host['name'])

    @request.addfinalizer
    def cleanup():
        for host_name in hosts_x_names + hosts_y_names:
            hosts = module_target_sat.api.Host().search(query={'search': f'name={host_name}'})
            if hosts:
                hosts[0].delete()
        for loc in [loc_x, loc_y, loc_z]:
            loc.delete()
        org.delete()

    with module_target_sat.ui_session() as session:
        session.organization.select(org.name)
        session.location.select(loc_x.name)

        session.all_hosts.change_associations_location(
            select_all_hosts=True,
            new_location=loc_z.name,
        )

    for host_name in hosts_x_names:
        host_entity = module_target_sat.api.Host().search(query={'search': f'name={host_name}'})[0]
        assert host_entity.location.id == loc_z.id, (
            f'Host {host_name} was not moved from location X to Z!'
        )

    for host_name in hosts_y_names:
        host_entity = module_target_sat.api.Host().search(query={'search': f'name={host_name}'})[0]
        assert host_entity.location.id == loc_y.id, (
            f'Host {host_name} in location Y was incorrectly affected by bulk action!'
        )


def read_host_collections_hosts(target_sat, host_col_names, org):
    """
    Helper function to read hosts in a host collections via API.

    :param target_sat: Satellite object for API access
    :param host_col_names: list of host collection names or a single host collection name
    :param org: Organization object the host collections belong to

    :return: dict mapping host collection names to lists of host names in each collection
    """

    if not isinstance(host_col_names, list):
        host_col_names = [host_col_names]

    hosts_assigned_to_collections = {}
    for host_col in host_col_names:
        if host_collection := target_sat.api.HostCollection(organization=org).search(
            query={'search': f'name="{host_col}"'}
        ):
            hc = host_collection[0].read()
            hosts_assigned_to_collections[host_col] = [host.read().name for host in hc.host]
        else:
            hosts_assigned_to_collections[host_col] = []

    return hosts_assigned_to_collections


def test_positive_all_hosts_manage_host_collections(target_sat, function_org, function_location):
    """
    Manage host collections for multiple hosts through All Hosts UI bulk actions.

    :id: ef9d1521-7cf1-4d86-a5b4-197f34fcf907

    :steps:
        1. Create 2 fake hosts
        2. Create 2 host collections with unlimited host limits
        3. Create 1 host collection with host limit set to 1
        4. Add 1 host to the host collection with limit=1 (should succeed)
        5. Attempt to add a second host to the host collection with limit=1 (should fail)
        6. Add both hosts to the 2 unlimited host collections (should succeed)
        7. Remove 1 host from 1 host collection
        8. Remove 1 host from 2 host collections simultaneously
        9. Add hosts back to both collections
        10. Remove all hosts from all collections

    :expectedresults:
        1. Hosts can be successfully added to host collections via bulk action
        2. Host collection limits are enforced (limit=1 prevents adding second host)
        3. Multiple hosts can be added to multiple host collections in one operation
        4. Hosts can be removed from single or multiple host collections
        5. Success/failure messages are displayed appropriately
        6. Host collection membership is correctly updated after each operation
        7. API verification confirms the UI changes match actual host collection membership

    :Verifies: SAT-31027
    """

    host_names = []
    for _ in range(2):
        host = target_sat.cli_factory.make_fake_host(
            {
                'organization': function_org.name,
                'location': function_location.name,
            }
        )
        host_names.append(host.name)

    host_col_names = ['TestHostCol_1', 'TestHostCol_2']

    # create two host collections with unlimited hosts limit
    for col_name in host_col_names:
        target_sat.api.HostCollection(
            name=col_name,
            max_hosts=None,
            organization=function_org,
        ).create()

    # create one host collection with host_limit=1
    host_col_limit_1_name = 'TestHostCol_limit_1'
    target_sat.api.HostCollection(
        name=host_col_limit_1_name,
        max_hosts=1,
        organization=function_org,
    ).create()

    with target_sat.ui_session() as session:
        session.organization.select(org_name=function_org.name)
        session.location.select(loc_name=function_location.name)

        # Add 1 host to host_collection with limit=1 - should succeed
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names[0],
            host_collections_to_select=host_col_limit_1_name,
            option='Add',
        )
        assert 'Added' in result
        hosts_in_host_col = read_host_collections_hosts(
            target_sat, host_col_limit_1_name, function_org
        )
        assert hosts_in_host_col[host_col_limit_1_name] == [host_names[0]]

        # Add second host to host_collection with limit=1 - should fail
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names[1],
            host_collections_to_select=host_col_limit_1_name,
            option='Add',
        )
        assert 'Failed to add host to host collection.' in result
        hosts_in_host_col = read_host_collections_hosts(
            target_sat, host_col_limit_1_name, function_org
        )
        assert hosts_in_host_col[host_col_limit_1_name] == [host_names[0]]

        # Add 2 hosts to 2 host collections
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names,
            host_collections_to_select=host_col_names,
            option='Add',
        )
        assert 'Added' in result
        hosts_in_host_col = read_host_collections_hosts(target_sat, host_col_names, function_org)
        assert sorted(hosts_in_host_col[host_col_names[0]]) == sorted(host_names)
        assert sorted(hosts_in_host_col[host_col_names[1]]) == sorted(host_names)

        # Remove 1 host from 1 host collection
        # at this state there are 2 hosts in both 'TestHostCol_1' and 'TestHostCol_2'
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names[0],
            host_collections_to_select=host_col_names[0],
            option='Remove',
        )
        assert 'Removed' in result
        hosts_in_host_col = read_host_collections_hosts(target_sat, host_col_names[0], function_org)
        assert hosts_in_host_col[host_col_names[0]] == [host_names[1]]

        # Remove 1 host from 2 host collections
        # at this state there is 1 host in 'TestHostCol_1' and 2 hosts in 'TestHostCol_2'
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names[1],
            host_collections_to_select=host_col_names,
            option='Remove',
        )
        assert 'Removed' in result
        hosts_in_host_col = read_host_collections_hosts(target_sat, host_col_names, function_org)
        assert hosts_in_host_col[host_col_names[0]] == []
        assert hosts_in_host_col[host_col_names[1]] == [host_names[0]]

        # Add hosts back, so both host collections have 2 hosts
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names,
            host_collections_to_select=host_col_names,
            option='Add',
        )
        hosts_in_host_col = read_host_collections_hosts(target_sat, host_col_names, function_org)
        assert sorted(hosts_in_host_col[host_col_names[0]]) == sorted(host_names)
        assert sorted(hosts_in_host_col[host_col_names[1]]) == sorted(host_names)

        # Remove 2 hosts from 2 host collections
        result = session.all_hosts.change_associations_host_collections(
            host_names=host_names,
            host_collections_to_select=host_col_names,
            option='Remove',
        )
        hosts_in_host_col = read_host_collections_hosts(target_sat, host_col_names, function_org)
        assert hosts_in_host_col[host_col_names[0]] == []
        assert hosts_in_host_col[host_col_names[1]] == []


@pytest.mark.parametrize(
    'module_repos_collection_with_setup',
    [{'YumRepository': {'url': settings.repos.yum_3.url}}],
    ids=['yum_3'],
    indirect=True,
)
@pytest.mark.rhel_ver_match([settings.content_host.default_rhel_version])
def test_positive_all_hosts_check_status_icon(
    target_sat,
    content_hosts,
    rhel_contenthost,
    module_repos_collection_with_setup,
    module_org,
    module_ak_with_cv,
):
    """
    Verify that host status icons in All Hosts table correctly reflect the host's actual
    dynamically recalculated status for all three states: success, warning, and danger.

    :id: 42089feb-5e3a-4d72-be4f-809c6042abce

    :steps:
        1. Create/register three content hosts with different statuses:
           - Host with content and latest packages installed (success status)
           - Host with applicable security errata (danger status)
           - Host with no package profile reported (warning status)
        2. For the success host, register with content and install latest packages
        3. For the errata host, install and then downgrade a package to create applicable errata
        4. Navigate to All Hosts page
        5. Read the status icon for each host
        6. Verify the status icon matches the expected state

    :expectedresults:
        1. Host with security errata shows 'danger' status with
           'Errata: Security errata applicable' details
        2. Host without package profile shows 'warning' status with appropriate message
        3. Host with content and latest packages shows 'success' status
        4. Status icons accurately represent each host's actual dynamically recalculated state

    :Verifies: SAT-36330, SAT-42220
    """
    success_host = rhel_contenthost
    errata_host = content_hosts[0]
    no_profile_host = content_hosts[1]

    # Setup errata host: register, install package, downgrade to create applicable errata
    errata_host.add_rex_key(target_sat)
    module_repos_collection_with_setup.setup_virtual_machine(errata_host, enable_custom_repos=True)
    assert errata_host.execute(f'yum install -y {FAKE_8_CUSTOM_PACKAGE}').status == 0
    assert errata_host.execute(f'yum downgrade -y {FAKE_8_CUSTOM_PACKAGE_NAME}').status == 0

    # Register no-profile host without package setup
    assert (
        no_profile_host.register(module_org, None, module_ak_with_cv.name, target_sat).status == 0
    )

    # Setup success host: register with content, install latest packages, no applicable errata
    module_repos_collection_with_setup.setup_virtual_machine(success_host, enable_custom_repos=True)
    assert success_host.execute(f'yum install -y {FAKE_8_CUSTOM_PACKAGE}').status == 0

    with target_sat.ui_session() as session:
        session.organization.select(org_name=ANY_CONTEXT['org'])
        session.location.select(loc_name=ANY_CONTEXT['location'])

        result = session.all_hosts.read_host_status_icon(host_name=errata_host.hostname)
        assert result['status'] == 'danger'
        assert 'Errata: Security errata applicable' in result['status_details']

        result = session.all_hosts.read_host_status_icon(host_name=no_profile_host.hostname)
        assert result['status'] == 'warning'
        assert (
            'Errata: No installed packages and/or enabled repositories have been reported by subscription-manager.'
            in result['status_details']
        )

        result = session.all_hosts.read_host_status_icon(host_name=success_host.hostname)
        assert result['status'] == 'success', (
            f'Expected success but got {result["status"]} for {success_host.hostname}. '
            f'Details: {result["status_details"]}'
        )
        assert 'Build: Installed' in result['status_details']


@pytest.mark.parametrize(
    'require_reboot',
    [False, True],
    ids=['service_restart', 'service_reboot'],
)
@pytest.mark.rhel_ver_match([settings.content_host.default_rhel_version])
def test_positive_all_hosts_manage_traces(target_sat, module_org, tracer_hosts, require_reboot):
    """Test bulk management of traces with multiple hosts via All Hosts page.

    :id: a107e6da-3c44-46d8-9e94-f15044e041b6

    :setup:
        1. Two RHEL content hosts registered to Satellite
        2. Tracer installed on both hosts
        3. Mock service package installed on both hosts

    :steps:
        1. For reboot scenario, use mock_service_as_rebootable context manager
        2. Create traces by downgrading mock service package
        3. Verify traces are detected on both hosts with correct type
        4. Navigate to All Hosts page
        5. Select both hosts
        6. Use bulk action to manage traces
        7. Trigger restart/reboot via remote execution
        8. Verify traces are resolved

    :expectedresults:
        1. Traces are detected on both hosts after package downgrade
        2. Service restart scenario: trace appears as type='daemon'
        3. Reboot required scenario: trace appears as type='static'
        4. Bulk traces management modal displays traces from both hosts
        5. REX job successfully restarts services or triggers reboot on both hosts
        6. Traces are resolved on both hosts after restart/reboot

    :parametrized: yes

    :Verifies: SAT-35465
    """
    host_names = []
    mock_service = settings.repos['MOCK_SERVICE_RPM']

    # Create traces on both hosts by downgrading packages
    for host in tracer_hosts:
        host_names.append(host.hostname)

        if require_reboot:
            # Use context manager to make service require reboot
            with mock_service_as_rebootable(host, mock_service):
                result = host.execute(f'yum -y downgrade {mock_service}')
                assert result.status == 0, f'Failed to downgrade {mock_service} on {host.hostname}'

                # Upload traces while in context (service marked as reboot-required)
                host.execute('dnf katello-tracer-upload')
        else:
            # Normal service restart scenario
            result = host.execute(f'yum -y downgrade {mock_service}')
            assert result.status == 0, f'Failed to downgrade {mock_service} on {host.hostname}'
            host.execute('dnf katello-tracer-upload')

    # Verify traces are detected on both hosts
    host_traces = {}
    for host in tracer_hosts:
        host_info = target_sat.cli.Host.info({'name': host.hostname})
        traces = target_sat.cli.HostTraces.list({'host-id': host_info['id']})
        assert len(traces) > 0, f'No traces detected on {host.hostname}'

        # Store traces by hostname for verification
        host_traces[host.hostname] = {trace['application']: trace for trace in traces}

        # Verify mock service trace is present
        mock_trace = [t for t in traces if t['application'] == mock_service]
        assert len(mock_trace) > 0, (
            f'Mock service trace not found on {host.hostname}. Found traces: '
            f'{[t["application"] for t in traces]}'
        )

        # Verify trace type matches expected behavior
        expected_type = 'static' if require_reboot else 'daemon'
        assert mock_trace[0]['type'] == expected_type, (
            f'Expected type={expected_type} for {mock_service} on {host.hostname}, '
            f'got type={mock_trace[0]["type"]}'
        )

    timestamp = (datetime.now(UTC) - timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M')

    with target_sat.ui_session() as session:
        session.organization.select(org_name=module_org.name)

        # Use bulk action to manage traces on both hosts
        alert_message = session.all_hosts.manage_traces(
            host_names=host_names, traces_to_select=[mock_service]
        )
        assert 'Danger alert' not in alert_message, 'Manage traces action failed'

    # Wait for REX job to complete
    search_query = f'action = "Run hosts job: Restart Services" and started_at >= "{timestamp}"'
    target_sat.wait_for_tasks(
        search_query=search_query,
        search_rate=15,
        max_tries=10,
    )

    if require_reboot:
        # Reboot hosts to clear static traces
        for host in tracer_hosts:
            host.power_control(state='reboot')

        for host in tracer_hosts:
            host.wait_for_connection()

    # Verify all traces are resolved on both hosts after restart/reboot
    for host in tracer_hosts:
        host_info = target_sat.cli.Host.info({'name': host.hostname})
        traces = target_sat.cli.HostTraces.list({'host-id': host_info['id']})
        remaining_apps = {t['application'] for t in traces}
        assert mock_service not in remaining_apps, (
            f'Trace {mock_service} still present on {host.hostname} after {"reboot" if require_reboot else "restart"}. '
            f'Remaining traces: {remaining_apps}'
        )


def verify_system_purpose_via_api(
    target_sat, content_hosts, expected_role, expected_usage, expected_service_level
):
    """
    Helper function to verify system purpose attributes via API for multiple hosts.

    :param target_sat: Satellite object for API access
    :param content_hosts: List of content host objects to verify
    :param expected_role: Expected role value (or '' for unset, or None to skip check)
    :param expected_usage: Expected usage value (or '' for unset, or None to skip check)
    :param expected_service_level: Expected service level value (or '' for unset, or None to skip check)
    """
    for host in content_hosts:
        host_info = target_sat.api.Host().search(query={'search': f'name={host.hostname}'})[0]
        host_details = host_info.read_json()
        sp_attrs = host_details['subscription_facet_attributes']

        if expected_role is not None:
            if expected_role == '':
                assert sp_attrs['purpose_role'] == '', f"Role not unset for {host.hostname}"
            else:
                assert sp_attrs['purpose_role'] == expected_role, (
                    f"Role mismatch for {host.hostname}"
                )

        if expected_usage is not None:
            if expected_usage == '':
                assert sp_attrs['purpose_usage'] == '', f"Usage not unset for {host.hostname}"
            else:
                assert sp_attrs['purpose_usage'] == expected_usage, (
                    f"Usage mismatch for {host.hostname}"
                )

        if expected_service_level is not None:
            if expected_service_level == '':
                assert sp_attrs['service_level'] == '', (
                    f"Service level not unset for {host.hostname}"
                )
            else:
                assert sp_attrs['service_level'] == expected_service_level, (
                    f"Service level mismatch for {host.hostname}"
                )


def verify_system_purpose_via_ui(
    session, host_hostname, expected_role, expected_usage, expected_service_level
):
    """
    Helper function to verify system purpose attributes via UI for a single host.

    :param session: UI session object
    :param host_hostname: Hostname of the host to verify
    :param expected_role: Expected role value (or '' for unset, or None to skip check)
    :param expected_usage: Expected usage value (or '' for unset, or None to skip check)
    :param expected_service_level: Expected service level value (or '' for unset, or None to skip check)
    """
    ui_values = session.host_new.get_details(host_hostname, widget_names='overview.system_purpose')

    if expected_role is not None:
        assert ui_values['overview']['system_purpose']['details']['role'] == (
            expected_role if expected_role != '' else ''
        ), f"UI role mismatch for {host_hostname}"

    if expected_usage is not None:
        assert ui_values['overview']['system_purpose']['details']['usage_type'] == (
            expected_usage if expected_usage != '' else ''
        ), f"UI usage mismatch for {host_hostname}"

    if expected_service_level is not None:
        assert ui_values['overview']['system_purpose']['details']['sla'] == (
            expected_service_level if expected_service_level != '' else ''
        ), f"UI service level mismatch for {host_hostname}"


@pytest.mark.rhel_ver_match('N-0')
def test_positive_all_hosts_manage_system_purpose(
    module_target_sat,
    module_sca_manifest_org,
    module_lce,
    mod_content_hosts,
):
    """
    Manage system purpose attributes for multiple hosts through All Hosts UI bulk actions.

    :id: e4d9eb79-518f-4976-af9b-6de58dbffab6

    :steps:
        1. Register multiple content hosts
        2. Navigate to All Hosts page
        3. Select multiple hosts
        4. Use bulk action to set system purpose attributes (role, usage, service level, release version)
        5. Verify system purpose attributes have been updated on all selected hosts

    :expectedresults:
        1. Multiple hosts can be selected in All Hosts page
        2. Bulk system purpose modal opens with dropdowns for role, usage, service level, and release version
        3. System purpose attributes are successfully updated for all selected hosts
        4. Changes are reflected in the API and CLI verification
        5. Task completes successfully for each host

    :Verifies: SAT-35464
    """
    content_hosts = mod_content_hosts
    host_names = []

    content_view = module_target_sat.api.ContentView(organization=module_sca_manifest_org).create()
    content_view.publish()
    content_view = content_view.read()
    cvv = content_view.version[0].read()
    cvv.promote(data={'environment_ids': module_lce.id})
    ak = module_target_sat.api.ActivationKey(
        content_view=content_view,
        environment=module_lce,
        organization=module_sca_manifest_org,
    ).create()

    # Register content hosts
    for host in content_hosts:
        host.add_rex_key(module_target_sat)
        result = host.register(module_sca_manifest_org, None, ak.name, module_target_sat)
        assert result.status == 0, f'Failed to register host: {result.stderr}'
        host_names.append(host.hostname)

    # Define system purpose attributes to set
    syspurpose_attributes = {
        'role': 'Red Hat Enterprise Linux Server',
        'usage': 'Production',
        'service_level': 'Standard',
    }

    with module_target_sat.ui_session() as session:
        session.organization.select(module_sca_manifest_org.name)
        session.location.select(loc_name=DEFAULT_LOC)

        # Apply system purpose attributes via bulk action
        session.all_hosts.manage_system_purpose(
            host_names=host_names,
            role=syspurpose_attributes['role'],
            usage=syspurpose_attributes['usage'],
            service_level=syspurpose_attributes['service_level'],
        )

        def wait_for_system_purpose_update_tasks():
            # Wait for tasks to complete
            for host in content_hosts:
                task_result = module_target_sat.wait_for_tasks(
                    search_query=(f'Updating System Purpose for host {host.hostname}'),
                    search_rate=2,
                    max_tries=60,
                )
                assert len(task_result) > 0, f'No task found for {host.hostname}'
                task_status = module_target_sat.api.ForemanTask(id=task_result[0].id).poll()
                assert task_status['result'] == 'success', f'Task failed for {host.hostname}'

        wait_for_system_purpose_update_tasks()

        # Verify system purpose attributes were set correctly
        verify_system_purpose_via_api(
            module_target_sat,
            content_hosts,
            syspurpose_attributes['role'],
            syspurpose_attributes['usage'],
            syspurpose_attributes['service_level'],
        )
        verify_system_purpose_via_ui(
            session,
            content_hosts[0].hostname,
            syspurpose_attributes['role'],
            syspurpose_attributes['usage'],
            syspurpose_attributes['service_level'],
        )

        # Test updating system purpose with different values
        updated_syspurpose_attributes = {
            'role': 'Red Hat Enterprise Linux Workstation',
            'usage': 'Development/Test',
            'service_level': 'Premium',
        }

        session.all_hosts.manage_system_purpose(
            host_names=host_names,
            role=updated_syspurpose_attributes['role'],
            usage=updated_syspurpose_attributes['usage'],
            service_level=updated_syspurpose_attributes['service_level'],
        )
        wait_for_system_purpose_update_tasks()

        # Verify updated system purpose attributes
        verify_system_purpose_via_api(
            module_target_sat,
            content_hosts,
            updated_syspurpose_attributes['role'],
            updated_syspurpose_attributes['usage'],
            updated_syspurpose_attributes['service_level'],
        )
        verify_system_purpose_via_ui(
            session,
            content_hosts[0].hostname,
            updated_syspurpose_attributes['role'],
            updated_syspurpose_attributes['usage'],
            updated_syspurpose_attributes['service_level'],
        )

        # Test unsetting system purpose attributes using '(unset)' option
        session.all_hosts.manage_system_purpose(
            host_names=host_names,
            role='(unset)',
            usage='(unset)',
            service_level='(unset)',
        )
        wait_for_system_purpose_update_tasks()

        # Verify system purpose attributes were unset
        verify_system_purpose_via_api(
            module_target_sat,
            content_hosts,
            expected_role='',
            expected_usage='',
            expected_service_level='',
        )
        verify_system_purpose_via_ui(
            session,
            content_hosts[0].hostname,
            expected_role='',
            expected_usage='',
            expected_service_level='',
        )


@pytest.mark.rhel_ver_match('N-1')
@pytest.mark.parametrize(
    'module_repos_collection_with_setup',
    [{'YumRepository': {'url': settings.repos.yum_3.url}}],
    ids=['yum_3'],
    indirect=True,
)
@pytest.mark.parametrize(
    ('cloud_provider', 'facts_attr', 'field_mappings', 'skip_null'),
    [
        ('gce', 'gce.cloud_billing_facts', {'gcp_license_codes': 'gcp_license_code'}, False),
        ('aws', 'ec2.cloud_billing_facts', {'aws_billing_products': 'aws_billing_product'}, True),
        ('azure', 'azurerm.cloud_billing_facts', {}, False),
    ],
    ids=['gce', 'aws', 'azure'],
)
def test_cloud_billing_details(
    target_sat,
    module_org,
    rhel_contenthost,
    module_repos_collection_with_setup,
    default_location,
    cloud_provider,
    facts_attr,
    field_mappings,
    skip_null,
):
    """Verify cloud billing details are displayed correctly in Host Details page.

    :id: 88d8eaac-9149-4cbc-8913-4954f3f017f8

    :steps:
        1. Register Content Host
        2. Create cloud provider facts file on registered host
        3. Upload facts to Satellite
        4. Navigate to Host Details page
        5. Verify Cloud Billing Details card is displayed
        6. Verify cloud provider-specific billing facts are shown

    :expectedresults: Cloud billing details are correctly displayed in UI

    :parametrized: yes

    :Verifies: SAT-39185
    """
    client = rhel_contenthost
    client.add_rex_key(target_sat)

    module_repos_collection_with_setup.setup_virtual_machine(client)

    with target_sat.ui_session() as session:
        session.organization.select(module_org.name)
        session.location.select(default_location.name)

        # Get facts from settings dynamically based on cloud provider
        facts_path = facts_attr.split('.')
        cloud_facts = settings
        for attr in facts_path:
            cloud_facts = getattr(cloud_facts, attr)

        # Create facts file on client host with cloud billing details
        facts_content = json.dumps(cloud_facts, indent=2)

        # Ensure facts directory exists
        client.execute('mkdir -p /etc/rhsm/facts')

        # Write cloud facts to host-billing.facts
        client.execute(f"cat > /etc/rhsm/facts/host-billing.facts << 'EOF'\n{facts_content}\nEOF")
        # Note: In above step we are using a workaround by uploading facts to contenthost
        # As in Cloud setup, cloud host can't send facts to Satellite due to different network(VPN vs Public Cloud)
        # Verify file created correctly
        result = client.execute('cat /etc/rhsm/facts/host-billing.facts')
        assert result.status == 0, 'Failed to read facts file'

        # Upload facts to Satellite
        client.execute('subscription-manager facts --update')

        # Get host details including cloud billing information
        host_details = session.host_new.get_details(
            client.hostname,
            widget_names='details',
        )

        cloud_billing = host_details['details']['cloud_billing_details']['details']

        # Validate all fields from cloud_facts
        for field, expected_value in cloud_facts.items():
            # Map config field name to UI field name
            ui_field = field_mappings.get(field, field)

            # Skip NULL AWS-only fields
            if skip_null and expected_value is None:
                assert ui_field not in cloud_billing, (
                    f'{field} (UI: {ui_field}) should not be displayed when value is null'
                )
                continue

            # Verify field exists
            assert ui_field in cloud_billing, (
                f'{cloud_provider.upper()}: {field} (UI: {ui_field}) not displayed in cloud billing details'
            )

            # Verify field value
            assert str(cloud_billing[ui_field]) == str(expected_value), (
                f'{cloud_provider.upper()}: {field} value mismatch. Expected: {expected_value}, Got: {cloud_billing[ui_field]}'
            )


@pytest.mark.rhel_ver_match(f'{settings.content_host.default_rhel_version}')
@pytest.mark.no_containers
def test_assign_multi_cv_from_host_page(
    module_target_sat, module_org, module_lce, module_cv_repo, rhel_contenthost
):
    """Ensure that multiple content views can be assigned via the host details page

    :id: 518e6229-d728-4908-b95d-28de5d9b6f65

    :steps:
        1. Create a content view and an activation key associated with the content view and the Library lifecycle environment.
        2. Register a host using the activation key.
        3. On the Overview tab of the details page for the host, select the 'Assign content view environments' action from the 'Content view environments' card dropdown.
        4. Assign a new content view environment with a non-Library lifecycle environment and the content view created previously.

    :expectedresults:
        1. The content view environments card on the host details page shows two content view environments.
        2. One of the content view environments is associated with the Library lifecycle environment.
        3. The other content view environment is associated with the non-Library lifecycle environment.

    :verifies: SAT-25846
    """
    ak = module_target_sat.api.ActivationKey(
        organization=module_org.id,
        content_view=module_cv_repo.id,
        environment=module_org.library.id,
    ).create()
    result = rhel_contenthost.register(module_org, None, ak.name, module_target_sat)
    assert result.status == 0

    with module_target_sat.ui_session() as session:
        session.organization.select(module_org.name)
        session.host_new.add_content_view_env(
            rhel_contenthost.hostname, module_cv_repo.name, module_lce.name
        )
        cv_envs = session.host_new.get_content_view_envs(rhel_contenthost.hostname)

    assert len(cv_envs) == 2

    library_cv_env = [env for env in cv_envs if env['lce'] == 'Library'][0]
    assert library_cv_env['content_view'] == module_cv_repo.name

    new_cv_env = [env for env in cv_envs if env['lce'] == module_lce.name][0]
    assert new_cv_env['content_view'] == module_cv_repo.name


@pytest.mark.rhel_ver_match(f'{settings.content_host.default_rhel_version}')
@pytest.mark.no_containers
def test_assign_different_cv_from_same_env(
    module_target_sat,
    module_org,
    module_lce,
    module_cv_repo,
    rhel_contenthost,
):
    """Ensure that a host can be switched to a different content view in the same lifecycle environment

    :id: 64ac7868-18a6-47cc-9a3d-e014bf3ac320

    :steps:
        1. Create a content view and an activation key associated with the content view and a new lifecycle environment.
        2. Register a host using the activation key.
        3. Create a second content view associated with the same lifecycle environment as the previous content view.
        4. On the Overview tab of the details page for the host, select the 'Assign content view environments' action from the 'Content view environments' card dropdown.
        5. Use the `Select content view` dropdown to select the content view created in step 3, then click the `Save` button.

    :expectedresults:
        The host's content view environment is updated to use the content view created in step 3.

    :verifies: SAT-25846
    """
    # Create activation key and register host
    ak = module_target_sat.api.ActivationKey(
        organization=module_org.id,
        content_view=module_cv_repo.id,
        environment=module_org.library.id,
    ).create()
    result = rhel_contenthost.register(module_org, None, ak.name, module_target_sat)
    assert result.status == 0

    # Sync a new repo, add it to a new CV, and promote the CV
    repo = module_target_sat.api.Repository(
        product=module_target_sat.api.Product(organization=module_org).create(),
        url=settings.repos.yum_0.url,
    ).create()
    repo.sync()
    cv = module_target_sat.api.ContentView(
        organization=module_org.id, name=gen_string('alpha')
    ).create()
    cv.repository = [repo]
    cv.update(['repository'])
    cv.publish()
    cv_version = cv.read().version[0]
    cv_version.promote(data={'environment_ids': module_lce.id})
    cv = cv.read()

    # In a UI session, use the CV environment modal to change the host's CV to the new CV
    with module_target_sat.ui_session() as session:
        session.organization.select(module_org.name)
        session.host_new.switch_associated_cv(rhel_contenthost.hostname, cv.name)

        cv_env = session.host_new.get_content_view_envs(rhel_contenthost.hostname)[0]
        assert cv_env['content_view'] == cv.name


def test_positive_only_single_library_option_in_create_form(target_sat):
    """Ensure that only 1 Library option is displayed in the Create Host form
    when location is set to "Any location"

    :id: 559f6324-dc17-4274-99f5-957ef0a2faf0

    :steps:
        1. Set location to "Any location"
        2. Read LCE dropdown options in the Create host form

    :expectedresults:
        1. Only 1 Library option is displayed

    :verifies: SAT-42710, SAT-32425

    :customerscenario: true
    """
    with target_sat.ui_session() as session:
        session.organization.select(org_name=DEFAULT_ORG)
        session.location.select(loc_name=ANY_CONTEXT['location'])
        create_form = session.host.get_create_form()
        create_form.host.lce.open_filter.click()
        # Check that 'Library' appears just once
        assert create_form.host.lce.filter_content.read().count('Library') == 1


def test_positive_search_by_report_origin_shows_all_hosts(target_sat):
    """Verify that filtering hosts by report origin displays all matching hosts,
    not just the first one when it has more reports than the per-page limit.

    Previously, searching for 'origin = Puppet' would cause the UI to show
    only the first host if that host had more reports than the per-page limit,
    because the backend query joined with reports producing duplicate rows.

    :id: 487c164c-5332-4312-84c2-8df6ac850911

    :steps:
        1. Create two hosts via API
        2. Insert 25 config reports (more than 20 per-page limit) with
           origin 'Puppet' for the first host via direct SQL
        3. Insert 2 config reports with origin 'Puppet' for the second host
        4. Navigate to All Hosts and search for 'origin = Puppet'

    :expectedresults: Both hosts appear in the search results

    :Verifies: SAT-41188

    :customerscenario: true
    """
    host1 = target_sat.api.Host().create()
    host2 = target_sat.api.Host().create()
    timestamp = '2025-05-16 16:16:16'
    report_count = 25

    target_sat.execute(
        'su - postgres -c "psql -d foreman -c \\"'
        f"INSERT INTO reports (host_id, origin, reported_at) "
        f"SELECT {host1.id}, 'Puppet', '{timestamp}'::timestamp + (i || ' seconds')::interval "
        f"FROM generate_series(1, {report_count}) AS i"
        '\\""'
    )
    target_sat.execute(
        'su - postgres -c "psql -d foreman -c \\"'
        f"INSERT INTO reports (host_id, origin, reported_at) "
        f"SELECT {host2.id}, 'Puppet', '{timestamp}'::timestamp + (i || ' seconds')::interval "
        f"FROM generate_series(1, 2) AS i"
        '\\""'
    )

    with target_sat.ui_session() as session:
        session.organization.select(org_name=ANY_CONTEXT['org'])
        session.location.select(loc_name=ANY_CONTEXT['location'])
        ui_hosts = session.all_hosts.search('origin = Puppet')
        ui_host_names = {row['Name'] for row in ui_hosts}
        assert host1.name in ui_host_names, (
            f'Host {host1.name} with {report_count} reports not found in UI results'
        )
        assert host2.name in ui_host_names, (
            f'Host {host2.name} not found in UI results; '
            f'only {ui_host_names} displayed (pagination bug)'
        )