Skip to content
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.

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

[6.14.z] notifications: new test for notification_recipients endpoint #14698

Merged
Changes from all commits
Commits
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
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())
Loading