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, async_completion, async_map_completion
  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     @pytest.mark.parametrize("input", [
  27         ((1, ), ),
  28         ((1, 2), ),
  29         (("hallo", ), ),
  30         (("hallo", "foo"), ),
  31     ])
  32     def test_async(self, input, cephadm_module):
  33         @async_completion
  34         def run(*args):
  35             return str(args)
  36
  37         assert wait(cephadm_module, run(*input)) == str(input)
  38
  39     @pytest.mark.parametrize("input,expected", [
  40         ([], []),
  41         ([1], ["(1,)"]),
  42         (["hallo"], ["('hallo',)"]),
  43         ("hi", ["('h',)", "('i',)"]),
  44         (list(range(5)), [str((x, )) for x in range(5)]),
  45         ([(1, 2), (3, 4)], ["(1, 2)", "(3, 4)"]),
  46     ])
  47     def test_async_map(self, input, expected, cephadm_module):
  48         @async_map_completion
  49         def run(*args):
  50             return str(args)
  51
  52         c = run(input)
  53         wait(cephadm_module, c)
  54         assert c.result == expected
  55
  56     def test_async_self(self, cephadm_module):
  57         class Run(object):
  58             def __init__(self):
  59                 self.attr = 1
  60
  61             @async_completion
  62             def run(self, x):
  63                 assert self.attr == 1
  64                 return x + 1
  65
  66         assert wait(cephadm_module, Run().run(1)) == 2
  67
  68     @pytest.mark.parametrize("input,expected", [
  69         ([], []),
  70         ([1], ["(1,)"]),
  71         (["hallo"], ["('hallo',)"]),
  72         ("hi", ["('h',)", "('i',)"]),
  73         (list(range(5)), [str((x, )) for x in range(5)]),
  74         ([(1, 2), (3, 4)], ["(1, 2)", "(3, 4)"]),
  75     ])
  76     def test_async_map_self(self, input, expected, cephadm_module):
  77         class Run(object):
  78             def __init__(self):
  79                 self.attr = 1
  80
  81             @async_map_completion
  82             def run(self, *args):
  83                 assert self.attr == 1
  84                 return str(args)
  85
  86         c = Run().run(input)
  87         wait(cephadm_module, c)
  88         assert c.result == expected
  89
  90     def test_then1(self, cephadm_module):
  91         @async_map_completion
  92         def run(x):
  93             return x+1
  94
  95         assert wait(cephadm_module, run([1,2]).then(str)) == '[2, 3]'
  96
  97     def test_then2(self, cephadm_module):
  98         @async_map_completion
  99         def run(x):
 100             time.sleep(0.1)
 101             return x+1
 102
 103         @async_completion
 104         def async_str(results):
 105             return str(results)
 106
 107         c = run([1,2]).then(async_str)
 108
 109         wait(cephadm_module, c)
 110         assert c.result == '[2, 3]'
 111
 112     def test_then3(self, cephadm_module):
 113         @async_map_completion
 114         def run(x):
 115             time.sleep(0.1)
 116             return x+1
 117
 118         def async_str(results):
 119             return async_completion(str)(results)
 120
 121         c = run([1,2]).then(async_str)
 122
 123         wait(cephadm_module, c)
 124         assert c.result == '[2, 3]'
 125
 126     def test_then4(self, cephadm_module):
 127         @async_map_completion
 128         def run(x):
 129             time.sleep(0.1)
 130             return x+1
 131
 132         def async_str(results):
 133             return async_completion(str)(results).then(lambda x: x + "hello")
 134
 135         c = run([1,2]).then(async_str)
 136
 137         wait(cephadm_module, c)
 138         assert c.result == '[2, 3]hello'
 139
 140     @pytest.mark.skip(reason="see limitation of async_map_completion")
 141     def test_then5(self, cephadm_module):
 142         @async_map_completion
 143         def run(x):
 144             time.sleep(0.1)
 145             return async_completion(str)(x+1)
 146
 147         c = run([1,2])
 148
 149         wait(cephadm_module, c)
 150         assert c.result == "['2', '3']"
 151
 152     def test_raise(self, cephadm_module):
 153         @async_completion
 154         def run(x):
 155             raise ZeroDivisionError()
 156
 157         with pytest.raises(ZeroDivisionError):
 158             wait(cephadm_module, run(1))
 159
 160     def test_progress(self, cephadm_module):
 161         @async_map_completion
 162         def run(*args):
 163             return str(args)
 164
 165         c = run(list(range(2)))
 166         c.update_progress = True
 167         c.add_progress(
 168             mgr=cephadm_module,
 169             message="my progress"
 170         )
 171         wait(cephadm_module, c)
 172         assert c.result == [str((x,)) for x in range(2)]
 173         assert cephadm_module.remote.mock_calls == [
 174             mock.call('progress', 'update', mock.ANY, 'my progress', float(i) / 2, [('origin', 'orchestrator')])
 175             for i in range(2+1)] + [
 176             mock.call('progress', 'update', mock.ANY, 'my progress', 1.0, [('origin', 'orchestrator')]),
 177             mock.call('progress', 'complete', mock.ANY),
 178         ]