ceph/src/pybind/mgr/cephadm/tests/test_completion.py

   1 import pytest
   2
   3 from ..module import forall_hosts
   4
   5
   6 class TestCompletion(object):
   7
   8     @pytest.mark.parametrize("input,expected", [
   9         ([], []),
  10         ([1], ["(1,)"]),
  11         (["hallo"], ["('hallo',)"]),
  12         ("hi", ["('h',)", "('i',)"]),
  13         (list(range(5)), [str((x, )) for x in range(5)]),
  14         ([(1, 2), (3, 4)], ["(1, 2)", "(3, 4)"]),
  15     ])
  16     def test_async_map(self, input, expected, cephadm_module):
  17         @forall_hosts
  18         def run_forall(*args):
  19             return str(args)
  20         assert run_forall(input) == expected
  21
  22     @pytest.mark.parametrize("input,expected", [
  23         ([], []),
  24         ([1], ["(1,)"]),
  25         (["hallo"], ["('hallo',)"]),
  26         ("hi", ["('h',)", "('i',)"]),
  27         (list(range(5)), [str((x, )) for x in range(5)]),
  28         ([(1, 2), (3, 4)], ["(1, 2)", "(3, 4)"]),
  29     ])
  30     def test_async_map_self(self, input, expected, cephadm_module):
  31         class Run(object):
  32             def __init__(self):
  33                 self.attr = 1
  34
  35             @forall_hosts
  36             def run_forall(self, *args):
  37                 assert self.attr == 1
  38                 return str(args)
  39
  40         assert Run().run_forall(input) == expected