cephadm: Update unittests for disk enclosures

Add testcases for the new Enclosure class used by
HostFacts (gather-facts subcommand)

Signed-off-by: Paul Cuzner <pcuzner@redhat.com>

diff --git a/src/cephadm/tests/fixtures.py b/src/cephadm/tests/fixtures.py
index 1f5f171214863ff577a2b6fa20277925b6c58fa4..49ba6f1b9bf66446580a5ba7764beb4a4c8c706c 100644 (file)
--- a/src/cephadm/tests/fixtures.py
+++ b/src/cephadm/tests/fixtures.py
@@ -1,3 +1,4 @@
+from importlib.resources import contents
 import mock
 import os
 import pytest
@@ -73,10 +74,38 @@ def cephadm_fs(
             fs.create_dir(cd.LOCK_DIR)
             fs.create_dir(cd.LOGROTATE_DIR)
             fs.create_dir(cd.UNIT_DIR)
+            fs.create_dir('/sys/block')
 
             yield fs
 
 
+@pytest.fixture()
+def host_sysfs(fs: fake_filesystem.FakeFilesystem):
+    """Create a fake filesystem to represent sysfs"""
+    enc_path = '/sys/class/scsi_generic/sg2/device/enclosure/0:0:1:0'
+    dev_path = '/sys/class/scsi_generic/sg2/device'
+    slot_count = 12 
+    fs.create_dir(dev_path)
+    fs.create_file(os.path.join(dev_path, 'vendor'), contents="EnclosuresInc")
+    fs.create_file(os.path.join(dev_path, 'model'), contents="D12")
+    fs.create_file(os.path.join(enc_path, 'id'), contents='1')
+    fs.create_file(os.path.join(enc_path, 'components'), contents=str(slot_count))
+    for slot_num in range(slot_count):
+        slot_dir = os.path.join(enc_path, str(slot_num))
+        fs.create_file(os.path.join(slot_dir, 'locate'), contents='0')
+        fs.create_file(os.path.join(slot_dir, 'fault'), contents='0')
+        fs.create_file(os.path.join(slot_dir, 'slot'), contents=str(slot_num))
+        if slot_num < 6:
+            fs.create_file(os.path.join(slot_dir, 'status'), contents='Ok')
+            slot_dev = os.path.join(slot_dir, 'device')
+            fs.create_dir(slot_dev)
+            fs.create_file(os.path.join(slot_dev, 'vpd_pg80'), contents=f'fake{slot_num:0>3}')
+        else:
+            fs.create_file(os.path.join(slot_dir, 'status'), contents='not installed')
+
+    yield fs
+
+
 @contextmanager
 def with_cephadm_ctx(
     cmd: List[str],

diff --git a/src/cephadm/tests/test_enclosure.py b/src/cephadm/tests/test_enclosure.py
new file mode 100644 (file)
index 0000000..d713573
--- /dev/null
+++ b/src/cephadm/tests/test_enclosure.py
@@ -0,0 +1,74 @@
+import pytest
+
+from unittest import mock
+from tests.fixtures import host_sysfs
+
+with mock.patch('builtins.open', create=True):
+    from importlib.machinery import SourceFileLoader
+    cd = SourceFileLoader('cephadm', 'cephadm').load_module()
+
+
+@pytest.fixture
+def enclosure(host_sysfs):
+    e = cd.Enclosure(
+        enc_id='1',
+        enc_path='/sys/class/scsi_generic/sg2/device/enclosure/0:0:1:0',
+        dev_path='/sys/class/scsi_generic/sg2')
+    yield e
+
+
+class TestEnclosure:
+
+    def test_enc_metadata(self, enclosure):
+        """Check metadata for the enclosure e.g. vendor and model"""
+       
+        assert enclosure.vendor == "EnclosuresInc"
+        assert enclosure.components == '12'
+        assert enclosure.model == "D12"
+        assert enclosure.enc_id == '1'
+
+        assert enclosure.ses_paths == ['sg2']
+        assert enclosure.path_count == 1
+
+    def test_enc_slots(self, enclosure):
+        """Check slot count"""
+
+        assert len(enclosure.slot_map) == 12
+
+    def test_enc_slot_format(self, enclosure):
+        """Check the attributes of a slot are as expected"""
+
+        assert all(k in ['fault', 'locate', 'serial', 'status'] 
+                   for k, _v in enclosure.slot_map['0'].items())
+
+    def test_enc_slot_status(self, enclosure):
+        """Check the number of occupied slots is correct"""
+
+        occupied_slots = [slot_id for slot_id in enclosure.slot_map 
+                          if enclosure.slot_map[slot_id].get('status').upper() == 'OK']
+
+        assert len(occupied_slots) == 6
+
+    def test_enc_disk_count(self, enclosure):
+        """Check the disks found matches the slot info"""
+
+        assert len(enclosure.device_lookup) == 6
+        assert enclosure.device_count == 6
+
+    def test_enc_device_serial(self, enclosure):
+        """Check the device serial numbers are as expected"""
+        
+        assert all(fake_serial in enclosure.device_lookup.keys() 
+                   for fake_serial in [
+                       'fake000',
+                       'fake001',
+                       'fake002',
+                       'fake003',
+                       'fake004',
+                       'fake005'])
+
+    def test_enc_slot_to_serial(self, enclosure):
+        """Check serial number to slot matches across slot_map and device_lookup"""
+
+        for serial, slot in enclosure.device_lookup.items():
+            assert enclosure.slot_map[slot].get('serial') == serial