diff --git a/robottelo/cli/user.py b/robottelo/cli/user.py index e624a1b5ad4..eb37133e786 100644 --- a/robottelo/cli/user.py +++ b/robottelo/cli/user.py @@ -105,3 +105,29 @@ 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') diff --git a/tests/foreman/api/test_notifications.py b/tests/foreman/api/test_notifications.py index a8b4c24ebf8..9bdbaa01f52 100644 --- a/tests/foreman/api/test_notifications.py +++ b/tests/foreman/api/test_notifications.py @@ -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 @@ -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. @@ -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 @@ -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 @@ -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) @@ -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