diff options
Diffstat (limited to 'src/pybind/mgr/cephadm/tests/test_node_proxy.py')
| -rw-r--r-- | src/pybind/mgr/cephadm/tests/test_node_proxy.py | 312 |
1 files changed, 312 insertions, 0 deletions
diff --git a/src/pybind/mgr/cephadm/tests/test_node_proxy.py b/src/pybind/mgr/cephadm/tests/test_node_proxy.py new file mode 100644 index 000000000..b19bb5dbc --- /dev/null +++ b/src/pybind/mgr/cephadm/tests/test_node_proxy.py @@ -0,0 +1,312 @@ +import cherrypy +import json +from _pytest.monkeypatch import MonkeyPatch +from urllib.error import URLError +from cherrypy.test import helper +from cephadm.agent import NodeProxyEndpoint +from unittest.mock import MagicMock, call, patch +from cephadm.inventory import AgentCache, NodeProxyCache, Inventory +from cephadm.ssl_cert_utils import SSLCerts +from . import node_proxy_data + +PORT = 58585 + + +class FakeMgr: + def __init__(self) -> None: + self.log = MagicMock() + self.get_store = MagicMock(return_value=json.dumps(node_proxy_data.mgr_inventory_cache)) + self.set_store = MagicMock() + self.set_health_warning = MagicMock() + self.remove_health_warning = MagicMock() + self.inventory = Inventory(self) + self.agent_cache = AgentCache(self) + self.agent_cache.agent_ports = {"host01": 1234} + self.node_proxy_cache = NodeProxyCache(self) + self.node_proxy_cache.save = MagicMock() + self.node_proxy = MagicMock() + self.http_server = MagicMock() + self.http_server.agent = MagicMock() + self.http_server.agent.ssl_certs = SSLCerts() + self.http_server.agent.ssl_certs.generate_root_cert(self.get_mgr_ip()) + + def get_mgr_ip(self) -> str: + return '0.0.0.0' + + +class TestNodeProxyEndpoint(helper.CPWebCase): + mgr = FakeMgr() + app = NodeProxyEndpoint(mgr) + mgr.node_proxy_cache.keyrings = {"host01": "fake-secret01", + "host02": "fake-secret02"} + mgr.node_proxy_cache.oob = {"host01": {"username": "oob-user01", + "password": "oob-pass01"}, + "host02": {"username": "oob-user02", + "password": "oob-pass02"}} + mgr.node_proxy_cache.data = node_proxy_data.full_set + + @classmethod + def setup_server(cls): + # cherrypy.tree.mount(NodeProxyEndpoint(TestNodeProxyEndpoint.mgr)) + cherrypy.tree.mount(TestNodeProxyEndpoint.app) + cherrypy.config.update({'global': { + 'server.socket_host': '127.0.0.1', + 'server.socket_port': PORT}}) + + def setUp(self): + self.PORT = PORT + self.monkeypatch = MonkeyPatch() + + def test_oob_data_misses_cephx_field(self): + data = '{}' + self.getPage("/oob", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('400 Bad Request') + + def test_oob_data_misses_name_field(self): + data = '{"cephx": {"secret": "fake-secret"}}' + self.getPage("/oob", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('400 Bad Request') + + def test_oob_data_misses_secret_field(self): + data = '{"cephx": {"name": "node-proxy.host01"}}' + self.getPage("/oob", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('400 Bad Request') + + def test_oob_agent_not_running(self): + data = '{"cephx": {"name": "node-proxy.host03", "secret": "fake-secret03"}}' + self.getPage("/oob", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('502 Bad Gateway') + + def test_oob_wrong_keyring(self): + data = '{"cephx": {"name": "node-proxy.host01", "secret": "wrong-keyring"}}' + self.getPage("/oob", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('403 Forbidden') + + def test_oob_ok(self): + data = '{"cephx": {"name": "node-proxy.host01", "secret": "fake-secret01"}}' + self.getPage("/oob", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('200 OK') + + def test_data_missing_patch(self): + data = '{"cephx": {"name": "node-proxy.host01", "secret": "fake-secret01"}}' + self.getPage("/data", method="POST", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('400 Bad Request') + + def test_data_raises_alert(self): + patch = node_proxy_data.full_set_with_critical + data = {"cephx": {"name": "node-proxy.host01", "secret": "fake-secret01"}, "patch": patch} + data_str = json.dumps(data) + self.getPage("/data", method="POST", body=data_str, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data_str)))]) + self.assertStatus('200 OK') + + calls = [call('HARDWARE_STORAGE', + count=2, + detail=['disk.bay.0:enclosure.internal.0-1:raid.integrated.1-1 is critical: Enabled', + 'disk.bay.9:enclosure.internal.0-1 is critical: Enabled'], + summary='2 storage members are not ok'), + call('HARDWARE_MEMORY', + count=1, + detail=['dimm.socket.a1 is critical: Enabled'], + summary='1 memory member is not ok')] + + assert TestNodeProxyEndpoint.mgr.set_health_warning.mock_calls == calls + + def test_led_GET_no_hostname(self): + self.getPage("/led", method="GET") + self.assertStatus('501 Not Implemented') + + def test_led_PATCH_no_hostname(self): + data = "{}" + self.getPage("/led", method="PATCH", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('501 Not Implemented') + + def test_set_led_no_type(self): + data = '{"state": "on", "keyring": "fake-secret01"}' + self.getPage("/host01/led", method="PATCH", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('400 Bad Request') + + def test_set_chassis_led(self): + data = '{"state": "on", "keyring": "fake-secret01"}' + with patch('cephadm.agent.http_req') as p: + p.return_value = [], '{}', 200 + self.getPage("/host01/led/chassis", method="PATCH", body=data, headers=[('Content-Type', 'application/json'), + ('Content-Length', str(len(data)))]) + self.assertStatus('200 OK') + + def test_get_led_missing_type(self): + self.getPage("/host01/led", method="GET") + self.assertStatus('400 Bad Request') + + def test_get_led_no_hostname(self): + self.getPage("/led", method="GET") + self.assertStatus('501 Not Implemented') + + def test_get_led_type_chassis_no_hostname(self): + self.getPage("/led/chassis", method="GET") + self.assertStatus('404 Not Found') + + def test_get_led_type_drive_no_hostname(self): + self.getPage("/led/chassis", method="GET") + self.assertStatus('404 Not Found') + + def test_get_led_type_drive_missing_id(self): + self.getPage("/host01/led/drive", method="GET") + self.assertStatus('400 Bad Request') + + def test_get_led_url_error(self): + with patch('cephadm.agent.http_req') as p: + p.side_effect = URLError('fake error') + self.getPage("/host02/led/chassis", method="GET") + self.assertStatus('502 Bad Gateway') + + def test_get_chassis_led_ok(self): + with patch('cephadm.agent.http_req', return_value=MagicMock()) as p: + p.return_value = [], '{}', 200 + self.getPage("/host01/led/chassis", method="GET") + self.assertStatus('200 OK') + + def test_get_drive_led_without_id(self): + self.getPage("/host01/led/drive", method="GET") + self.assertStatus('400 Bad Request') + + def test_get_drive_led_with_id(self): + with patch('cephadm.agent.http_req', return_value=MagicMock()) as p: + p.return_value = [], '{}', 200 + self.getPage("/host01/led/drive/123", method="GET") + self.assertStatus('200 OK') + + def test_fullreport_with_valid_hostname(self): + # data = '{"cephx": {"name": "node-proxy.host01", "secret": "fake-secret01"}}' + # self.getPage("/host02/fullreport", method="POST", body=data, headers=[('Content-Type', 'application/json'), ('Content-Length', str(len(data)))]) + self.getPage("/host02/fullreport", method="GET") + self.assertStatus('200 OK') + + def test_fullreport_no_hostname(self): + # data = '{"cephx": {"name": "node-proxy.host01", "secret": "fake-secret01"}}' + # self.getPage("/fullreport", method="POST", body=data, headers=[('Content-Type', 'application/json'), ('Content-Length', str(len(data)))]) + self.getPage("/fullreport", method="GET") + self.assertStatus('200 OK') + + def test_fullreport_with_invalid_hostname(self): + # data = '{"cephx": {"name": "node-proxy.host03", "secret": "fake-secret03"}}' + # self.getPage("/host03/fullreport", method="POST", body=data, headers=[('Content-Type', 'application/json'), ('Content-Length', str(len(data)))]) + self.getPage("/host03/fullreport", method="GET") + self.assertStatus('404 Not Found') + + def test_summary_with_valid_hostname(self): + self.getPage("/host02/summary", method="GET") + self.assertStatus('200 OK') + + def test_summary_no_hostname(self): + self.getPage("/summary", method="GET") + self.assertStatus('200 OK') + + def test_summary_with_invalid_hostname(self): + self.getPage("/host03/summary", method="GET") + self.assertStatus('404 Not Found') + + def test_criticals_with_valid_hostname(self): + self.getPage("/host02/criticals", method="GET") + self.assertStatus('200 OK') + + def test_criticals_no_hostname(self): + self.getPage("/criticals", method="GET") + self.assertStatus('200 OK') + + def test_criticals_with_invalid_hostname(self): + self.getPage("/host03/criticals", method="GET") + self.assertStatus('404 Not Found') + + def test_memory_with_valid_hostname(self): + self.getPage("/host02/memory", method="GET") + self.assertStatus('200 OK') + + def test_memory_no_hostname(self): + self.getPage("/memory", method="GET") + self.assertStatus('200 OK') + + def test_memory_with_invalid_hostname(self): + self.getPage("/host03/memory", method="GET") + self.assertStatus('404 Not Found') + + def test_network_with_valid_hostname(self): + self.getPage("/host02/network", method="GET") + self.assertStatus('200 OK') + + def test_network_no_hostname(self): + self.getPage("/network", method="GET") + self.assertStatus('200 OK') + + def test_network_with_invalid_hostname(self): + self.getPage("/host03/network", method="GET") + self.assertStatus('404 Not Found') + + def test_processors_with_valid_hostname(self): + self.getPage("/host02/processors", method="GET") + self.assertStatus('200 OK') + + def test_processors_no_hostname(self): + self.getPage("/processors", method="GET") + self.assertStatus('200 OK') + + def test_processors_with_invalid_hostname(self): + self.getPage("/host03/processors", method="GET") + self.assertStatus('404 Not Found') + + def test_storage_with_valid_hostname(self): + self.getPage("/host02/storage", method="GET") + self.assertStatus('200 OK') + + def test_storage_no_hostname(self): + self.getPage("/storage", method="GET") + self.assertStatus('200 OK') + + def test_storage_with_invalid_hostname(self): + self.getPage("/host03/storage", method="GET") + self.assertStatus('404 Not Found') + + def test_power_with_valid_hostname(self): + self.getPage("/host02/power", method="GET") + self.assertStatus('200 OK') + + def test_power_no_hostname(self): + self.getPage("/power", method="GET") + self.assertStatus('200 OK') + + def test_power_with_invalid_hostname(self): + self.getPage("/host03/power", method="GET") + self.assertStatus('404 Not Found') + + def test_fans_with_valid_hostname(self): + self.getPage("/host02/fans", method="GET") + self.assertStatus('200 OK') + + def test_fans_no_hostname(self): + self.getPage("/fans", method="GET") + self.assertStatus('200 OK') + + def test_fans_with_invalid_hostname(self): + self.getPage("/host03/fans", method="GET") + self.assertStatus('404 Not Found') + + def test_firmwares_with_valid_hostname(self): + self.getPage("/host02/firmwares", method="GET") + self.assertStatus('200 OK') + + def test_firmwares_no_hostname(self): + self.getPage("/firmwares", method="GET") + self.assertStatus('200 OK') + + def test_firmwares_with_invalid_hostname(self): + self.getPage("/host03/firmwares", method="GET") + self.assertStatus('404 Not Found') |