mgr/cephadm/tests: move with_service to fixtures.py

Signed-off-by: Sebastian Wagner <sebastian.wagner@suse.com>

diff --git a/src/pybind/mgr/cephadm/tests/fixtures.py b/src/pybind/mgr/cephadm/tests/fixtures.py
index a65f501926458ab43b5253d87618e2c08a3ae9d4..045a3015dc0015eeed1aaae1eeacea9ec5912f02 100644 (file)
--- a/src/pybind/mgr/cephadm/tests/fixtures.py
+++ b/src/pybind/mgr/cephadm/tests/fixtures.py
@@ -3,6 +3,7 @@ import time
 import fnmatch
 from contextlib import contextmanager
 
+from ceph.deployment.service_spec import PlacementSpec, ServiceSpec
 from cephadm.module import CEPH_DATEFMT
 
 try:
@@ -117,3 +118,28 @@ def with_host(m:CephadmOrchestrator, name):
     wait(m, m.add_host(HostSpec(hostname=name)))
     yield
     wait(m, m.remove_host(name))
+
+
+def assert_rm_service(cephadm, srv_name):
+    assert wait(cephadm, cephadm.remove_service(srv_name)) == f'Removed service {srv_name}'
+    cephadm._apply_all_services()
+
+
+@contextmanager
+def with_service(cephadm_module: CephadmOrchestrator, spec: ServiceSpec, meth, host: str):
+    if spec.placement.is_empty():
+        spec.placement = PlacementSpec(hosts=[host], count=1)
+    c = meth(cephadm_module, spec)
+    assert wait(cephadm_module, c) == f'Scheduled {spec.service_name()} update...'
+    specs = [d.spec for d in wait(cephadm_module, cephadm_module.describe_service())]
+    assert spec in specs
+
+    cephadm_module._apply_all_services()
+
+    dds = wait(cephadm_module, cephadm_module.list_daemons())
+    names = {dd.service_name() for dd in dds}
+    assert spec.service_name() in names, dds
+
+    yield
+
+    assert_rm_service(cephadm_module, spec.service_name())
\ No newline at end of file

diff --git a/src/pybind/mgr/cephadm/tests/test_cephadm.py b/src/pybind/mgr/cephadm/tests/test_cephadm.py
index d4fff4604322016ed947ffaab78ca6eb90e29a53..1e9633173a6a23d1c81bef451f744eaed291f53f 100644 (file)
--- a/src/pybind/mgr/cephadm/tests/test_cephadm.py
+++ b/src/pybind/mgr/cephadm/tests/test_cephadm.py
@@ -23,7 +23,7 @@ from orchestrator import ServiceDescription, DaemonDescription, InventoryHost, \
     HostSpec, OrchestratorError
 from tests import mock
 from .fixtures import cephadm_module, wait, _run_cephadm, match_glob, with_host, \
-    with_cephadm_module
+    with_cephadm_module, with_service, assert_rm_service
 from cephadm.module import CephadmOrchestrator, CEPH_DATEFMT
 
 """
@@ -34,11 +34,6 @@ TODOs:
 """
 
 
-def assert_rm_service(cephadm, srv_name):
-    assert wait(cephadm, cephadm.remove_service(srv_name)) == f'Removed service {srv_name}'
-    cephadm._apply_all_services()
-
-
 def assert_rm_daemon(cephadm: CephadmOrchestrator, prefix, host):
     dds: List[DaemonDescription] = wait(cephadm, cephadm.list_daemons(host=host))
     d_names = [dd.name() for dd in dds if dd.name().startswith(prefix)]
@@ -66,26 +61,6 @@ def with_daemon(cephadm_module: CephadmOrchestrator, spec: ServiceSpec, meth, ho
     assert False, 'Daemon not found'
 
 
-@contextmanager
-def with_service(cephadm_module: CephadmOrchestrator, spec: ServiceSpec, meth, host: str):
-    if spec.placement.is_empty():
-        spec.placement = PlacementSpec(hosts=[host], count=1)
-    c = meth(cephadm_module, spec)
-    assert wait(cephadm_module, c) == f'Scheduled {spec.service_name()} update...'
-    specs = [d.spec for d in wait(cephadm_module, cephadm_module.describe_service())]
-    assert spec in specs
-
-    cephadm_module._apply_all_services()
-
-    dds = wait(cephadm_module, cephadm_module.list_daemons())
-    names = {dd.service_name() for dd in dds}
-    assert spec.service_name() in names
-
-    yield
-
-    assert_rm_service(cephadm_module, spec.service_name())
-
-
 class TestCephadm(object):
 
     def test_get_unique_name(self, cephadm_module):