Merge pull request #1138 from jpnurmi/disk-model-vendor

filesystem: add disk model and vendor name

subiquity/common/filesystem/labels.py

@@ -270,7 +270,9 @@ def _for_client_disk(disk, *, min_size=0):
         partitions=[for_client(p) for p in disk._partitions],
         free_for_partitions=disk.free_for_partitions,
         boot_device=boot.is_boot_device(disk),
-        ok_for_guided=disk.size >= min_size)
+        ok_for_guided=disk.size >= min_size,
+        model=getattr(disk, 'model', None),
+        vendor=getattr(disk, 'vendor', None))
 
 
 @for_client.register(Partition)

subiquity/common/types.py

@@ -272,6 +272,8 @@ class Disk:
     path: Optional[str]
     free_for_partitions: int
     boot_device: bool
+    model: Optional[str] = None
+    vendor: Optional[str] = None
 
 
 @attr.s(auto_attribs=True)

subiquity/models/filesystem.py

@@ -569,7 +569,6 @@ class Disk(_Device):
     wwn = attr.ib(default=None)
     multipath = attr.ib(default=None)
     path = attr.ib(default=None)
-    model = attr.ib(default=None)
     wipe = attr.ib(default=None)
     preserve = attr.ib(default=False)
     name = attr.ib(default="")
@@ -642,6 +641,20 @@ class Disk(_Device):
 
     ok_for_lvm_vg = ok_for_raid
 
+    @property
+    def model(self):
+        return self._decode_id('ID_MODEL_ENC')
+
+    @property
+    def vendor(self):
+        return self._decode_id('ID_VENDOR_ENC')
+
+    def _decode_id(self, id):
+        id = self._info.raw.get(id)
+        if id is None:
+            return None
+        return id.encode('utf-8').decode('unicode_escape').strip()
+
 
 @fsobj("partition")
 class Partition(_Formattable):

subiquity/tests/api/test_api.py

@@ -675,6 +675,22 @@ class TestInfo(TestAPI):
             sda = first(resp['disks'], 'id', disk_id)
             self.assertEqual('/dev/sda', sda['path'])
 
+    async def test_model_and_vendor(self):
+        async with start_server('examples/simple.json') as inst:
+            disk_id = 'disk-sda'
+            resp = await inst.get('/storage/v2')
+            sda = first(resp['disks'], 'id', disk_id)
+            self.assertEqual('QEMU HARDDISK', sda['model'])
+            self.assertEqual('ATA', sda['vendor'])
+
+    async def test_no_vendor(self):
+        async with start_server('examples/many-nics-and-disks.json') as inst:
+            disk_id = 'disk-sda'
+            resp = await inst.get('/storage/v2')
+            sda = first(resp['disks'], 'id', disk_id)
+            self.assertEqual('QEMU HARDDISK', sda['model'])
+            self.assertEqual(None, sda['vendor'])
+
 
 class TestFree(TestAPI):
     @timeout(5)