)
enabled, state, installed = _cephadm.check_unit(ctx, "foobar")
assert (enabled, state, installed) == expected
+
+
+class FakeEnabler:
+ def __init__(self, should_be_called):
+ self._should_be_called = should_be_called
+ self._services = []
+
+ def enable_service(self, service):
+ self._services.append(service)
+
+ def check_expected(self):
+ if not self._should_be_called:
+ assert not self._services
+ return
+ # there are currently seven chron/chrony type services that
+ # cephadm looks for. Make sure it probed for each of them
+ # or more in case someone adds to the list.
+ assert len(self._services) >= 7
+ assert "chrony.service" in self._services
+ assert "ntp.service" in self._services
+
+
+@pytest.mark.parametrize(
+ "call_fn, enabler, expected",
+ [
+ # Test that time sync services are not enabled
+ (
+ _mk_fake_call(
+ enabled=("", "", 1),
+ active=("", "", 1),
+ ),
+ None,
+ False,
+ ),
+ # Test that time sync service is enabled
+ (
+ _mk_fake_call(
+ enabled=("", "", 0),
+ active=("active", "", 0),
+ ),
+ None,
+ True,
+ ),
+ # Test that time sync is not enabled, and try to enable them.
+ # This one needs to be not running, but installed in order to
+ # call the enabler. It should call the enabler with every known
+ # service name.
+ (
+ _mk_fake_call(
+ enabled=("disabled", "", 1),
+ active=("", "", 1),
+ ),
+ FakeEnabler(True),
+ False,
+ ),
+ # Test that time sync is enabled, with an enabler passed which
+ # will check that the enabler was never called.
+ (
+ _mk_fake_call(
+ enabled=("", "", 0),
+ active=("active", "", 0),
+ ),
+ FakeEnabler(False),
+ True,
+ ),
+ ],
+)
+def test_check_time_sync(call_fn, enabler, expected):
+ """The check_time_sync call actually checks if a time synchronization service
+ is enabled. It is also the only consumer of check_units.
+ """
+ with with_cephadm_ctx([]) as ctx:
+ _cephadm.call.side_effect = call_fn
+ result = _cephadm.check_time_sync(ctx, enabler=enabler)
+ assert result == expected
+ if enabler is not None:
+ enabler.check_expected()