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

   1 import sys
   2 import time
   3
   4
   5 try:
   6     from typing import Any
   7 except ImportError:
   8     pass
   9
  10 import pytest
  11
  12
  13 from tests import mock
  14 from .fixtures import cephadm_module, wait
  15 from ..module import trivial_completion, forall_hosts
  16
  17
  18 class TestCompletion(object):
  19
  20     def test_trivial(self, cephadm_module):
  21         @trivial_completion
  22         def run(x):
  23             return x+1
  24         assert wait(cephadm_module, run(1)) == 2
  25
  26     def test_exception(self, cephadm_module):
  27         @trivial_completion
  28         def run(x):
  29             raise ValueError
  30         c = run(1)
  31         with pytest.raises(ValueError):
  32             wait(cephadm_module, c)
  33
  34     @pytest.mark.parametrize("input,expected", [
  35         ([], []),
  36         ([1], ["(1,)"]),
  37         (["hallo"], ["('hallo',)"]),
  38         ("hi", ["('h',)", "('i',)"]),
  39         (list(range(5)), [str((x, )) for x in range(5)]),
  40         ([(1, 2), (3, 4)], ["(1, 2)", "(3, 4)"]),
  41     ])
  42     def test_async_map(self, input, expected, cephadm_module):
  43         @forall_hosts
  44         def run_forall(*args):
  45             return str(args)
  46         assert run_forall(input) == expected
  47
  48
  49     @pytest.mark.parametrize("input,expected", [
  50         ([], []),
  51         ([1], ["(1,)"]),
  52         (["hallo"], ["('hallo',)"]),
  53         ("hi", ["('h',)", "('i',)"]),
  54         (list(range(5)), [str((x, )) for x in range(5)]),
  55         ([(1, 2), (3, 4)], ["(1, 2)", "(3, 4)"]),
  56     ])
  57     def test_async_map_self(self, input, expected, cephadm_module):
  58         class Run(object):
  59             def __init__(self):
  60                 self.attr = 1
  61
  62             @forall_hosts
  63             def run_forall(self, *args):
  64                 assert self.attr == 1
  65                 return str(args)
  66
  67         assert Run().run_forall(input) == expected