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

   1 from unittest.mock import MagicMock
   2 from cephadm.agent import Root
   3
   4
   5 class FakeDaemonDescription:
   6     def __init__(self, ip, ports, hostname, service_name='', daemon_type=''):
   7         self.ip = ip
   8         self.ports = ports
   9         self.hostname = hostname
  10         self._service_name = service_name
  11         self.daemon_type = daemon_type
  12
  13     def service_name(self):
  14         return self._service_name
  15
  16
  17 class FakeCache:
  18     def get_daemons_by_service(self, service_type):
  19         return [FakeDaemonDescription('1.2.3.4', [9100], 'node0'),
  20                 FakeDaemonDescription('1.2.3.5', [9200], 'node1')]
  21
  22     def get_daemons_by_type(self, daemon_type):
  23         return [FakeDaemonDescription('1.2.3.4', [9100], 'node0', 'ingress', 'haproxy'),
  24                 FakeDaemonDescription('1.2.3.5', [9200], 'node1', 'ingress', 'haproxy')]
  25
  26
  27 class FakeInventory:
  28     def get_addr(self, name: str):
  29         return '1.2.3.4'
  30
  31
  32 class FakeServiceSpec:
  33     def __init__(self, port):
  34         self.monitor_port = port
  35
  36
  37 class FakeSpecDescription:
  38     def __init__(self, port):
  39         self.spec = FakeServiceSpec(port)
  40
  41
  42 class FakeSpecStore():
  43     def __init__(self, mgr):
  44         self.mgr = mgr
  45         self._specs = {'ingress': FakeSpecDescription(9049)}
  46
  47     def __contains__(self, name):
  48         return name in self._specs
  49
  50     def __getitem__(self, name):
  51         return self._specs['ingress']
  52
  53
  54 class FakeMgr:
  55     def __init__(self):
  56         self.config = ''
  57         self.check_mon_command = MagicMock(side_effect=self._check_mon_command)
  58         self.mon_command = MagicMock(side_effect=self._check_mon_command)
  59         self.template = MagicMock()
  60         self.log = MagicMock()
  61         self.inventory = FakeInventory()
  62         self.cache = FakeCache()
  63         self.spec_store = FakeSpecStore(self)
  64
  65     def list_servers(self):
  66
  67         servers = [
  68             {'hostname': 'node0',
  69              'ceph_version': '16.2',
  70              'services': [{'type': 'mgr'}, {'type': 'mon'}]},
  71             {'hostname': 'node1',
  72              'ceph_version': '16.2',
  73              'services': [{'type': 'mgr'}, {'type': 'mon'}]}
  74         ]
  75
  76         return servers
  77
  78     def _check_mon_command(self, cmd_dict, inbuf=None):
  79         prefix = cmd_dict.get('prefix')
  80         if prefix == 'get-cmd':
  81             return 0, self.config, ''
  82         if prefix == 'set-cmd':
  83             self.config = cmd_dict.get('value')
  84             return 0, 'value set', ''
  85         return -1, '', 'error'
  86
  87     def get_module_option_ex(self, module, option, default_value):
  88         return "9283"
  89
  90
  91 class TestCephadmService:
  92
  93     def test_get_sd_config_prometheus(self):
  94         mgr = FakeMgr()
  95         root = Root(mgr)
  96         cfg = root.get_sd_config('mgr-prometheus')
  97
  98         # check response structure
  99         assert cfg
 100         for entry in cfg:
 101             assert 'labels' in entry
 102             assert 'targets' in entry
 103
 104         # check content
 105         assert cfg[0]['targets'] == ['node0:9283', 'node1:9283']
 106
 107     def test_get_sd_config_node_exporter(self):
 108         mgr = FakeMgr()
 109         root = Root(mgr)
 110         cfg = root.get_sd_config('node-exporter')
 111
 112         # check response structure
 113         assert cfg
 114         for entry in cfg:
 115             assert 'labels' in entry
 116             assert 'targets' in entry
 117
 118         # check content
 119         assert cfg[0]['targets'] == ['1.2.3.4:9100']
 120         assert cfg[0]['labels'] == {'instance': 'node0'}
 121         assert cfg[1]['targets'] == ['1.2.3.5:9200']
 122         assert cfg[1]['labels'] == {'instance': 'node1'}
 123
 124     def test_get_sd_config_alertmgr(self):
 125         mgr = FakeMgr()
 126         root = Root(mgr)
 127         cfg = root.get_sd_config('alertmanager')
 128
 129         # check response structure
 130         assert cfg
 131         for entry in cfg:
 132             assert 'labels' in entry
 133             assert 'targets' in entry
 134
 135         # check content
 136         assert cfg[0]['targets'] == ['1.2.3.4:9100', '1.2.3.5:9200']
 137
 138     def test_get_sd_config_haproxy(self):
 139         mgr = FakeMgr()
 140         root = Root(mgr)
 141         cfg = root.get_sd_config('haproxy')
 142
 143         # check response structure
 144         assert cfg
 145         for entry in cfg:
 146             assert 'labels' in entry
 147             assert 'targets' in entry
 148
 149         # check content
 150         assert cfg[0]['targets'] == ['1.2.3.4:9049']
 151         assert cfg[0]['labels'] == {'instance': 'ingress'}
 152
 153     def test_get_sd_config_invalid_service(self):
 154         mgr = FakeMgr()
 155         root = Root(mgr)
 156         cfg = root.get_sd_config('invalid-service')
 157         assert cfg == []