Skip to content

Commit

Permalink
[6.14.z] notifications: new test for notification_recipients endpoint (
Browse files Browse the repository at this point in the history
…#14698)

notifications: new test for notification_recipients endpoint (#14592)

This new test verifies that endpoint `/notification_recipients` works
and returns correct data structure.

It covers bug https://bugzilla.redhat.com/2249970
where this broken endpoint caused some web UI pages to fail to load.
  • Loading branch information
pnovotny authored Apr 9, 2024
1 parent 2ae28f7 commit 7170370
Showing 1 changed file with 57 additions and 21 deletions.
78 changes: 57 additions & 21 deletions tests/foreman/api/test_notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@

from robottelo.config import settings
from robottelo.constants import DEFAULT_LOC, DEFAULT_ORG
from robottelo.utils.issue_handlers import is_open


@pytest.fixture
Expand Down Expand Up @@ -54,8 +53,8 @@ def reschedule_long_running_tasks_notification(target_sat):

assert (
target_sat.execute(
f"FOREMAN_TASKS_CHECK_LONG_RUNNING_TASKS_CRONLINE='{every_minute_cron_schedule}' "
"foreman-rake foreman_tasks:reschedule_long_running_tasks_checker"
"foreman-rake foreman_tasks:reschedule_long_running_tasks_checker "
f"FOREMAN_TASKS_CHECK_LONG_RUNNING_TASKS_CRONLINE='{every_minute_cron_schedule}'"
).status
== 0
)
Expand All @@ -64,14 +63,14 @@ def reschedule_long_running_tasks_notification(target_sat):

assert (
target_sat.execute(
f"FOREMAN_TASKS_CHECK_LONG_RUNNING_TASKS_CRONLINE='{default_cron_schedule}' "
"foreman-rake foreman_tasks:reschedule_long_running_tasks_checker"
"foreman-rake foreman_tasks:reschedule_long_running_tasks_checker "
f"FOREMAN_TASKS_CHECK_LONG_RUNNING_TASKS_CRONLINE='{default_cron_schedule}'"
).status
== 0
)


@pytest.fixture
@pytest.fixture(autouse=True)
def start_postfix_service(target_sat):
"""Start postfix service (disabled by default)."""
assert target_sat.execute('systemctl start postfix').status == 0
Expand All @@ -92,33 +91,42 @@ def clean_root_mailbox(target_sat):
target_sat.execute(f'mv -f {root_mailbox_backup} {root_mailbox}')


@pytest.fixture
def wait_for_long_running_task_mail(target_sat, clean_root_mailbox, long_running_task):
"""Wait until the long-running task ID is found in the Satellite's mbox file."""
timeout = 300
def wait_for_mail(sat_obj, mailbox_file, contains_string, timeout=300, delay=5):
"""
Wait until the desired string is found in the Satellite's mbox file.
"""
try:
wait_for(
func=target_sat.execute,
func_args=[f'grep --quiet {long_running_task["task"]["id"]} {clean_root_mailbox}'],
fail_condition=lambda res: res.status == 0,
func=sat_obj.execute,
func_args=[f"grep --quiet '{contains_string}' {mailbox_file}"],
fail_condition=lambda res: res.status != 0,
timeout=timeout,
delay=5,
delay=delay,
)
except TimedOutError as err:
raise AssertionError(
f'No notification e-mail with long-running task ID {long_running_task["task"]["id"]} '
f'has arrived to {clean_root_mailbox} after {timeout} seconds.'
f'No e-mail with text "{contains_string}" has arrived to mailbox {mailbox_file} '
f'after {timeout} seconds.'
) from err
return True


@pytest.fixture
def root_mailbox_copy(target_sat, clean_root_mailbox, wait_for_long_running_task_mail):
def wait_for_long_running_task_mail(target_sat, clean_root_mailbox, long_running_task):
"""Wait until the long-running task ID is found in the Satellite's mbox file."""
return wait_for_mail(
sat_obj=target_sat,
mailbox_file=clean_root_mailbox,
contains_string=long_running_task["task"]["id"],
)


@pytest.fixture
def root_mailbox_copy(target_sat, clean_root_mailbox):
"""Parsed local system copy of the Satellite's root user mailbox.
:returns: :class:`mailbox.mbox` instance
"""
assert wait_for_long_running_task_mail
result = target_sat.execute(f'cat {clean_root_mailbox}')
assert result.status == 0, f'Could not read mailbox {clean_root_mailbox} on Satellite host.'
mbox_content = result.stdout
Expand Down Expand Up @@ -173,7 +181,7 @@ def long_running_task(target_sat):
@pytest.mark.usefixtures(
'admin_user_with_localhost_email',
'reschedule_long_running_tasks_notification',
'start_postfix_service',
'wait_for_long_running_task_mail',
)
def test_positive_notification_for_long_running_tasks(long_running_task, root_mailbox_copy):
"""Check that a long-running task (i.e., running or paused for more than two days)
Expand Down Expand Up @@ -218,5 +226,33 @@ def test_positive_notification_for_long_running_tasks(long_running_task, root_ma
'/foreman_tasks/tasks?search=state+%5E+%28running%2C+paused'
'%29+AND+state_updated_at' in body_text
), 'Link for long-running tasks is missing in the e-mail body.'
if not is_open('BZ:2223996'):
assert findall(r'_\("[\w\s]*"\)', body_text), 'Untranslated strings found.'
assert not findall(r'_\("[\w\s]*"\)', body_text), 'Untranslated strings found.'


@pytest.mark.tier1
def test_positive_notification_recipients(target_sat):
"""Check that endpoint `/notification_recipients` works and returns correct data structure.
:id: 10e0fac2-f11f-11ee-ba60-000c2989e153
:steps:
1. Do a GET request to /notification_recipients endpoint.
2. Check the returned data structure for expected keys.
:BZ: 2249970
:customerscenario: true
"""
notification_keys = [
'id',
'seen',
'level',
'text',
'created_at',
'group',
'actions',
]

recipients = target_sat.api.NotificationRecipients().read()
for notification in recipients.notifications:
assert set(notification_keys) == set(notification.keys())

0 comments on commit 7170370

Please sign in to comment.