mirror of
https://github.com/saltstack/salt.git
synced 2025-04-17 10:10:20 +00:00
Migrate the service module integration tests to pytest
This commit is contained in:
MKLeb
Pedro Algarvio
parent
bba34f1963
commit
cb4f1be42e
4 changed files with 188 additions and 227 deletions
|
|
@ -30,7 +30,7 @@ salt/modules/(aix_group|groupadd|mac_group|pw_group|solaris_group|win_groupadd)\
|
|||
|
||||
salt/modules/(debian_service|freebsdservice|gentoo_service|launchctl_service|mac_service|netbsdservice|openbsdrcctl_service|openbsdservice|rh_service|runit|linux_service|smf_service|systemd_service|upstart_service|win_service)\.py:
|
||||
- pytests.unit.states.test_service
|
||||
- integration.modules.test_service
|
||||
- pytests.integration.modules.test_service
|
||||
- pytests.integration.states.test_service
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,183 +0,0 @@
|
|||
import pytest
|
||||
|
||||
import salt.utils.path
|
||||
import salt.utils.platform
|
||||
import salt.utils.systemd
|
||||
from tests.support.case import ModuleCase
|
||||
|
||||
|
||||
@pytest.mark.destructive_test
|
||||
@pytest.mark.windows_whitelisted
|
||||
class ServiceModuleTest(ModuleCase):
|
||||
"""
|
||||
Module testing the service module
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.service_name = "cron"
|
||||
cmd_name = "crontab"
|
||||
os_family = self.run_function("grains.get", ["os_family"])
|
||||
os_release = self.run_function("grains.get", ["osrelease"])
|
||||
if os_family == "RedHat":
|
||||
if os_release[0] == "7":
|
||||
self.skipTest(
|
||||
"Disabled on CentOS 7 until we can fix SSH connection issues."
|
||||
)
|
||||
self.service_name = "crond"
|
||||
elif os_family == "Arch":
|
||||
self.service_name = "sshd"
|
||||
cmd_name = "systemctl"
|
||||
elif os_family == "NILinuxRT":
|
||||
self.service_name = "syslog"
|
||||
cmd_name = "syslog-ng"
|
||||
elif os_family == "MacOS":
|
||||
self.service_name = "com.apple.AirPlayXPCHelper"
|
||||
elif salt.utils.platform.is_windows():
|
||||
self.service_name = "Spooler"
|
||||
|
||||
self.pre_srv_status = self.run_function("service.status", [self.service_name])
|
||||
self.pre_srv_enabled = (
|
||||
True
|
||||
if self.service_name in self.run_function("service.get_enabled")
|
||||
else False
|
||||
)
|
||||
|
||||
if (
|
||||
salt.utils.path.which(cmd_name) is None
|
||||
and not salt.utils.platform.is_windows()
|
||||
):
|
||||
self.skipTest("{} is not installed".format(cmd_name))
|
||||
|
||||
def tearDown(self):
|
||||
post_srv_status = self.run_function("service.status", [self.service_name])
|
||||
post_srv_enabled = (
|
||||
True
|
||||
if self.service_name in self.run_function("service.get_enabled")
|
||||
else False
|
||||
)
|
||||
|
||||
if post_srv_status != self.pre_srv_status:
|
||||
if self.pre_srv_status:
|
||||
self.run_function("service.enable", [self.service_name])
|
||||
else:
|
||||
self.run_function("service.disable", [self.service_name])
|
||||
|
||||
if post_srv_enabled != self.pre_srv_enabled:
|
||||
if self.pre_srv_enabled:
|
||||
self.run_function("service.enable", [self.service_name])
|
||||
else:
|
||||
self.run_function("service.disable", [self.service_name])
|
||||
del self.service_name
|
||||
|
||||
@pytest.mark.flaky(max_runs=4)
|
||||
@pytest.mark.slow_test
|
||||
def test_service_status_running(self):
|
||||
"""
|
||||
test service.status execution module
|
||||
when service is running
|
||||
"""
|
||||
self.run_function("service.start", [self.service_name])
|
||||
check_service = self.run_function("service.status", [self.service_name])
|
||||
self.assertTrue(check_service)
|
||||
|
||||
@pytest.mark.slow_test
|
||||
def test_service_status_dead(self):
|
||||
"""
|
||||
test service.status execution module
|
||||
when service is dead
|
||||
"""
|
||||
self.run_function("service.stop", [self.service_name])
|
||||
check_service = self.run_function("service.status", [self.service_name])
|
||||
self.assertFalse(check_service)
|
||||
|
||||
@pytest.mark.slow_test
|
||||
def test_service_restart(self):
|
||||
"""
|
||||
test service.restart
|
||||
"""
|
||||
self.assertTrue(self.run_function("service.restart", [self.service_name]))
|
||||
|
||||
@pytest.mark.slow_test
|
||||
def test_service_enable(self):
|
||||
"""
|
||||
test service.get_enabled and service.enable module
|
||||
"""
|
||||
# disable service before test
|
||||
self.assertTrue(self.run_function("service.disable", [self.service_name]))
|
||||
|
||||
self.assertTrue(self.run_function("service.enable", [self.service_name]))
|
||||
self.assertIn(self.service_name, self.run_function("service.get_enabled"))
|
||||
|
||||
@pytest.mark.slow_test
|
||||
def test_service_disable(self):
|
||||
"""
|
||||
test service.get_disabled and service.disable module
|
||||
"""
|
||||
# enable service before test
|
||||
self.assertTrue(self.run_function("service.enable", [self.service_name]))
|
||||
|
||||
self.assertTrue(self.run_function("service.disable", [self.service_name]))
|
||||
if salt.utils.platform.is_darwin():
|
||||
self.assertTrue(self.run_function("service.disabled", [self.service_name]))
|
||||
else:
|
||||
self.assertIn(self.service_name, self.run_function("service.get_disabled"))
|
||||
|
||||
@pytest.mark.slow_test
|
||||
def test_service_disable_doesnot_exist(self):
|
||||
"""
|
||||
test service.get_disabled and service.disable module
|
||||
when service name does not exist
|
||||
"""
|
||||
# enable service before test
|
||||
srv_name = "doesnotexist"
|
||||
enable = self.run_function("service.enable", [srv_name])
|
||||
systemd = salt.utils.systemd.booted()
|
||||
|
||||
# check service was not enabled
|
||||
try:
|
||||
self.assertFalse(enable)
|
||||
except AssertionError:
|
||||
self.assertIn("ERROR", enable)
|
||||
|
||||
# check service was not disabled
|
||||
if (
|
||||
tuple(
|
||||
self.run_function("grains.item", ["osrelease_info"])["osrelease_info"]
|
||||
)
|
||||
== (14, 0o4)
|
||||
and not systemd
|
||||
):
|
||||
# currently upstart does not have a mechanism to report if disabling a service fails if does not exist
|
||||
self.assertTrue(self.run_function("service.disable", [srv_name]))
|
||||
elif (
|
||||
self.run_function("grains.item", ["os"])["os"] == "Debian"
|
||||
and self.run_function("grains.item", ["osmajorrelease"])["osmajorrelease"]
|
||||
< 9
|
||||
and systemd
|
||||
):
|
||||
# currently disabling a service via systemd that does not exist
|
||||
# on Debian 8 results in a True return code
|
||||
self.assertTrue(self.run_function("service.disable", [srv_name]))
|
||||
else:
|
||||
try:
|
||||
disable = self.run_function("service.disable", [srv_name])
|
||||
self.assertFalse(disable)
|
||||
except AssertionError:
|
||||
self.assertTrue("error" in disable.lower())
|
||||
|
||||
if salt.utils.platform.is_darwin():
|
||||
self.assertEqual(
|
||||
self.run_function("service.disabled", [srv_name]),
|
||||
"ERROR: Service not found: {}".format(srv_name),
|
||||
)
|
||||
else:
|
||||
self.assertNotIn(srv_name, self.run_function("service.get_disabled"))
|
||||
|
||||
@pytest.mark.skip_unless_on_windows
|
||||
@pytest.mark.slow_test
|
||||
def test_service_get_service_name(self):
|
||||
"""
|
||||
test service.get_service_name
|
||||
"""
|
||||
ret = self.run_function("service.get_service_name")
|
||||
self.assertIn(self.service_name, ret.values())
|
||||
185
tests/pytests/integration/modules/test_service.py
Normal file
185
tests/pytests/integration/modules/test_service.py
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
import salt.utils.path
|
||||
import salt.utils.platform
|
||||
import salt.utils.systemd
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.windows_whitelisted,
|
||||
pytest.mark.destructive_test,
|
||||
pytest.mark.slow_test,
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service_name(grains, salt_cli, salt_minion):
|
||||
# For local testing purposes
|
||||
env_name = os.environ.get("SALT_INTEGRATION_TEST_SERVICE_NAME")
|
||||
if env_name is not None:
|
||||
return env_name
|
||||
|
||||
service_name = "cron"
|
||||
cmd_name = "crontab"
|
||||
os_family = grains["os_family"]
|
||||
is_systemd = grains["systemd"]
|
||||
if os_family == "RedHat":
|
||||
service_name = "crond"
|
||||
elif os_family == "Arch":
|
||||
service_name = "sshd"
|
||||
cmd_name = "systemctl"
|
||||
elif os_family == "MacOS":
|
||||
service_name = "com.apple.AirPlayXPCHelper"
|
||||
elif os_family == "Windows":
|
||||
service_name = "Spooler"
|
||||
|
||||
if os_family != "Windows" and salt.utils.path.which(cmd_name) is None:
|
||||
pytest.skip("{} is not installed".format(cmd_name))
|
||||
|
||||
if is_systemd and salt_cli.run("service.offline", minion_tgt=salt_minion.id):
|
||||
pytest.skip("systemd is OFFLINE")
|
||||
|
||||
return service_name
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_service(service_name, salt_cli, salt_minion):
|
||||
pre_srv_status = salt_cli.run("service.status", minion_tgt=salt_minion.id).data
|
||||
pre_srv_enabled = (
|
||||
service_name
|
||||
in salt_cli.run("service.get_enabled", minion_tgt=salt_minion.id).data
|
||||
)
|
||||
|
||||
yield pre_srv_status
|
||||
|
||||
post_srv_status = salt_cli.run("service.status", minion_tgt=salt_minion.id).data
|
||||
post_srv_enabled = (
|
||||
service_name
|
||||
in salt_cli.run("service.get_enabled", minion_tgt=salt_minion.id).data
|
||||
)
|
||||
|
||||
if post_srv_status != pre_srv_status:
|
||||
if pre_srv_status:
|
||||
salt_cli.run("service.enable", service_name, minion_tgt=salt_minion.id)
|
||||
else:
|
||||
salt_cli.run("service.disable", service_name, minion_tgt=salt_minion.id)
|
||||
|
||||
if post_srv_enabled != pre_srv_enabled:
|
||||
if pre_srv_enabled:
|
||||
salt_cli.run("service.enable", service_name, minion_tgt=salt_minion.id)
|
||||
else:
|
||||
salt_cli.run("service.disable", service_name, minion_tgt=salt_minion.id)
|
||||
|
||||
|
||||
@pytest.mark.flaky(max_runs=4)
|
||||
def test_service_status_running(salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
test service.status execution module
|
||||
when service is running
|
||||
"""
|
||||
salt_cli.run("service.start", service_name, minion_tgt=salt_minion.id)
|
||||
check_service = salt_cli.run(
|
||||
"service.status", service_name, minion_tgt=salt_minion.id
|
||||
).data
|
||||
assert check_service
|
||||
|
||||
|
||||
def test_service_status_dead(salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
test service.status execution module
|
||||
when service is dead
|
||||
"""
|
||||
salt_cli.run("service.stop", service_name, minion_tgt=salt_minion.id)
|
||||
check_service = salt_cli.run(
|
||||
"service.status", service_name, minion_tgt=salt_minion.id
|
||||
).data
|
||||
assert not check_service
|
||||
|
||||
|
||||
def test_service_restart(salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
test service.restart
|
||||
"""
|
||||
assert salt_cli.run("service.stop", service_name, minion_tgt=salt_minion.id).data
|
||||
|
||||
|
||||
def test_service_enable(salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
test service.get_enabled and service.enable module
|
||||
"""
|
||||
# disable service before test
|
||||
assert salt_cli.run("service.disable", service_name, minion_tgt=salt_minion.id).data
|
||||
|
||||
assert salt_cli.run("service.enable", service_name, minion_tgt=salt_minion.id).data
|
||||
assert (
|
||||
service_name
|
||||
in salt_cli.run("service.get_enabled", minion_tgt=salt_minion.id).data
|
||||
)
|
||||
|
||||
|
||||
def test_service_disable(salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
test service.get_disabled and service.disable module
|
||||
"""
|
||||
# enable service before test
|
||||
assert salt_cli.run("service.enable", service_name, minion_tgt=salt_minion.id).data
|
||||
|
||||
assert salt_cli.run("service.disable", service_name, minion_tgt=salt_minion.id).data
|
||||
if salt.utils.platform.is_darwin():
|
||||
assert salt_cli.run(
|
||||
"service.disabled", service_name, minion_tgt=salt_minion.id
|
||||
).data
|
||||
else:
|
||||
assert (
|
||||
service_name
|
||||
in salt_cli.run("service.get_disabled", minion_tgt=salt_minion.id).data
|
||||
)
|
||||
|
||||
|
||||
def test_service_disable_doesnot_exist(salt_cli, salt_minion):
|
||||
"""
|
||||
test service.get_disabled and service.disable module
|
||||
when service name does not exist
|
||||
"""
|
||||
# enable service before test
|
||||
srv_name = "doesnotexist"
|
||||
enable = salt_cli.run("service.enable", srv_name, minion_tgt=salt_minion.id).data
|
||||
systemd = salt.utils.systemd.booted()
|
||||
|
||||
# check service was not enabled
|
||||
try:
|
||||
assert not enable
|
||||
except AssertionError:
|
||||
assert "error" in enable.lower()
|
||||
|
||||
else:
|
||||
try:
|
||||
disable = salt_cli.run(
|
||||
"service.disable", srv_name, minion_tgt=salt_minion.id
|
||||
).data
|
||||
assert not disable
|
||||
except AssertionError:
|
||||
assert "error" in disable.lower()
|
||||
|
||||
if salt.utils.platform.is_darwin():
|
||||
assert (
|
||||
"ERROR: Service not found: {}".format(srv_name)
|
||||
in salt_cli.run(
|
||||
"service.disabled", srv_name, minion_tgt=salt_minion.id
|
||||
).stdout
|
||||
)
|
||||
else:
|
||||
assert (
|
||||
srv_name
|
||||
not in salt_cli.run("service.get_disabled", minion_tgt=salt_minion.id).data
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.skip_unless_on_windows
|
||||
def test_service_get_service_name(salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
test service.get_service_name
|
||||
"""
|
||||
ret = salt_cli.run("service.get_service_name", minion_tgt=salt_minion.id).data
|
||||
assert service_name in ret.data.values()
|
||||
|
|
@ -9,9 +9,6 @@ import pytest
|
|||
import salt.utils.path
|
||||
import salt.utils.platform
|
||||
|
||||
INIT_DELAY = 5
|
||||
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.windows_whitelisted,
|
||||
pytest.mark.destructive_test,
|
||||
|
|
@ -19,6 +16,7 @@ pytestmark = [
|
|||
]
|
||||
|
||||
|
||||
INIT_DELAY = 5
|
||||
STOPPED = False
|
||||
RUNNING = True
|
||||
|
||||
|
|
@ -26,7 +24,7 @@ RUNNING = True
|
|||
@pytest.fixture
|
||||
def service_name(grains, salt_cli, salt_minion):
|
||||
# For local testing purposes
|
||||
env_name = os.environ.get("SALT_SERVICE_STATE_TEST_SERVICE")
|
||||
env_name = os.environ.get("SALT_INTEGRATION_TEST_SERVICE_NAME")
|
||||
if env_name is not None:
|
||||
return env_name
|
||||
|
||||
|
|
@ -70,45 +68,6 @@ def setup_service(service_name, salt_cli, salt_minion):
|
|||
salt_cli.run("service.disable", service_name, minion_tgt=salt_minion.id)
|
||||
|
||||
|
||||
# def setUp(self):
|
||||
# self.service_name = "cron"
|
||||
# cmd_name = "crontab"
|
||||
# os_family = self.run_function("grains.get", ["os_family"])
|
||||
# os_release = self.run_function("grains.get", ["osrelease"])
|
||||
# is_systemd = self.run_function("grains.get", ["systemd"])
|
||||
# self.stopped = False
|
||||
# self.running = True
|
||||
# if os_family == "RedHat":
|
||||
# self.service_name = "crond"
|
||||
# elif os_family == "Arch":
|
||||
# self.service_name = "sshd"
|
||||
# cmd_name = "systemctl"
|
||||
# elif os_family == "MacOS":
|
||||
# self.service_name = "com.apple.AirPlayXPCHelper"
|
||||
# elif os_family == "Windows":
|
||||
# self.service_name = "Spooler"
|
||||
|
||||
# self.pre_srv_enabled = (
|
||||
# True
|
||||
# if self.service_name in self.run_function("service.get_enabled")
|
||||
# else False
|
||||
# )
|
||||
# self.post_srv_disable = False
|
||||
# if not self.pre_srv_enabled:
|
||||
# self.run_function("service.enable", name=self.service_name)
|
||||
# self.post_srv_disable = True
|
||||
|
||||
# if os_family != "Windows" and salt.utils.path.which(cmd_name) is None:
|
||||
# self.skipTest("{} is not installed".format(cmd_name))
|
||||
|
||||
# if is_systemd and self.run_function("service.offline"):
|
||||
# self.skipTest("systemd is OFFLINE")
|
||||
|
||||
# def tearDown(self):
|
||||
# if self.post_srv_disable:
|
||||
# self.run_function("service.disable", name=self.service_name)
|
||||
|
||||
|
||||
def check_service_status(exp_return, salt_cli, salt_minion, service_name):
|
||||
"""
|
||||
helper method to check status of service
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue