From 2ffa81bb91618eb70708073096f39bc1f8e2a8e6 Mon Sep 17 00:00:00 2001 From: Paul Cuzner Date: Fri, 12 Nov 2021 16:19:00 +1300 Subject: [PATCH] mgr/cephadm: Add unit tests for snmp-gateway support Adds tests to validate the deployed configuration given a known input context, and check the parameters created based on input various input scenarios Signed-off-by: Paul Cuzner --- src/cephadm/tests/test_cephadm.py | 156 ++++++++- src/pybind/mgr/cephadm/tests/test_services.py | 157 ++++++++- src/pybind/mgr/cephadm/tests/test_spec.py | 299 +++++++++++++++++- src/python-common/ceph/tests/test_utils.py | 40 ++- 4 files changed, 644 insertions(+), 8 deletions(-) diff --git a/src/cephadm/tests/test_cephadm.py b/src/cephadm/tests/test_cephadm.py index df0f932248c..a087553617e 100644 --- a/src/cephadm/tests/test_cephadm.py +++ b/src/cephadm/tests/test_cephadm.py @@ -630,7 +630,7 @@ class TestMaintenance: fsid = '0ea8cdd0-1bbf-11ec-a9c7-5254002763fa' def test_systemd_target_OK(self, tmp_path): - base = tmp_path + base = tmp_path wants = base / "ceph.target.wants" wants.mkdir() target = wants / TestMaintenance.systemd_target @@ -641,7 +641,7 @@ class TestMaintenance: assert cd.systemd_target_state(ctx, target.name) def test_systemd_target_NOTOK(self, tmp_path): - base = tmp_path + base = tmp_path ctx = cd.CephadmContext() ctx.unit_dir = str(base) assert not cd.systemd_target_state(ctx, TestMaintenance.systemd_target) @@ -1392,7 +1392,7 @@ class TestPull: class TestApplySpec: - + def test_parse_yaml(self, cephadm_fs): yaml = '''service_type: host hostname: vm-00 @@ -1416,7 +1416,7 @@ addr: 192.168.122.165''' retdic = [{'service_type': 'host', 'hostname': 'vm-00', 'addr': '192.168.122.44', 'labels': '- example1- example2'}, {'service_type': 'host', 'hostname': 'vm-01', 'addr': '192.168.122.247', 'labels': '- grafana'}, {'service_type': 'host', 'hostname': 'vm-02', 'addr': '192.168.122.165'}] - + with open('spec.yml') as f: dic = cd.parse_yaml_objs(f) assert dic == retdic @@ -1440,11 +1440,157 @@ addr: 192.168.122.165''' assert retval == 1 +class TestSNMPGateway: + V2c_config = { + 'snmp_community': 'public', + 'destination': '192.168.1.10:162', + 'snmp_version': 'V2c', + } + V3_no_priv_config = { + 'destination': '192.168.1.10:162', + 'snmp_version': 'V3', + 'snmp_v3_auth_username': 'myuser', + 'snmp_v3_auth_password': 'mypassword', + 'snmp_v3_auth_protocol': 'SHA', + 'snmp_v3_engine_id': '8000C53F00000000', + } + V3_priv_config = { + 'destination': '192.168.1.10:162', + 'snmp_version': 'V3', + 'snmp_v3_auth_username': 'myuser', + 'snmp_v3_auth_password': 'mypassword', + 'snmp_v3_auth_protocol': 'SHA', + 'snmp_v3_priv_protocol': 'DES', + 'snmp_v3_priv_password': 'mysecret', + 'snmp_v3_engine_id': '8000C53F00000000', + } + no_destination_config = { + 'snmp_version': 'V3', + 'snmp_v3_auth_username': 'myuser', + 'snmp_v3_auth_password': 'mypassword', + 'snmp_v3_auth_protocol': 'SHA', + 'snmp_v3_priv_protocol': 'DES', + 'snmp_v3_priv_password': 'mysecret', + 'snmp_v3_engine_id': '8000C53F00000000', + } + bad_version_config = { + 'snmp_community': 'public', + 'destination': '192.168.1.10:162', + 'snmp_version': 'V1', + } + + def test_unit_run_V2c(self, cephadm_fs): + fsid = 'ca734440-3dc6-11ec-9b98-5254002537a6' + with with_cephadm_ctx(['--image=docker.io/maxwo/snmp-notifier:v1.2.1'], list_networks={}) as ctx: + import json + ctx.config_json = json.dumps(self.V2c_config) + ctx.fsid = fsid + ctx.tcp_ports = '9464' + cd.get_parm.return_value = self.V2c_config + c = cd.get_container(ctx, fsid, 'snmp-gateway', 'daemon_id') + + cd.make_data_dir(ctx, fsid, 'snmp-gateway', 'daemon_id') + + cd.create_daemon_dirs(ctx, fsid, 'snmp-gateway', 'daemon_id', 0, 0) + with open(f'/var/lib/ceph/{fsid}/snmp-gateway.daemon_id/snmp-gateway.conf', 'r') as f: + conf = f.read().rstrip() + assert conf == 'SNMP_NOTIFIER_COMMUNITY=public' + + cd.deploy_daemon_units( + ctx, + fsid, + 0, 0, + 'snmp-gateway', + 'daemon_id', + c, + True, True + ) + with open(f'/var/lib/ceph/{fsid}/snmp-gateway.daemon_id/unit.run', 'r') as f: + run_cmd = f.readlines()[-1].rstrip() + assert run_cmd.endswith('docker.io/maxwo/snmp-notifier:v1.2.1 --web.listen-address=:9464 --snmp.destination=192.168.1.10:162 --snmp.version=V2c --log.level=info --snmp.trap-description-template=/etc/snmp_notifier/description-template.tpl') + + def test_unit_run_V3_noPriv(self, cephadm_fs): + fsid = 'ca734440-3dc6-11ec-9b98-5254002537a6' + with with_cephadm_ctx(['--image=docker.io/maxwo/snmp-notifier:v1.2.1'], list_networks={}) as ctx: + import json + ctx.config_json = json.dumps(self.V3_no_priv_config) + ctx.fsid = fsid + ctx.tcp_ports = '9465' + cd.get_parm.return_value = self.V3_no_priv_config + c = cd.get_container(ctx, fsid, 'snmp-gateway', 'daemon_id') + + cd.make_data_dir(ctx, fsid, 'snmp-gateway', 'daemon_id') + + cd.create_daemon_dirs(ctx, fsid, 'snmp-gateway', 'daemon_id', 0, 0) + with open(f'/var/lib/ceph/{fsid}/snmp-gateway.daemon_id/snmp-gateway.conf', 'r') as f: + conf = f.read() + assert conf == 'SNMP_NOTIFIER_AUTH_USERNAME=myuser\nSNMP_NOTIFIER_AUTH_PASSWORD=mypassword\n' + + cd.deploy_daemon_units( + ctx, + fsid, + 0, 0, + 'snmp-gateway', + 'daemon_id', + c, + True, True + ) + with open(f'/var/lib/ceph/{fsid}/snmp-gateway.daemon_id/unit.run', 'r') as f: + run_cmd = f.readlines()[-1].rstrip() + assert run_cmd.endswith('docker.io/maxwo/snmp-notifier:v1.2.1 --web.listen-address=:9465 --snmp.destination=192.168.1.10:162 --snmp.version=V3 --log.level=info --snmp.trap-description-template=/etc/snmp_notifier/description-template.tpl --snmp.authentication-enabled --snmp.authentication-protocol=SHA --snmp.security-engine-id=8000C53F00000000') + + def test_unit_run_V3_Priv(self, cephadm_fs): + fsid = 'ca734440-3dc6-11ec-9b98-5254002537a6' + with with_cephadm_ctx(['--image=docker.io/maxwo/snmp-notifier:v1.2.1'], list_networks={}) as ctx: + import json + ctx.config_json = json.dumps(self.V3_priv_config) + ctx.fsid = fsid + ctx.tcp_ports = '9464' + cd.get_parm.return_value = self.V3_priv_config + c = cd.get_container(ctx, fsid, 'snmp-gateway', 'daemon_id') + cd.make_data_dir(ctx, fsid, 'snmp-gateway', 'daemon_id') + cd.create_daemon_dirs(ctx, fsid, 'snmp-gateway', 'daemon_id', 0, 0) + with open(f'/var/lib/ceph/{fsid}/snmp-gateway.daemon_id/snmp-gateway.conf', 'r') as f: + conf = f.read() + assert conf == 'SNMP_NOTIFIER_AUTH_USERNAME=myuser\nSNMP_NOTIFIER_AUTH_PASSWORD=mypassword\nSNMP_NOTIFIER_PRIV_PASSWORD=mysecret\n' + cd.deploy_daemon_units( + ctx, + fsid, + 0, 0, + 'snmp-gateway', + 'daemon_id', + c, + True, True + ) + with open(f'/var/lib/ceph/{fsid}/snmp-gateway.daemon_id/unit.run', 'r') as f: + run_cmd = f.readlines()[-1].rstrip() + assert run_cmd.endswith('docker.io/maxwo/snmp-notifier:v1.2.1 --web.listen-address=:9464 --snmp.destination=192.168.1.10:162 --snmp.version=V3 --log.level=info --snmp.trap-description-template=/etc/snmp_notifier/description-template.tpl --snmp.authentication-enabled --snmp.authentication-protocol=SHA --snmp.security-engine-id=8000C53F00000000 --snmp.private-enabled --snmp.private-protocol=DES') + def test_unit_run_no_dest(self, cephadm_fs): + fsid = 'ca734440-3dc6-11ec-9b98-5254002537a6' + with with_cephadm_ctx(['--image=docker.io/maxwo/snmp-notifier:v1.2.1'], list_networks={}) as ctx: + import json + ctx.config_json = json.dumps(self.no_destination_config) + ctx.fsid = fsid + ctx.tcp_ports = '9464' + cd.get_parm.return_value = self.no_destination_config + with pytest.raises(Exception) as e: + c = cd.get_container(ctx, fsid, 'snmp-gateway', 'daemon_id') + assert str(e.value) == "config is missing destination attribute(:) of the target SNMP listener" + def test_unit_run_bad_version(self, cephadm_fs): + fsid = 'ca734440-3dc6-11ec-9b98-5254002537a6' + with with_cephadm_ctx(['--image=docker.io/maxwo/snmp-notifier:v1.2.1'], list_networks={}) as ctx: + import json + ctx.config_json = json.dumps(self.bad_version_config) + ctx.fsid = fsid + ctx.tcp_ports = '9464' + cd.get_parm.return_value = self.bad_version_config - + with pytest.raises(Exception) as e: + c = cd.get_container(ctx, fsid, 'snmp-gateway', 'daemon_id') + assert str(e.value) == 'not a valid snmp version: V1' diff --git a/src/pybind/mgr/cephadm/tests/test_services.py b/src/pybind/mgr/cephadm/tests/test_services.py index c72bb8aea5d..7ec0b419040 100644 --- a/src/pybind/mgr/cephadm/tests/test_services.py +++ b/src/pybind/mgr/cephadm/tests/test_services.py @@ -16,7 +16,7 @@ from cephadm.services.monitoring import GrafanaService, AlertmanagerService, Pro NodeExporterService from cephadm.module import CephadmOrchestrator from ceph.deployment.service_spec import IscsiServiceSpec, MonitoringSpec, AlertManagerSpec, \ - ServiceSpec, RGWSpec, GrafanaSpec + ServiceSpec, RGWSpec, GrafanaSpec, SNMPGatewaySpec from cephadm.tests.fixtures import with_host, with_service, _run_cephadm, async_side_effect from orchestrator import OrchestratorError @@ -498,3 +498,158 @@ class TestRGWService: 'key': 'rgw_frontends', }) assert f == expected + + +class TestSNMPGateway: + + @patch("cephadm.serve.CephadmServe._run_cephadm") + def test_snmp_v2c_deployment(self, _run_cephadm, cephadm_module: CephadmOrchestrator): + _run_cephadm.side_effect = async_side_effect(('{}', '', 0)) + + spec = SNMPGatewaySpec( + snmp_version='V2c', + snmp_destination='192.168.1.1:162', + credentials={ + 'snmp_community': 'public' + }) + + config = { + "destination": spec.snmp_destination, + "snmp_version": spec.snmp_version, + "snmp_community": spec.credentials.get('snmp_community') + } + + with with_host(cephadm_module, 'test'): + with with_service(cephadm_module, spec): + _run_cephadm.assert_called_with( + 'test', + 'snmp-gateway.test', + 'deploy', + [ + '--name', 'snmp-gateway.test', + '--meta-json', + '{"service_name": "snmp-gateway", "ports": [9464], "ip": null, "deployed_by": [], "rank": null, "rank_generation": null}', + '--config-json', '-', + '--tcp-ports', '9464' + ], + stdin=json.dumps(config), + image='' + ) + + @patch("cephadm.serve.CephadmServe._run_cephadm") + def test_snmp_v2c_with_port(self, _run_cephadm, cephadm_module: CephadmOrchestrator): + _run_cephadm.side_effect = async_side_effect(('{}', '', 0)) + + spec = SNMPGatewaySpec( + snmp_version='V2c', + snmp_destination='192.168.1.1:162', + credentials={ + 'snmp_community': 'public' + }, + port=9465) + + config = { + "destination": spec.snmp_destination, + "snmp_version": spec.snmp_version, + "snmp_community": spec.credentials.get('snmp_community') + } + + with with_host(cephadm_module, 'test'): + with with_service(cephadm_module, spec): + _run_cephadm.assert_called_with( + 'test', + 'snmp-gateway.test', + 'deploy', + [ + '--name', 'snmp-gateway.test', + '--meta-json', + '{"service_name": "snmp-gateway", "ports": [9465], "ip": null, "deployed_by": [], "rank": null, "rank_generation": null}', + '--config-json', '-', + '--tcp-ports', '9465' + ], + stdin=json.dumps(config), + image='' + ) + + @patch("cephadm.serve.CephadmServe._run_cephadm") + def test_snmp_v3nopriv_deployment(self, _run_cephadm, cephadm_module: CephadmOrchestrator): + _run_cephadm.side_effect = async_side_effect(('{}', '', 0)) + + spec = SNMPGatewaySpec( + snmp_version='V3', + snmp_destination='192.168.1.1:162', + engine_id='8000C53F00000000', + credentials={ + 'snmp_v3_auth_username': 'myuser', + 'snmp_v3_auth_password': 'mypassword' + }) + + config = { + 'destination': spec.snmp_destination, + 'snmp_version': spec.snmp_version, + 'snmp_v3_auth_protocol': 'SHA', + 'snmp_v3_auth_username': 'myuser', + 'snmp_v3_auth_password': 'mypassword', + 'snmp_v3_engine_id': '8000C53F00000000' + } + + with with_host(cephadm_module, 'test'): + with with_service(cephadm_module, spec): + _run_cephadm.assert_called_with( + 'test', + 'snmp-gateway.test', + 'deploy', + [ + '--name', 'snmp-gateway.test', + '--meta-json', + '{"service_name": "snmp-gateway", "ports": [9464], "ip": null, "deployed_by": [], "rank": null, "rank_generation": null}', + '--config-json', '-', + '--tcp-ports', '9464' + ], + stdin=json.dumps(config), + image='' + ) + + @patch("cephadm.serve.CephadmServe._run_cephadm") + def test_snmp_v3priv_deployment(self, _run_cephadm, cephadm_module: CephadmOrchestrator): + _run_cephadm.side_effect = async_side_effect(('{}', '', 0)) + + spec = SNMPGatewaySpec( + snmp_version='V3', + snmp_destination='192.168.1.1:162', + engine_id='8000C53F00000000', + auth_protocol='MD5', + privacy_protocol='AES', + credentials={ + 'snmp_v3_auth_username': 'myuser', + 'snmp_v3_auth_password': 'mypassword', + 'snmp_v3_priv_password': 'mysecret', + }) + + config = { + 'destination': spec.snmp_destination, + 'snmp_version': spec.snmp_version, + 'snmp_v3_auth_protocol': 'MD5', + 'snmp_v3_auth_username': spec.credentials.get('snmp_v3_auth_username'), + 'snmp_v3_auth_password': spec.credentials.get('snmp_v3_auth_password'), + 'snmp_v3_engine_id': '8000C53F00000000', + 'snmp_v3_priv_protocol': spec.privacy_protocol, + 'snmp_v3_priv_password': spec.credentials.get('snmp_v3_priv_password'), + } + + with with_host(cephadm_module, 'test'): + with with_service(cephadm_module, spec): + _run_cephadm.assert_called_with( + 'test', + 'snmp-gateway.test', + 'deploy', + [ + '--name', 'snmp-gateway.test', + '--meta-json', + '{"service_name": "snmp-gateway", "ports": [9464], "ip": null, "deployed_by": [], "rank": null, "rank_generation": null}', + '--config-json', '-', + '--tcp-ports', '9464' + ], + stdin=json.dumps(config), + image='' + ) diff --git a/src/pybind/mgr/cephadm/tests/test_spec.py b/src/pybind/mgr/cephadm/tests/test_spec.py index 458958973c8..8697d19533f 100644 --- a/src/pybind/mgr/cephadm/tests/test_spec.py +++ b/src/pybind/mgr/cephadm/tests/test_spec.py @@ -3,12 +3,13 @@ # fmt: off import json +import yaml import pytest from ceph.deployment.service_spec import ServiceSpec, NFSServiceSpec, RGWSpec, \ IscsiServiceSpec, HostPlacementSpec, CustomContainerSpec - +from ceph.deployment.hostspec import SpecValidationError from orchestrator import DaemonDescription, OrchestratorError @@ -587,3 +588,299 @@ def test_daemon_description_service_name(spec: ServiceSpec, else: with pytest.raises(OrchestratorError): dd.service_name() + + +class YAMLdoc: + def __init__(self, text): + self.content = yaml.safe_load(text) + + +@pytest.mark.parametrize( + "yaml_doc,snmp_type", [ + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public +port: 9464 +snmp_destination: 192.168.1.42:162 +snmp_version: V2c +"""), 'snmp V2c'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword +port: 9464 +engine_id: 8000C53F00000000 +auth_protocol: MD5 +snmp_destination: 192.168.1.42:162 +snmp_version: V3 +"""), 'SNMP V3 authNoPriv'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +engine_id: 8000C53F00000000 +privacy_protocol: AES +snmp_destination: 192.168.1.42:162 +snmp_version: V3 +"""), 'SNMP V3 authPriv'), + ]) +def test_valid_snmp_gateway_spec(yaml_doc: YAMLdoc, snmp_type: str): + spec = ServiceSpec.from_json(yaml_doc.content) + spec.validate() + + +@pytest.mark.parametrize( + "yaml_doc,expected_error", [ + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public +port: 9464 +snmp_destination: 192.168.1.42:162 +snmp_version: V4 +"""), 'snmp_version unsupported. Must be one of V2c, V3'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public +port: 9464 +snmp_destination: 192.168.1.42:162 +"""), 'Missing SNMP version (snmp_version)'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword +port: 9464 +auth_protocol: wah +snmp_destination: 192.168.1.42:162 +snmp_version: V3 +"""), 'auth_protocol unsupported. Must be one of MD5, SHA'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: weewah +snmp_destination: 192.168.1.42:162 +snmp_version: V3 +"""), 'privacy_protocol unsupported. Must be one of AES, DES'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +snmp_destination: 192.168.1.42:162 +snmp_version: V3 +"""), 'Must provide an engine_id for SNMP V3 notifications'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_community: public +port: 9464 +snmp_destination: 192.168.1.42 +snmp_version: V2c +"""), 'SNMP destination (snmp_destination) type (IPv4) is invalid. Must be either: IPv4:Port, Name:Port'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: bogus +snmp_destination: 192.168.1.42:162 +snmp_version: V3 +"""), 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword +port: 9464 +auth_protocol: SHA +engine_id: 8000C53F0000000000 +snmp_version: V3 +"""), 'SNMP destination (snmp_destination) must be provided'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: 8000C53F0000000000 +snmp_destination: my.imaginary.snmp-host +snmp_version: V3 +"""), 'SNMP destination (snmp_destination) is invalid: DNS lookup failed'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: 8000C53F0000000000 +snmp_destination: 10.79.32.10:fred +snmp_version: V3 +"""), 'SNMP destination (snmp_destination) is invalid: Port must be numeric'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: 8000C53 +snmp_destination: 10.79.32.10:162 +snmp_version: V3 +"""), 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: 8000C53DOH! +snmp_destination: 10.79.32.10:162 +snmp_version: V3 +"""), 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: 8000C53FCA7344403DC611EC9B985254002537A6C53FCA7344403DC6112537A60 +snmp_destination: 10.79.32.10:162 +snmp_version: V3 +"""), 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + (YAMLdoc(""" +--- +service_type: snmp-gateway +service_name: snmp-gateway +placement: + count: 1 +spec: + credentials: + snmp_v3_auth_username: myuser + snmp_v3_auth_password: mypassword + snmp_v3_priv_password: mysecret +port: 9464 +auth_protocol: SHA +privacy_protocol: AES +engine_id: 8000C53F00000 +snmp_destination: 10.79.32.10:162 +snmp_version: V3 +"""), 'engine_id must be a string containing 10-64 hex characters. Its length must be divisible by 2'), + ]) +def test_invalid_snmp_gateway_spec(yaml_doc: YAMLdoc, expected_error: str): + with pytest.raises(SpecValidationError) as e: + ServiceSpec.from_json(yaml_doc.content) + assert str(e.value) == expected_error diff --git a/src/python-common/ceph/tests/test_utils.py b/src/python-common/ceph/tests/test_utils.py index fff67e1702b..8a94ac400b5 100644 --- a/src/python-common/ceph/tests/test_utils.py +++ b/src/python-common/ceph/tests/test_utils.py @@ -1,4 +1,7 @@ -from ceph.deployment.utils import is_ipv6, unwrap_ipv6, wrap_ipv6 +import pytest + +from ceph.deployment.utils import is_ipv6, unwrap_ipv6, wrap_ipv6, valid_addr +from typing import NamedTuple def test_is_ipv6(): @@ -35,3 +38,38 @@ def test_wrap_ipv6(): ('', ''), ('fd00::1::1', 'fd00::1::1')] for address, expected in tests: wrap_test(address, expected) + + +class Address(NamedTuple): + addr: str + status: bool + description: str + + +@pytest.mark.parametrize('addr_object', [ + Address('www.ibm.com', True, 'Name'), + Address('www.google.com:162', True, 'Name:Port'), + Address('my.big.domain.name.for.big.people', False, 'DNS lookup failed'), + Address('192.168.122.1', True, 'IPv4'), + Address('[192.168.122.1]', False, 'IPv4 address wrapped in brackets is invalid'), + Address('10.40003.200', False, 'Invalid partial IPv4 address'), + Address('10.7.5', False, 'Invalid partial IPv4 address'), + Address('10.7', False, 'Invalid partial IPv4 address'), + Address('192.168.122.5:7090', True, 'IPv4:Port'), + Address('fe80::7561:c8fb:d3d7:1fa4', True, 'IPv6'), + Address('[fe80::7561:c8fb:d3d7:1fa4]:9464', True, 'IPv6:Port'), + Address('[fe80::7561:c8fb:d3d7:1fa4]', True, 'IPv6'), + Address('[fe80::7561:c8fb:d3d7:1fa4', False, + 'Address has incorrect/incomplete use of enclosing brackets'), + Address('fe80::7561:c8fb:d3d7:1fa4]', False, + 'Address has incorrect/incomplete use of enclosing brackets'), + Address('fred.knockinson.org', False, 'DNS lookup failed'), + Address('tumbleweed.pickles.gov.au', False, 'DNS lookup failed'), + Address('192.168.122.5:00PS', False, 'Port must be numeric'), + Address('[fe80::7561:c8fb:d3d7:1fa4]:DOH', False, 'Port must be numeric') +]) +def test_valid_addr(addr_object: Address): + + valid, description = valid_addr(addr_object.addr) + assert valid == addr_object.status + assert description == addr_object.description -- 2.39.5