]> git.apps.os.sepia.ceph.com Git - ceph-ci.git/commitdiff
python-common: update smb service spec to use new smb constants
authorJohn Mulligan <jmulligan@redhat.com>
Wed, 13 Aug 2025 15:08:40 +0000 (11:08 -0400)
committerJohn Mulligan <jmulligan@redhat.com>
Fri, 15 Aug 2025 16:41:40 +0000 (12:41 -0400)
Update the SMBSpec class to use the new constants provided by the
recently added python-common smb package.

Signed-off-by: John Mulligan <jmulligan@redhat.com>
src/python-common/ceph/deployment/service_spec.py

index 0765f0c2d5532b98bd8207a84dd852f6cbaa5626..e5c7ddf64d4de3960376fc80079fc48c03e16685 100644 (file)
@@ -35,6 +35,7 @@ from ceph.deployment.utils import unwrap_ipv6, valid_addr, verify_non_negative_i
 from ceph.deployment.utils import verify_positive_int, verify_non_negative_number
 from ceph.deployment.utils import verify_boolean, verify_enum
 from ceph.utils import is_hex
+from ceph.smb import constants as smbconst
 
 ServiceSpecT = TypeVar('ServiceSpecT', bound='ServiceSpec')
 FuncT = TypeVar('FuncT', bound=Callable)
@@ -3315,8 +3316,8 @@ class SMBClusterBindIPSpec:
 
 class SMBSpec(ServiceSpec):
     service_type = 'smb'
-    _valid_features = {'domain', 'clustered', 'cephfs-proxy', 'remote-control'}
-    _valid_service_names = {'smb', 'smbmetrics', 'ctdb', 'remote-control'}
+    _valid_features = smbconst.FEATURES
+    _valid_service_names = smbconst.SERVICES
     _default_cluster_meta_obj = 'cluster.meta.json'
     _default_cluster_lock_obj = 'cluster.meta.lock'
 
@@ -3430,23 +3431,24 @@ class SMBSpec(ServiceSpec):
                 raise ValueError(
                     f'invalid feature flags: {", ".join(invalid)}'
                 )
-        if 'clustered' in self.features and not self.cluster_meta_uri:
+        _clustered = smbconst.CLUSTERED
+        if _clustered in self.features and not self.cluster_meta_uri:
             # derive a cluster meta uri from config uri by default (if possible)
             self.cluster_meta_uri = self._derive_cluster_uri(
                 self.config_uri,
                 self._default_cluster_meta_obj,
             )
-        if 'clustered' not in self.features and self.cluster_meta_uri:
+        if _clustered not in self.features and self.cluster_meta_uri:
             raise ValueError(
                 'cluster meta uri unsupported when "clustered" feature not set'
             )
-        if 'clustered' in self.features and not self.cluster_lock_uri:
+        if _clustered in self.features and not self.cluster_lock_uri:
             # derive a cluster meta uri from config uri by default (if possible)
             self.cluster_lock_uri = self._derive_cluster_uri(
                 self.config_uri,
                 self._default_cluster_lock_obj,
             )
-        if 'clustered' not in self.features and self.cluster_lock_uri:
+        if _clustered not in self.features and self.cluster_lock_uri:
             raise ValueError(
                 'cluster lock uri unsupported when "clustered" feature not set'
             )
@@ -3465,12 +3467,7 @@ class SMBSpec(ServiceSpec):
         return uri
 
     def _default_ports(self) -> Dict[str, int]:
-        return {
-            'smb': 445,
-            'smbmetrics': 9922,
-            'ctdb': 4379,
-            'remote-control': 54445,
-        }
+        return dict(smbconst.DEFAULT_PORTS)
 
     def service_ports(self) -> Dict[str, int]:
         ports = self._default_ports()
@@ -3479,13 +3476,13 @@ class SMBSpec(ServiceSpec):
         return ports
 
     def metrics_exporter_port(self) -> int:
-        return self.service_ports()['smbmetrics']
+        return self.service_ports()[smbconst.SMBMETRICS]
 
     def get_port_start(self) -> List[int]:
         _ports = self.service_ports()
-        ports = [_ports['smb'], _ports['smbmetrics']]
-        if 'clustered' in self.features:
-            ports.append(_ports['ctdb'])
+        ports = [_ports[smbconst.SMB], _ports[smbconst.SMBMETRICS]]
+        if smbconst.CLUSTERED in self.features:
+            ports.append(_ports[smbconst.CTDB])
         return ports
 
     def strict_cluster_ip_specs(self) -> List[Dict[str, Any]]: