Skip to content

improved rex UI tests reliability #21742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pondrejk merged 1 commit into from
Jul 27, 2026
Merged

improved rex UI tests reliability #21742

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
151 changes: 91 additions & 60 deletions tests/foreman/ui/test_remoteexecution.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
from wait_for import wait_for

from robottelo.config import settings
from robottelo.constants import ANY_CONTEXT
from robottelo.constants import ANY_CONTEXT, DEFAULT_LOC
from robottelo.utils.datafactory import (
gen_string,
valid_hostgroups_list_short,
Expand Down Expand Up @@ -98,7 +98,6 @@ def test_positive_hostgroups_full_nested_names(

@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
def test_positive_run_default_job_template(
session,
target_sat,
rex_contenthost,
module_org,
Expand Down Expand Up @@ -129,7 +128,15 @@ def test_positive_run_default_job_template(

with target_sat.ui_session() as session:
session.organization.select(module_org.name)
assert session.host.search(hostname)[0]['Name'] == hostname
session.location.select(loc_name=DEFAULT_LOC)
# Search with retry to handle potential context/indexing delays
result = wait_for(
lambda: session.all_hosts.search(f"name = {hostname}"),
timeout=30,
delay=2,
fail_condition=[],
).out
assert result[0]['Name'] == hostname
Comment thread
sourcery-ai[bot] marked this conversation as resolved.
Comment thread
pnovotny marked this conversation as resolved.
command = 'ls'
session.jobinvocation.run(
{
Expand All @@ -153,7 +160,7 @@ def test_positive_run_default_job_template(


@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
def test_rex_through_host_details(session, target_sat, rex_contenthost, module_org):
def test_rex_through_host_details(target_sat, rex_contenthost, module_org):
"""Run remote execution using the new host details page

:id: ee625595-4995-43b2-9e6d-633c9b33ff93
Expand Down Expand Up @@ -185,7 +192,17 @@ def test_rex_through_host_details(session, target_sat, rex_contenthost, module_o
)
task_status = target_sat.api.ForemanTask(id=task_result[0].id).poll()
assert task_status['result'] == 'success'
recent_jobs = session.host_new.get_details(hostname, "overview.recent_jobs")['overview']

# Wait for recent jobs table to load with job data
recent_jobs = wait_for(
lambda: session.host_new.get_details(hostname, "overview.recent_jobs")['overview'],
timeout=30,
delay=2,
fail_condition=lambda result: (
not result.get('recent_jobs', {}).get('finished', {}).get('table')
),
).out

assert recent_jobs['recent_jobs']['finished']['table'][0]['column0'] == "Run ls"
assert recent_jobs['recent_jobs']['finished']['table'][0]['column2'] == "succeeded"

Expand All @@ -195,7 +212,7 @@ def test_rex_through_host_details(session, target_sat, rex_contenthost, module_o
'ui_user', [{'admin': True}, {'admin': False}], indirect=True, ids=['adminuser', 'nonadminuser']
)
def test_positive_run_custom_job_template(
session, module_org, default_location, target_sat, ui_user, rex_contenthost
module_org, default_location, target_sat, ui_user, rex_contenthost
):
"""Run a job template on a host

Expand Down Expand Up @@ -226,7 +243,15 @@ def test_positive_run_custom_job_template(
job_template_name = gen_string('alpha')
with target_sat.ui_session() as session:
session.organization.select(module_org.name)
assert session.host.search(hostname)[0]['Name'] == hostname
session.location.select(default_location.name)
# Search with retry to handle potential context/indexing delays
result = wait_for(
lambda: session.all_hosts.search(f"name = {hostname}"),
timeout=30,
delay=2,
fail_condition=[],
).out
assert result[0]['Name'] == hostname
session.jobtemplate.create(
{
'template.name': job_template_name,
Expand Down Expand Up @@ -255,9 +280,7 @@ def test_positive_run_custom_job_template(

@pytest.mark.upgrade
@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
def test_positive_run_job_template_multiple_hosts(
session, module_org, target_sat, rex_contenthosts
):
def test_positive_run_job_template_multiple_hosts(module_org, target_sat, rex_contenthosts):
"""Run a job template against multiple hosts

:id: c4439ec0-bb80-47f6-bc31-fa7193bfbeeb
Expand All @@ -268,7 +291,7 @@ def test_positive_run_job_template_multiple_hosts(

1. Set remote_execution_connect_by_ip on hosts to true
2. Navigate to the hosts page and select at least two hosts
3. Click the "Select Action"
3. Click the "Schedule a job"
4. Select the job and appropriate template
5. Run the job

Expand All @@ -282,9 +305,15 @@ def test_positive_run_job_template_multiple_hosts(
with target_sat.ui_session() as session:
session.organization.select(module_org.name)
for host in host_names:
assert session.host.search(host)[0]['Name'] == host
session.host.reset_search()
job_status = session.host.schedule_remote_job(
# Search with retry to handle potential context/indexing delays
result = wait_for(
lambda h=host: session.all_hosts.search(f"name = {h}"),
timeout=30,
delay=2,
fail_condition=[],
).out
assert result[0]['Name'] == host
job_status = session.host_new.schedule_remote_job(
host_names,
{
'category_and_template.job_category': 'Commands',
Expand All @@ -298,7 +327,7 @@ def test_positive_run_job_template_multiple_hosts(


@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
def test_positive_run_scheduled_job_template_by_ip(session, module_org, rex_contenthost):
def test_positive_run_scheduled_job_template_by_ip(module_org, target_sat, rex_contenthost):
"""Schedule a job to be ran against a host by ip

:id: 4387bed9-969d-45fb-80c2-b0905bb7f1bd
Expand All @@ -323,13 +352,20 @@ def test_positive_run_scheduled_job_template_by_ip(session, module_org, rex_cont
"""
job_time = 6 * 60
hostname = rex_contenthost.hostname
with session:
with target_sat.ui_session() as session:
session.organization.select(module_org.name)
session.location.select('Default Location')
assert session.host.search(hostname)[0]['Name'] == hostname
session.location.select(loc_name=DEFAULT_LOC)
# Search with retry to handle potential context/indexing delays
result = wait_for(
lambda: session.all_hosts.search(f"name = {hostname}"),
timeout=30,
delay=2,
fail_condition=[],
).out
assert result[0]['Name'] == hostname
plan_time = session.browser.get_client_datetime() + datetime.timedelta(seconds=job_time)
command_to_run = 'sleep 10'
job_status = session.host.schedule_remote_job(
job_status = session.host_new.schedule_remote_job(
[hostname],
{
'category_and_template.job_category': 'Commands',
Expand All @@ -341,52 +377,47 @@ def test_positive_run_scheduled_job_template_by_ip(session, module_org, rex_cont
},
wait_for_results=False,
)
# Note that to create this host scheduled job we spent some time from that plan_time, as it
# was calculated before creating the job
# Track state transitions through polling
job_left_time = (plan_time - session.browser.get_client_datetime()).total_seconds()
# assert that we have time left to wait, otherwise we have to use more job time,
# the job_time must be significantly greater than job creation time.
assert job_left_time > 0
assert job_status['hosts'][0]['Name'] == hostname
assert job_status['hosts'][0]['Status'] in ('Awaiting start', 'N/A')
# sleep 3/4 of the left time
time.sleep(job_left_time * 3 / 4)
job_status = session.jobinvocation.read(f'Run {command_to_run}', hostname, 'hosts')
assert job_left_time > 0, "Job scheduled time already passed during creation"

assert job_status['hosts'][0]['Name'] == hostname
assert job_status['hosts'][0]['Status'] in (
'Awaiting start',
'N/A',
'Succeeded',
)
# recalculate the job left time to be more accurate
job_left_time = (plan_time - session.browser.get_client_datetime()).total_seconds()
# the last read time should not take more than 1/4 of the last left time
assert job_left_time > 0
wait_for(
lambda: (
session.jobinvocation.read(f'Run {command_to_run}', hostname, 'hosts')['hosts'][0][
'Status'
]
== 'Pending'
),
timeout=(job_left_time + 30),
delay=1,
)
# wait the job to change status to "Succeeded"
wait_for(
lambda: (
session.jobinvocation.read(f'Run {command_to_run}', hostname, 'hosts')['hosts'][0][
'Status'
]
== 'Succeeded'
),
timeout=30,
delay=1,
)
assert job_status['hosts'][0]['Status'] in ('Awaiting start', 'Scheduled')

# Poll for state transitions and record them
observed_states = [job_status['hosts'][0]['Status']]
poll_timeout = job_left_time + 60 # Extra buffer for job execution

@pondrejk pondrejk Jul 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd drop the comment

poll_start = time.time()

while time.time() - poll_start < poll_timeout:
job_status = session.jobinvocation.read(f'Run {command_to_run}', hostname, 'hosts')
current_status = job_status['hosts'][0]['Status']

# Record state changes

@pondrejk pondrejk Jul 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd drop these comments in while body, what it does is already summarized in comment on line 393

if current_status != observed_states[-1]:
observed_states.append(current_status)

# Exit when job completes
if current_status in ('Succeeded', 'Failed', 'Cancelled'):
break

time.sleep(2)

# Read final complete status
job_status = session.jobinvocation.read(f'Run {command_to_run}', hostname)
final_status = job_status['hosts'][0]['Status']
if final_status != observed_states[-1]:
observed_states.append(final_status)

# Verify state progression
# Expected: Awaiting start/Scheduled -> Running/Pending -> Succeeded
assert 'Succeeded' in observed_states, f"Job never succeeded. States: {observed_states}"
Comment thread
pondrejk marked this conversation as resolved.

has_pre_run_state = any(s in ('Awaiting start', 'Scheduled') for s in observed_states[:-1])
assert has_pre_run_state, f"Missing pre-run state. States: {observed_states}"

assert job_status['overall_status']['is_success']
assert job_status['hosts'][0]['Name'] == hostname
assert job_status['hosts'][0]['Status'] == 'Succeeded'


@pytest.mark.rhel_ver_list([settings.content_host.default_rhel_version])
Expand Down
Loading