Avoid code duplication when connecting to Rados.

control/discovery.py

@@ -319,13 +319,7 @@ def __init__(self, config):
             if gateway_group else "nvmeof.state"
         self.logger.info(f"log pages info from omap: {self.omap_name}")
 
-        ceph_pool = self.config.get("ceph", "pool")
-        ceph_conf = self.config.get("ceph", "config_file")
-        rados_id = self.config.get_with_default("ceph", "id", "")
-        conn = rados.Rados(conffile=ceph_conf, rados_id=rados_id)
-        conn.connect()
-        self.ioctx = conn.open_ioctx(ceph_pool)
-
+        self.ioctx = self.omap_state.open_rados_connection(config)
         self.discovery_addr = self.config.get_with_default("discovery", "addr", "0.0.0.0")
         self.discovery_port = self.config.get_with_default("discovery", "port", "8009")
         if not self.discovery_addr or not self.discovery_port:

control/state.py

@@ -170,14 +170,9 @@ def __init__(self, config):
         self.watch = None
         gateway_group = self.config.get("gateway", "group")
         self.omap_name = f"nvmeof.{gateway_group}.state" if gateway_group else "nvmeof.state"
-        ceph_pool = self.config.get("ceph", "pool")
-        ceph_conf = self.config.get("ceph", "config_file")
-        rados_id = self.config.get_with_default("ceph", "id", "")
 
         try:
-            conn = rados.Rados(conffile=ceph_conf, rados_id=rados_id)
-            conn.connect()
-            self.ioctx = conn.open_ioctx(ceph_pool)
+            self.ioctx = self.open_rados_connection(self.config)
             # Create a new gateway persistence OMAP object
             with rados.WriteOpCtx() as write_op:
                 # Set exclusive parameter to fail write_op if object exists
@@ -198,6 +193,15 @@ def __exit__(self, exc_type, exc_value, traceback):
             self.watch.close()
         self.ioctx.close()
 
+    def open_rados_connection(self, config):
+        ceph_pool = config.get("ceph", "pool")
+        ceph_conf = config.get("ceph", "config_file")
+        rados_id = config.get_with_default("ceph", "id", "")
+        conn = rados.Rados(conffile=ceph_conf, rados_id=rados_id)
+        conn.connect()
+        ioctx = conn.open_ioctx(ceph_pool)
+        return ioctx
+
     def get_local_version(self) -> int:
         """Returns local version."""
         return self.version

tests/test_state.py

@@ -4,14 +4,10 @@
 from control.state import LocalGatewayState, OmapGatewayState, GatewayStateHandler
 
 
-@pytest.fixture(scope="module")
-def ioctx(config):
+@pytest.fixture
+def ioctx(omap_state, config):
     """Opens IO context to ceph pool."""
-    ceph_pool = config.get("ceph", "pool")
-    ceph_conf = config.get("ceph", "config_file")
-    conn = rados.Rados(conffile=ceph_conf)
-    conn.connect()
-    ioctx = conn.open_ioctx(ceph_pool)
+    ioctx = omap_state.open_rados_connection(config)
     yield ioctx
     ioctx.close()