Skip to content

Commit

Permalink
Notifications: new test for failed repo sync
Browse files Browse the repository at this point in the history
This patch adds new API test for mail notification
for a failed repository sync.

Other changes:

- Waiting for desired e-mail has been refactored to `wait_for_mail` function
  be more reusable for other and future tests.
  Also `wait_for:fail_condition` lambda was fixed due to confusion
  how it actually works.

- Fixed one assert in `test_positive_notification_for_long_running_tasks` test.
  • Loading branch information
pnovotny committed Nov 29, 2023
1 parent 7d1cdf6 commit 662de6a
Show file tree
Hide file tree
Showing 2 changed files with 155 additions and 18 deletions.
24 changes: 24 additions & 0 deletions robottelo/cli/user.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,3 +105,27 @@ def access_token(cls, action=None, options=None):
"""
cls.command_sub = f'access-token {action}'
return cls.execute(cls._construct_command(options), output_format='csv')

@classmethod
def mail_notification_add(cls, options=None):
"""
Usage:
hammer user mail-notification add [OPTIONS]
Options:
--interval VALUE Mail notification interval option, e.g. Daily,
Weekly or Monthly.
Required for summary notification
--location[-id|-title] VALUE/NUMBER Set the current location context for the request
--mail-notification[-id] VALUE/NUMBER
--mail-query VALUE Relevant only for audit summary notification
--organization[-id|-title] VALUE/NUMBER Set the current organization context
for the request
--subscription VALUE Mail notification subscription option, e.g.
Subscribe, Subscribe to my hosts or
Subscribe to all hosts. Required for host built
and config error state
--user[-id] VALUE
"""
cls.command_sub = 'mail-notification add'
return cls.execute(cls._construct_command(options), output_format='csv')
149 changes: 131 additions & 18 deletions tests/foreman/api/test_notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,7 @@
from wait_for import TimedOutError, wait_for

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


@pytest.fixture
Expand All @@ -49,6 +48,37 @@ def admin_user_with_localhost_email(target_sat):
user.delete()


@pytest.fixture
def sysadmin_user_with_subscription_reposync_fail(target_sat):
"""System admin user with `root@localhost` e-mail
and subscription to `Repository sync failure` notification.
"""
sysadmin_role = target_sat.api.Role().search(query={'search': 'name="System admin"'})[0]
user = target_sat.api.User(
admin=False,
default_organization=DEFAULT_ORG,
default_location=DEFAULT_LOC,
description='created by nailgun',
login=gen_string("alphanumeric"),
password=gen_string("alphanumeric"),
mail='root@localhost',
role=[sysadmin_role.id],
).create()
user.mail_enabled = True
user.update()
target_sat.cli.User.mail_notification_add(
{
'user-id': user.id,
'mail-notification': 'repository_sync_failure',
'subscription': 'Subscribe',
}
)

yield user

user.delete()


@pytest.fixture
def reschedule_long_running_tasks_notification(target_sat):
"""Reschedule long-running tasks checker from midnight (default) to every minute.
Expand Down Expand Up @@ -76,7 +106,7 @@ def reschedule_long_running_tasks_notification(target_sat):
)


@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 @@ -97,33 +127,52 @@ 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:
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.'
)
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 wait_for_failed_repo_sync_mail(
target_sat, clean_root_mailbox, fake_yum_repo, failed_repo_sync_task
):
"""Wait until the repo name that didn't sync is found in the Satellite's mbox file."""
return wait_for_mail(
sat_obj=target_sat, mailbox_file=clean_root_mailbox, contains_string=fake_yum_repo.name
)


@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 @@ -174,11 +223,37 @@ def long_running_task(target_sat):
assert 'cancelled' in result


@pytest.fixture
def fake_yum_repo(target_sat):
"""Create a fake YUM repo. Delete it afterwards."""
repo = target_sat.api.Repository(
content_type='yum', url=repo_constants.FAKE_YUM_DRPM_REPO
).create()

yield repo

repo.delete()


@pytest.fixture
def failed_repo_sync_task(target_sat, fake_yum_repo):
"""
Do a repo sync that should fail. Return the result.
"""
fake_yum_repo.sync(synchronous=False)
task_result = target_sat.wait_for_tasks(
search_query=f"Synchronize repository '{fake_yum_repo.name}'", must_succeed=False
)[0]
task_status = target_sat.api.ForemanTask(id=task_result.id).poll(must_succeed=False)
assert task_status['result'] != 'success'
return task_status


@pytest.mark.tier3
@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 @@ -223,5 +298,43 @@ 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.tier3
@pytest.mark.usefixtures(
'sysadmin_user_with_subscription_reposync_fail',
'wait_for_failed_repo_sync_mail',
)
def test_positive_notification_failed_repo_sync(failed_repo_sync_task, root_mailbox_copy):
"""Check that a failed repository sync emits an email notification to the subscribed user.
:id: 19c477a2-8e39-11ee-9e3c-000c2989e153
:setup:
1. Create a user with 'System admin' role.
2. Subscribe the user to the 'Repository sync failure' email notification.
3. Create a non-existent YUM repository.
4. Run repository sync, it should fail.
:steps:
1. Check that email notification has been sent.
2. Check the e-mail if it contains all the important information, like,
repository name, failed task ID and link to the task.
:BZ: 1393613
:customerscenario: true
"""
task_id = failed_repo_sync_task['id']
repo_name = failed_repo_sync_task['input']['repository']['name']
product_name = failed_repo_sync_task['input']['product']['name']
for email in root_mailbox_copy:
if task_id in email.as_string():
assert f'Repository {repo_name} failed to synchronize' in email.get(
'Subject'
), f'Notification e-mail has wrong subject: {email.get("Subject")}'
for mime_body in email.get_payload():
body_text = mime_body.as_string()
assert product_name in body_text
assert f'/foreman_tasks/tasks/{task_id}' in body_text

0 comments on commit 662de6a

Please sign in to comment.