#!/usr/bin/python3

# This file is part of Cockpit.
#
# Copyright (C) 2013 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <http://www.gnu.org/licenses/>.

import parent
import testvm
from testlib import *


@nondestructive
@skipImage("Do not test BaseOS packages", "rhel-8-3-distropkg", "rhel-8-4-distropkg")
class TestServices(MachineCase):

    def setUp(self):
        super().setUp()

        # Make sure the system finishes "booting" so that
        # when we add additional services to this target below
        # we don't race with the boot process
        self.machine.execute("systemctl start default.target")

        # We manipulate with `/etc/systemd/system` so `daemon-reload` to keep it in proper state
        # Also `reset-failed` as failed units can be still listed with `systemctl` even when all files
        # were removed and daemon reloaded.
        self.restore_dir("/etc/systemd/system", "systemctl daemon-reload && systemctl reset-failed", True)

    def pick_tab(self, n):
        self.browser.click('#services-filter li:nth-child({0}) a'.format(n))

    def wait_onoff(self, val):
        self.browser.wait_visible(".service-top-panel .pf-c-switch__input" + (":checked" if val else ":not(:checked)"))

    def toggle_onoff(self):
        self.browser.click(".service-top-panel .pf-c-switch__input")

    def do_action(self, action):
        self.browser.click(".service-top-panel .pf-c-dropdown button")
        self.browser.click(".service-top-panel .pf-c-dropdown__menu a:contains('{0}')".format(action))

    def check_service_details(self, statuses, actions, enabled, onoff=True, kebab=True):
        self.browser.wait_collected_text("#statuses .status", "".join(statuses))
        if onoff:
            self.wait_onoff(enabled)
        else:
            self.browser.wait_not_present(".service-top-panel .pf-c-switch")

        if kebab:
            self.browser.click(".service-top-panel .pf-c-dropdown button")
            self.browser.wait_text(".service-top-panel .pf-c-dropdown__menu", "".join(actions))
            # Click away to close the pf-c-dropdown__menu
            self.browser.click(".service-top-panel .pf-c-dropdown button")
        else:
            self.browser.wait_not_present(".service-top-panel .pf-c-dropdown")

    def svc_sel(self, service):
        return 'li[data-goto-unit="%s"]' % service

    def wait_service_state(self, service, state):
        state_new = state
        if 'inactive' in state:
            state_new  = 'Not running'
        elif 'active' in state:
            state_new = 'Running'
        elif 'failed' in state:
            state_new = 'Failed to'

        self.browser.wait_in_text(self.svc_sel(service), state_new)

    def wait_page_load(self):
        self.browser.wait_not_present(".pf-c-empty-state .pf-c-spinner[aria-valuetext='Loading...']")

    def wait_service_in_panel(self, service, title):
        self.wait_service_state(service, title)

    def wait_service_present(self, service):
        if service.endswith(".service"):
            service_name = service[:-8]
        self.browser.wait_text('li[data-goto-unit="{0}"] .service-unit-id'.format(service), service_name)

    def testBasic(self):
        m = self.machine
        b = self.browser

        self.write_file("/etc/systemd/system/test.service",
                """
[Unit]
Description=Test Service

[Service]
ExecStart=/bin/sh /usr/local/bin/test-service
ExecReload=/bin/sh true

[Install]
WantedBy=default.target
""")
        self.write_file("/usr/local/bin/test-service",
                """
#! /bin/sh

trap "echo STOP" 0

echo START
while true; do
  sleep 5
  echo WORKING
done
""")
        self.write_file("/etc/systemd/system/test-fail.service",
                """
[Unit]
Description=Failing Test Service

[Service]
ExecStart=/usr/bin/false

[Install]
WantedBy=default.target
""")
        self.write_file("/etc/systemd/system/test.timer",
                """
[Unit]
Description=Test Timer

[Timer]
OnCalendar=*:1/2
""")
        self.write_file("/etc/systemd/system/test-onboot.timer",
                """
[Unit]
Description=Test OnBoot Timer

[Timer]
OnBootSec=200min
Unit=test.service
""")

        # After writing files out tell systemd about them
        m.execute("systemctl daemon-reload")
        self.machine.execute("systemctl start default.target")

        # When the test fails while `test.service` is active, the process keeps running and
        # `systemct status test.service` still shows it as active until this process dies
        self.addCleanup(self.machine.execute, "systemctl stop test")

        self.login_and_go("/system/services")
        self.wait_page_load()

        self.wait_service_present("test.service")
        self.wait_service_in_panel("test.service", "Disabled")
        self.wait_service_state("test.service", "inactive")

        m.execute("systemctl start test.service")
        self.wait_service_state("test.service", "active")
        self.wait_service_in_panel("test.service", "Disabled")

        m.execute("systemctl stop test.service")
        self.wait_service_state("test.service", "inactive")
        self.wait_service_in_panel("test.service", "Disabled")

        m.execute("systemctl enable test.service")
        self.wait_service_state("test.service", "inactive")
        self.wait_service_in_panel("test.service", "Enabled")

        b.wait_attr("#services-toolbar", "data-loading", "false")

        # Selects Targets tab
        self.pick_tab(2)
        b.wait_not_in_text("#services-list", "test")
        self.wait_service_state("sockets.target", "active")

        # Make sure that symlinked services also appear in the list
        self.wait_service_state("reboot.target", "inactive")
        self.wait_service_state("ctrl-alt-del.target", "inactive")
        self.wait_service_state("runlevel6.target", "inactive")
        b.click(self.svc_sel("runlevel6.target"))
        b.wait_in_text(".service-name", "Reboot")
        b.click("#services-page .pf-c-breadcrumb__link")

        # Selects Sockets tab
        self.pick_tab(3)
        b.wait_not_in_text("#services-list", "test")
        b.wait_not_in_text("#services-list", ".target")
        self.wait_service_state("dbus.socket", "active (running)")

        # Selects Timer tab
        self.pick_tab(4)
        b.wait_not_in_text("#services-list", "test-fail")
        b.wait_not_in_text("#services-list", ".target")
        b.wait_not_in_text("#services-list", ".socket")
        b.wait_visible(self.svc_sel('test.timer'))
        b.wait_text(self.svc_sel('test.timer') + ' .service-unit-triggers', '')
        m.execute("systemctl start test.timer")
        b.wait_in_text(self.svc_sel('test.timer') + ' .service-unit-next-trigger', "Today")  # next run
        b.wait_in_text(self.svc_sel('test.timer') + ' .service-unit-last-trigger', "unknown")  # last trigger
        m.execute("systemctl stop test.timer")

        b.wait_visible(self.svc_sel('test-onboot.timer'))
        b.wait_text(self.svc_sel('test-onboot.timer') + ' .service-unit-triggers', '')
        m.execute("systemctl start test-onboot.timer")
        # Check the next run. Since it triggers 200mins after the boot, it might be today or tomorrow (after 20:40)
        try:
            b.wait_in_text(self.svc_sel('test-onboot.timer') + ' .service-unit-next-trigger', "Today")
        except Error:
            b.wait_in_text(self.svc_sel('test-onboot.timer') + ' .service-unit-next-trigger', "Tomorrow")
        b.wait_in_text(self.svc_sel('test-onboot.timer') + ' .service-unit-last-trigger', "unknown")  # last trigger
        m.execute("systemctl stop test-onboot.timer")

        # Make sure that test.service was not started by either test.timer or test-onboot.timer
        m.execute("systemctl stop test.service")

        # Selects Paths tab
        self.pick_tab(5)
        b.wait_not_in_text("#services-list", "test")
        b.wait_not_in_text("#services-list", ".target")
        b.wait_not_in_text("#services-list", ".socket")
        b.wait_not_in_text("#services-list", ".timer")
        b.wait_visible(self.svc_sel("systemd-ask-password-console.path"))

        # Selects Services tab
        self.pick_tab(1)

        # Check test.service and then start it
        b.click(self.svc_sel("test.service"))
        self.check_service_details(["Not running", "Automatically starts"], ["Start", "Disallow running (mask)"], True)
        self.do_action("Start")
        self.check_service_details(["Running", "Automatically starts"], ["Reload", "Restart", "Stop", "Disallow running (mask)"], True)
        b.wait_in_text('.cockpit-log-panel .pf-c-card__body', "START")
        b.wait_in_text('.cockpit-log-panel .pf-c-card__body', "WORKING")
        b.click(".cockpit-log-panel .pf-c-card__header button")

        b.enter_page("/system/logs")
        b.wait_in_text("#journal-box .cockpit-log-panel > div:nth-child(2)", "WORKING")
        t = b.text("#journal-box .cockpit-log-panel > div:nth-child(3)") + b.text("#journal-box .cockpit-log-panel > div:nth-child(4)")
        self.assertIn("START", t)
        # Debian supports grep only from 19.10, see https://bugs.launchpad.net/ubuntu/+source/systemd/+bug/1751006
        if m.image != "debian-stable":
            b.wait_val("#journal-grep", "priority:debug service:test.service ")
        b.go("/system/services#test.service")
        b.enter_page("/system/services")

        b.click(".cockpit-logline:contains('WORKING')")
        b.enter_page("/system/logs")
        b.wait_text(".pf-c-card__title", "WORKING")
        b.click(".pf-c-breadcrumb a:contains('Logs')")
        # Debian supports grep only from 19.10, see https://bugs.launchpad.net/ubuntu/+source/systemd/+bug/1751006
        if m.image != "debian-stable":
            b.wait_val("#journal-grep", "priority:debug service:test.service ")
        b.go("/system/services#test.service")
        b.enter_page("/system/services")

        b.click(".cockpit-logline:contains('WORKING')")
        b.enter_page("/system/logs")
        b.click("button:contains('Go to test.service')")
        b.enter_page("/system/services")
        b.wait_js_cond('window.location.hash === "#/test.service"')

        # Stop and disable and back again
        self.toggle_onoff()
        self.check_service_details(["Disabled"], ["Start", "Disallow running (mask)"], False)
        self.toggle_onoff()
        self.check_service_details(["Running", "Automatically starts"], ["Reload", "Restart", "Stop", "Disallow running (mask)"], True)

        # Survives a burst of events
        b.go("#/")
        self.wait_service_present("test.service")
        m.execute("udevadm trigger; udevadm settle")
        b.click(self.svc_sel("test.service"))
        self.check_service_details(["Running", "Automatically starts"], ["Reload", "Restart", "Stop", "Disallow running (mask)"], True)
        self.toggle_onoff()
        self.check_service_details(["Disabled"], ["Start", "Disallow running (mask)"], False)
        self.toggle_onoff()
        self.check_service_details(["Running", "Automatically starts"], ["Reload", "Restart", "Stop", "Disallow running (mask)"], True)

        # Check without permissions
        b.relogin('/system/services#/test.service', superuser=False)
        self.wait_page_load()

        b.relogin('/system/services#/test.service', superuser=True)
        self.wait_page_load()

        self.toggle_onoff() # later on we need some disabled test
        self.check_service_details(["Disabled"], ["Start", "Disallow running (mask)"], False)

        # Check service that fails to start
        b.go("#/")
        b.click(self.svc_sel("test-fail.service"))
        self.check_service_details(["Disabled"], ["Start", "Disallow running (mask)"], False)
        self.toggle_onoff()
        self.check_service_details(["Failed to start", "Automatically starts"], ["Start", "Clear 'Failed to start'", "Disallow running (mask)"], True)
        b.click(".action-button:contains('Start service')")
        b.go("#/")
        self.wait_service_present("test-fail.service")
        self.wait_service_state("test-fail.service", "failed")

        # Check static service
        b.click(self.svc_sel("systemd-halt.service"))
        self.check_service_details(["Static", "Not running"], ["Start", "Disallow running (mask)"], True, onoff=False)

        # Mask and unmask
        self.do_action("Disallow running (mask)")
        b.click("#mask-service button.pf-m-danger")
        self.check_service_details(["Masked"], ["Allow running (unmask)"], False, onoff=False)
        self.do_action("Allow running (unmask)")
        self.check_service_details(["Static", "Not running"], ["Start", "Disallow running (mask)"], True, onoff=False)

        def wait_service_not_present(service):
            b.wait_not_present(self.svc_sel(service))

        def init_filter_state():
            b.set_input_text("#services-text-filter", "")
            self.wait_service_present("test.service")
            self.wait_service_present("test-fail.service")
            b.select_from_dropdown("#services-dropdown", "all")

        b.go("#/")

        # Filter by id
        init_filter_state()
        b.set_input_text("#services-text-filter", "fail.ser")
        wait_service_not_present("test.service")
        self.wait_service_present("test-fail.service")

        # Filter by description capital letters included
        init_filter_state()
        b.set_input_text("#services-text-filter", "Test Service")
        self.wait_service_present("test.service")
        self.wait_service_present("test-fail.service")

        # Filter by description capitalization not matching the unit description
        init_filter_state()
        b.set_input_text("#services-text-filter", "failing test service")
        wait_service_not_present("test.service")
        self.wait_service_present("test-fail.service")

        # Filter by Id capitalization not matching the unit Id
        init_filter_state()
        b.set_input_text("#services-text-filter", "networkmanager")
        self.wait_service_present("NetworkManager.service")

        # Select only static services
        init_filter_state()
        b.select_from_dropdown("#services-dropdown", "static")
        wait_service_not_present("test.service")
        wait_service_not_present("test-fail.service")
        self.wait_service_present("systemd-halt.service")

        # Select only disabled services
        b.select_from_dropdown("#services-dropdown", "disabled")
        self.wait_service_present("test.service")
        wait_service_not_present("test-fail.service")
        wait_service_not_present("systemd-halt.service")

        # Check filtering and selecting together - empty state
        b.set_input_text("#services-text-filter", "failing")
        wait_service_not_present("test.service")
        b.wait_visible("#services-page .pf-c-empty-state")

        # Check resetting filter
        b.click("#clear-all-filters")
        self.wait_service_present("test.service")
        self.wait_service_present("test-fail.service")
        self.wait_service_present("systemd-halt.service")
        self.assertEqual(b.val("#services-text-filter"), "")
        self.assertEqual(b.val("#services-dropdown"), "all")

        # Check that journalbox shows empty state
        b.click(self.svc_sel("systemd-halt.service"))
        b.wait_text('.cockpit-log-panel .pf-c-card__body', "No log entries")

    def testApi(self):
        m = self.machine
        b = self.browser

        self.write_file("/etc/systemd/system/test.service",
                """
[Unit]
Description=Test Service

[Service]
ExecStart=/bin/sh -c "while true; do sleep 5; done"

[Install]
WantedBy=default.target
""")

        # After writing files out tell systemd about them
        m.execute("systemctl daemon-reload")
        m.execute("systemctl stop test")

        self.login_and_go("/playground/service#/test")

        b.wait_text('#exists', 'true')
        b.wait_text('#state', '"stopped"')
        b.wait_text('#enabled', 'false')

        b.click('#start')
        b.wait_text('#state', '"running"')
        b.click('#stop')
        b.wait_text('#state', '"stopped"')

        b.click('#enable')
        b.wait_text('#enabled', 'true')
        b.click('#disable')
        b.wait_text('#enabled', 'false')

        b.go('#/foo')
        b.wait_text('#exists', 'false')

    def testConditions(self):
        m = self.machine
        b = self.browser
        self.write_file("/etc/systemd/system/condtest.service",
                """
[Unit]
Description=Test Service
ConditionDirectoryNotEmpty=/var/tmp/empty

[Service]
ExecStart=/bin/sh -c "while true; do sleep 5; done"

[Install]
WantedBy=multi-user.target
""")

        m.execute("mkdir -p /var/tmp/empty")
        m.execute("rm -rf /var/tmp/empty/*")
        self.addCleanup(m.execute, "systemctl stop condtest")

        # After writing files out tell systemd about them
        m.execute("systemctl daemon-reload")
        self.machine.execute("systemctl start multi-user.target")

        # This does not work for not enabled services. See:
        # https://github.com/systemd/systemd/issues/2234
        self.machine.execute("systemctl enable condtest")

        self.login_and_go("/system/services")

        # Selects Services tab
        self.pick_tab(1)

        self.wait_service_in_panel("condtest.service", "Enabled")
        b.click(self.svc_sel("condtest.service"))
        self.check_service_details(["Not running", "Automatically starts"], ["Start", "Disallow running (mask)"], True)
        self.do_action("Start")
        b.wait_text("#condition", "Condition ConditionDirectoryNotEmpty=/var/tmp/empty was not met")

        # If the condition succeeds the message disappears
        m.execute("touch /var/tmp/empty/non-empty")
        self.addCleanup(m.execute, "rm /var/tmp/empty/non-empty")
        self.do_action("Start")
        self.check_service_details(["Running", "Automatically starts"], ["Restart", "Stop", "Disallow running (mask)"], True)
        b.wait_not_present("#condition")

    def testRelationships(self):
        m = self.machine
        b = self.browser

        self.write_file("/etc/systemd/system/test-a.service",
                """
[Unit]
Description=A Service
Before=test-b.service
Conflicts=test-c.service

[Service]
ExecStart=/bin/sh -c "while true; do sleep 5; done"
""")

        self.write_file("/etc/systemd/system/test-b.service",
                """
[Unit]
Description=B Service
After=test-a.service

[Service]
ExecStart=/bin/sh -c "while true; do sleep 5; done"
""")

        self.write_file("/etc/systemd/system/test-c.service",
                """
[Unit]
Description=C Service
Conflicts=test-a.service
PartOf=test-b.service

[Service]
ExecStart=/bin/sh -c "while true; do sleep 5; done"
""")

        # After writing files out tell systemd about them
        m.execute("systemctl daemon-reload")

        def rel_sel(reltype, service):
            return "#{0} a:contains('{1}')".format("".join(reltype.split(" ")), service)

        # services page
        self.login_and_go("/system/services")
        self.wait_page_load()
        self.wait_service_present("test-a.service")
        b.click(self.svc_sel("test-a.service"))

        # service a
        b.wait_js_cond('window.location.hash === "#/test-a.service"')
        b.wait_visible(rel_sel("Before", "test-b.service"))
        b.wait_visible(rel_sel("Conflicts", "test-c.service"))
        b.click(rel_sel("Before", "test-b.service"))

        # service b
        b.wait_js_cond('window.location.hash === "#/test-b.service"')
        b.wait_visible(rel_sel("After", "test-a.service"))
        b.click(rel_sel("After", "test-a.service"))

        # service a
        b.wait_js_cond('window.location.hash === "#/test-a.service"')
        b.wait_visible(rel_sel("Conflicts", "test-c.service"))
        b.click(rel_sel("Conflicts", "test-c.service"))

        # service c
        b.wait_js_cond('window.location.hash === "#/test-c.service"')
        b.wait_visible(rel_sel("Conflicts", "test-a.service"))
        b.wait_visible(rel_sel("Part of", "test-b.service"))
        b.click(rel_sel("Part of", "test-b.service"))

        # service b
        b.wait_js_cond('window.location.hash === "#/test-b.service"')
        b.wait_visible(rel_sel("After", "test-a.service"))

    def testNotFound(self):
        m = self.machine
        b = self.browser

        self.write_file("/etc/systemd/system/test.service",
                """
[Unit]
Description=Test Service
Requires=not-found.service

[Service]
ExecStart=/bin/true
""")
        m.execute("systemctl daemon-reload")
        self.machine.execute("systemctl start default.target")

        self.login_and_go("/system/services")
        self.wait_page_load()

        b.wait_visible(self.svc_sel("test.service"))
        b.wait_not_present(self.svc_sel("not-found.service"))

        b.click(self.svc_sel("test.service"))
        b.wait_js_cond('window.location.hash === "#/test.service"')

        b.wait_attr("#Requires a:contains('not-found.service')", "class", "disabled")

    def testResetFailed(self):
        m = self.machine
        b = self.browser

        # Put it in /run instead of in /etc so that we can mask it,
        # which needs to put a symlink into /etc.
        self.write_file("/run/systemd/system/test-fail.service",
                """
[Unit]
Description=Failing Test Service

[Service]
ExecStart=/usr/bin/false

[Install]
WantedBy=default.target
""")

        m.execute("systemctl daemon-reload")
        self.machine.execute("systemctl start default.target")

        self.login_and_go("/system/services")
        self.wait_page_load()

        b.click(self.svc_sel("test-fail.service"))
        b.wait_js_cond('window.location.hash === "#/test-fail.service"')

        self.check_service_details(["Disabled"], ["Start", "Disallow running (mask)"], False)

        self.wait_onoff(False)
        self.toggle_onoff()

        self.check_service_details(["Failed to start", "Automatically starts"],
                                   ["Start", "Clear 'Failed to start'", "Disallow running (mask)"], True)

        # Disabling should reset the "Failed to start" status
        self.wait_onoff(True)
        self.toggle_onoff()

        self.check_service_details(["Disabled"], ["Start", "Disallow running (mask)"], False)

        self.wait_onoff(False)
        self.toggle_onoff()

        self.check_service_details(["Failed to start", "Automatically starts"],
                                   ["Start", "Clear 'Failed to start'", "Disallow running (mask)"], True)

        # Just reset the failed status, but leave it enabled
        self.do_action("Clear")

        self.check_service_details(["Not running", "Automatically starts"],
                                   ["Start", "Disallow running (mask)"], True)

        self.do_action("Start")
        self.check_service_details(["Failed to start", "Automatically starts"],
                                   ["Start", "Clear 'Failed to start'", "Disallow running (mask)"], True)

        # Masking should also clear the failed status
        self.do_action("Disallow running (mask)")
        b.click("#mask-service button.pf-m-danger")
        self.check_service_details(["Masked"], ["Allow running (unmask)"], False, onoff=False)

    def count_failures(self):
        return int(self.machine.execute("systemctl --failed --plain --no-legend | wc -l"))

    def testNotifyFailed(self):
        m = self.machine
        b = self.browser

        # We use two services with prefix "aaa" and "aab" so that
        # they always occupy the first two rows and we can check their
        # relative order depending on whether "aab" has failed or not.

        self.write_file("/etc/systemd/system/aaa.service",
                """
[Unit]
Description=First Row Service

[Service]
ExecStart=/usr/bin/true
""")
        self.write_file("/etc/systemd/system/aab-fail.service",
                """
[Unit]
Description=Failing Test Service

[Service]
ExecStart=/usr/bin/false
""")

        m.execute("systemctl daemon-reload")
        self.machine.execute("systemctl start default.target")
        self.machine.execute("systemctl reset-failed")

        self.login_and_go("/system/services")
        self.wait_page_load()

        # Start test-fail
        m.execute("systemctl start aab-fail")

        # Services tab should have icon
        # Aab-fail should be at the top, and have failed
        b.wait_visible('a:contains("System services") .fa-exclamation-circle')
        b.wait_visible('li[data-goto-unit="aab-fail.service"]')
        b.wait_visible('li[data-goto-unit="aab-fail.service"] .service-unit-status:contains("Failed")')

        # Nav link should have icon
        b.switch_to_top()
        b.wait_visible("#services-error")

        n_failures = self.count_failures()
        health_message = "1 service has failed" if n_failures == 1 else "%s services have failed" % n_failures

        # System page should have notification
        b.click('#host-apps a[href="/system"]')
        b.enter_page('/system')
        b.wait_in_text(".system-health-events", health_message)

        # Clear aab-fail
        m.execute("systemctl reset-failed aab-fail")

        # aab-fail should not be at the top
        b.switch_to_top()
        b.click('#host-apps a[href="/system/services"]')
        b.enter_page('/system/services')
        b.wait_not_present('li:nth-child(1)[data-goto-unit="aab-fail.service"]')

        if self.count_failures() == 0:
            # Services tab should not have icon
            b.wait_not_present('a:contains("System services") .fa-exclamation-circle')

            # Nav link should not have icon
            b.switch_to_top()
            b.wait_not_present("#services-error")

            # System page should not have notification
            b.click('#host-apps a[href="/system"]')
            b.enter_page('/system')
            b.wait_not_in_text(".system-health-events", "service has failed")
            b.wait_not_in_text(".system-health-events", "services have failed")

    def testHiddenFailure(self):
        m = self.machine
        b = self.browser

        self.write_file("/etc/systemd/system/fail.mount",
                """
[Unit]
Description=Failing Mount

[Mount]
What=wrong
Where=/fail
""")

        m.execute("systemctl daemon-reload")
        self.machine.execute("systemctl start default.target")
        self.machine.execute("systemctl reset-failed")
        m.execute("systemctl start fail.mount || true")

        self.login_and_go("/system/services")
        self.wait_page_load()

        if self.count_failures() == 1:
            # Nav link should not have icon
            b.switch_to_top()
            b.wait_not_present("#services-error")

            # System page should not have notification
            b.click('#host-apps a[href="/system"]')
            b.enter_page('/system')
            b.wait_not_in_text(".system-health-events", "service has failed")
            b.wait_not_in_text(".system-health-events", "services have failed")

        self.allow_journal_messages(".*type=1400 audit(.*): avc:  denied  { create } .* comm=\"systemd\" name=\"fail\".*")


@skipImage("Do not test BaseOS packages", "rhel-8-3-distropkg", "rhel-8-4-distropkg")
class TestTimers(MachineCase):
    def svc_sel(self, service):
        return 'li[data-goto-unit="%s"]' % service

    def wait_page_load(self):
        self.browser.wait_not_present(".pf-c-empty-state .pf-c-spinner[aria-valuetext='Loading...']")

    def testCreate(self):
        m = self.machine
        b = self.browser

        def wait_systemctl_timer(time):
            with testvm.Timeout(seconds=20, machine=m, error_message="Timeout while waiting for systemctl list-timers"):
                m.execute("cmd='systemctl list-timers'; until $cmd | grep -m 1 '%s'; do sleep 1; done" % time)

        # HACK: pmie and pmlogger take a looooong time to shutdown (https://bugzilla.redhat.com/show_bug.cgi?id=1703348)
        # so disable them for this test, we don't test PCP here
        m.execute("systemctl disable --now pmie pmlogger || true")

        # set an initial baseline date/time, to ensure that we never jump backwards in subsequent tests
        m.execute("timedatectl set-timezone UTC")
        m.execute("""ntp=`timedatectl show --property NTP --value`
                     if [ $ntp == "yes" ]; then
                         timedatectl set-ntp off
                     fi""")
        wait(lambda: "false" in m.execute(
            "busctl get-property org.freedesktop.timedate1 /org/freedesktop/timedate1 org.freedesktop.timedate1 NTP"))
        m.execute("timedatectl set-time '2036-01-01 12:00:00'")
        m.spawn("sync && sync && sync && sleep 0.1 && reboot", "reboot")
        m.wait_reboot()

        m.execute("timedatectl set-time '2036-01-01 15:30:00'")
        self.login_and_go("/system/services")
        self.wait_page_load()
        # Select "Timers" tab
        self.browser.click('#services-filter li:nth-child(4) a')
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "testing timer")
        m.execute("rm -f /tmp/date")
        b.set_input_text("#command", "/bin/sh -c '/bin/date >> /tmp/date'")
        b.select_from_dropdown('#boot-or-specific-time', 'specific-time')
        b.set_input_text(".create-timer-time-picker input", "24:6s")
        b.blur(".create-timer-time-picker input")
        b.click("#timer-save-button")

        # checks for invalid input messages
        b.wait_text("#servicename-helper", "Only alphabets, numbers, : , _ , . , @ , - are allowed")
        b.wait_text("#description-helper", "This field cannot be empty")
        b.wait_text(".pf-c-date-picker__helper-text", "Invalid time format")

        # creates a new yearly timer at 2036-01-01 16:00 and at 2037-01-01 01:22
        b.set_input_text("#servicename", "yearly_timer")
        b.set_input_text("#description", "Yearly timer")
        b.select_from_dropdown("#drop-repeat", "yearly")
        b.click("[data-index=0] [aria-label=Add]")
        b.set_input_text("[data-index=0] .pf-c-date-picker:nth-child(1) input", "2036-01-01")
        b.set_input_text("[data-index=0] .pf-c-date-picker:nth-child(2) input", "16:00")
        b.set_input_text("[data-index=1] .pf-c-date-picker:nth-child(1) input", "2037-01-01")
        b.set_input_text("[data-index=1] .pf-c-date-picker:nth-child(2) input", "01:22")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('yearly_timer.timer'))

        m.execute("timedatectl set-time '2036-01-01 15:30:00'")
        b.wait_visible(self.svc_sel('yearly_timer.timer'))
        b.wait_in_text(self.svc_sel('yearly_timer.timer'), "Yearly timer")
        wait_systemctl_timer("2036-01-01 16:00")
        self.assertIn("2036-01-01 16:00", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-01-01 16:10:00'")
        # checks if yearly timer repeats yearly on 2037-01-01 01:22
        wait_systemctl_timer("2037-01-01 01:22")
        self.assertIn("2037-01-01 01:22", m.execute("systemctl list-timers"))
        # creates a new monthly timer that runs on 6th at 14:12 and 8th at 21:12 of every month
        b.wait_visible("#create-timer")
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "monthly_timer")
        b.set_input_text("#description", "Monthly timer")
        b.set_input_text("#command", "/bin/sh -c '/bin/date >> /tmp/date'")
        b.select_from_dropdown("#boot-or-specific-time", "specific-time")
        b.select_from_dropdown("#drop-repeat", "monthly")
        b.select_from_dropdown("[data-index=0] .month-days", "6")
        b.set_input_text("[data-index=0] .create-timer-time-picker input", "14:12")
        b.click("[data-index=0] [aria-label=Add]")
        b.select_from_dropdown("[data-index=1] .month-days", "8")
        b.set_input_text("[data-index=1] .create-timer-time-picker input", "21:12")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('monthly_timer.timer'))

        m.execute("timedatectl set-time '2036-01-01 16:15:00'")
        b.wait_visible(self.svc_sel('monthly_timer.timer'))
        b.wait_in_text(self.svc_sel('monthly_timer.timer'), "Monthly timer")
        wait_systemctl_timer("2036-01-06 14:12")
        self.assertIn("2036-01-06 14:12", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-01-07 00:00:00'")
        wait_systemctl_timer("2036-01-08 21:12")
        self.assertIn("2036-01-08 21:12", m.execute("systemctl list-timers"))
        # checks if timer runs on next month February 2036 on same dates
        m.execute("timedatectl set-time '2036-01-08 21:23'")
        wait_systemctl_timer("2036-02-06 14:12")
        self.assertIn("2036-02-06 14:12", m.execute("systemctl list-timers"))
        # checks if timer runs on 8th March 2036 at 21:12
        m.execute("timedatectl set-time '2036-03-07 00:00:00'")
        wait_systemctl_timer("2036-03-08 21:12")
        self.assertIn("2036-03-08 21:12", m.execute("systemctl list-timers"))
        # creates a new weekly timer that runs on Fri at 12:45 and Sun at 20:12 every week
        b.wait_visible("#create-timer")
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "weekly_timer")
        b.set_input_text("#description", "Weekly timer")
        b.set_input_text("#command", "/bin/sh -c '/bin/date >> /tmp/date'")
        b.select_from_dropdown("#boot-or-specific-time", "specific-time")
        b.select_from_dropdown("#drop-repeat", "weekly")
        b.wait_visible("[data-index=0] .week-days")
        b.click("[data-index=0] [aria-label=Add]")
        b.select_from_dropdown("[data-index=0] .week-days", "fri")
        b.set_input_text("[data-index=0] .create-timer-time-picker input", "12:45")
        b.select_from_dropdown("[data-index=1] .week-days", "sun")
        b.set_input_text("[data-index=1] .create-timer-time-picker input", "20:12")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('weekly_timer.timer'))

        m.execute("timedatectl set-time '2036-03-08 00:00:00'")
        b.wait_visible(self.svc_sel('weekly_timer.timer'))
        b.wait_in_text(self.svc_sel('weekly_timer.timer'), "Weekly timer")
        wait_systemctl_timer("Sun 2036-03-09 20:12")
        self.assertIn("Sun 2036-03-09 20:12", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-03-10 00:00:00'")
        wait_systemctl_timer("Fri 2036-03-14 12:45")
        self.assertIn("Fri 2036-03-14 12:45", m.execute("systemctl list-timers"))
        # checks if timer runs on next week's Friday and Sunday
        m.execute("timedatectl set-time '2036-03-15 00:00:00'")
        wait_systemctl_timer("Sun 2036-03-16 20:12")
        self.assertIn("Sun 2036-03-16 20:12", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-03-17 00:00:00'")
        wait_systemctl_timer("Fri 2036-03-21 12:45")
        self.assertIn("Fri 2036-03-21 12:45", m.execute("systemctl list-timers"))
        # creates a new daily timer that runs at 2:40 and at 21:15 every day
        b.wait_visible("#create-timer")
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "daily_timer")
        b.set_input_text("#description", "Daily timer")
        b.set_input_text("#command", "/bin/sh -c '/bin/date >> /tmp/date'")
        b.select_from_dropdown("#boot-or-specific-time", "specific-time")
        b.select_from_dropdown("#drop-repeat", "daily")
        b.click("[data-index=0] [aria-label=Add]")
        b.set_input_text("[data-index=0] .create-timer-time-picker input", "2:40")
        b.set_input_text("[data-index=1] .create-timer-time-picker input", "21:15")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('daily_timer.timer'))

        m.execute("timedatectl set-time '2036-03-17 00:00:00'")
        b.wait_visible(self.svc_sel('daily_timer.timer'))
        b.wait_in_text(self.svc_sel('daily_timer.timer'), "Daily timer")
        wait_systemctl_timer("2036-03-17 02:40")
        self.assertIn("2036-03-17 02:40", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-03-17 03:00:00'")
        wait_systemctl_timer("2036-03-17 21:15")
        self.assertIn("2036-03-17 21:15", m.execute("systemctl list-timers"))
        # checks if timer runs on 2036-04-10 at 02:40 and 21:15
        m.execute("timedatectl set-time '2036-04-10 00:00:00'")
        wait_systemctl_timer("2036-04-10 02:40")
        self.assertIn("2036-04-10 02:40", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-04-10 03:00:00'")
        wait_systemctl_timer("2036-04-10 21:15")
        self.assertIn("2036-04-10 21:15", m.execute("systemctl list-timers"))
        # creates a new houry timer that runs at *:05 and at *:26
        b.wait_visible("#create-timer")
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "hourly_timer")
        b.set_input_text("#description", "Hourly timer")
        b.set_input_text("#command", "/bin/sh -c '/bin/date >> /tmp/date'")
        b.select_from_dropdown("#boot-or-specific-time", "specific-time")
        b.select_from_dropdown("#drop-repeat", "hourly")
        b.click("[data-index=0] [aria-label=Add]")
        b.set_input_text("[data-index=0] input", "05")
        b.set_input_text("[data-index=1] input", "26")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('hourly_timer.timer'))

        m.execute("timedatectl set-time '2036-04-10 03:00:00'")
        b.wait_visible(self.svc_sel('hourly_timer.timer'))
        b.wait_in_text(self.svc_sel('hourly_timer.timer'), "Hourly timer")
        wait_systemctl_timer("2036-04-10 03:05")
        self.assertIn("2036-04-10 03:05", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-04-10 03:07:00'")
        wait_systemctl_timer("2036-04-10 03:26")
        self.assertIn("2036-04-10 03:26", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-04-10 04:00:00'")
        wait_systemctl_timer("2036-04-10 04:05")
        # checks if timer runs on next hour at 5 min and 26 min
        self.assertIn("2036-04-10 04:05", m.execute("systemctl list-timers"))
        m.execute("timedatectl set-time '2036-04-10 04:10:00'")
        wait_systemctl_timer("2036-04-10 04:26")
        self.assertIn("2036-04-10 04:26", m.execute("systemctl list-timers"))
        # creates a new timer that runs at today at 23:59
        b.wait_visible("#create-timer")
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "no_repeat_timer")
        b.set_input_text("#description", "No repeat timer")
        b.set_input_text("#command", "/bin/sh -c '/bin/date >> /tmp/date'")
        b.select_from_dropdown("#boot-or-specific-time", "specific-time")
        b.set_input_text(".create-timer-time-picker input", "23:59")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('no_repeat_timer.timer'))
        b.wait_in_text(self.svc_sel('no_repeat_timer.timer'), "No repeat timer")

        m.execute("timedatectl set-time '2036-04-10 04:10:00'")
        b.wait_visible(self.svc_sel('no_repeat_timer.timer'))
        wait_systemctl_timer("2036-04-10 23:59")
        self.assertIn("2036-04-10 23:59", m.execute("systemctl list-timers"))

        # creates a boot timer that runs after 10 sec from boot
        b.wait_visible("#create-timer")
        b.click('#create-timer')
        b.wait_visible("#timer-dialog")
        b.set_input_text("#servicename", "boot_timer")
        b.set_input_text("#description", "Boot timer")
        b.set_input_text("#command", "/bin/sh -c 'echo hello >> /tmp/hello'")
        b.select_from_dropdown("#boot-or-specific-time", "system-boot")
        b.set_input_text(".delay-group input", "2")
        b.click("#timer-save-button")
        b.wait_not_present("#timer-dialog")
        b.wait_visible(self.svc_sel('boot_timer.timer'))
        m.spawn("sync && sync && sync && sleep 0.1 && reboot", "reboot")
        m.wait_reboot()
        m.start_cockpit()
        with testvm.Timeout(seconds=15, machine=m, error_message="Timeout while waiting for boot timer to run"):
            m.execute("while [ ! -f /tmp/hello ] ; do sleep 0.5; done")
        self.assertIn("hello", m.execute("cat /tmp/hello"))

        self.allow_restart_journal_messages()


if __name__ == '__main__':
    test_main()