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
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)
class TestApplySpec:
-
+
def test_parse_yaml(self, cephadm_fs):
yaml = '''service_type: host
hostname: vm-00
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
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(<ip>:<port>) 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'
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
'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=''
+ )
# 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
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
-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():
('', ''), ('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
projects / ceph.git / commitdiff