storage/v2: enforce some partition rules

* add some exceptions on likely desired behaviors that are not yet
  implemented, skip the weird ones
* misc cleanup also

subiquity/server/controllers/filesystem.py

@@ -297,6 +297,12 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                     break
         return [labels.for_client(disk) for disk in bitlockered_disks]
 
+    def get_partition(self, disk, number):
+        for p in disk.partitions():
+            if p._number == number:
+                return p
+        raise ValueError(f'Partition {number} on {disk.id} not found')
+
     async def v2_GET(self) -> StorageResponseV2:
         disks = self.model._all(type='disk')
         return StorageResponseV2(disks=[labels.for_client(d) for d in disks])
@@ -328,45 +334,45 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
 
     async def v2_add_partition_POST(self, data: ModifyPartitionV2) \
             -> StorageResponseV2:
+        if data.partition.format is None or data.partition.mount is None:
+            raise ValueError('add_partition must supply format and mount')
+        if data.partition.grub_device is not None:
+            raise ValueError('add_partition does not support changing '
+                             + 'grub_device')
+
         disk = self.model._one(id=data.disk_id)
-        flag = ""
-        wipe = "superblock"
-        grub_device = data.partition.grub_device
-        if grub_device:
-            flag = "boot"
         size = data.partition.size
         if size is None or size < 0:
             size = disk.free_for_partitions
         spec = {
-            "size": size,
+            'size': size,
-            "fstype": data.partition.format,
+            'fstype': data.partition.format,
-            "mount": data.partition.mount
+            'mount': data.partition.mount,
         }
 
-        self.create_partition(disk, spec, flag, wipe, grub_device)
+        self.create_partition(disk, spec, '', 'superblock', None)
         return await self.v2_GET()
 
-    def get_partition(self, disk_id, number):
-        disk = self.model._one(id=disk_id)
-        for p in disk.partitions():
-            if p._number == number:
-                return p
-        raise ValueError(f'Partition {number} on {disk_id} not found')
-
     async def v2_delete_partition_POST(self, data: ModifyPartitionV2) \
             -> StorageResponseV2:
-        partition = self.get_partition(data.disk_id, data.partition.number)
+        disk = self.model._one(id=data.disk_id)
+        partition = self.get_partition(disk, data.partition.number)
         self.delete_partition(partition)
         return await self.v2_GET()
 
     async def v2_edit_partition_POST(self, data: ModifyPartitionV2) \
             -> StorageResponseV2:
-        partition = self.get_partition(data.disk_id, data.partition.number)
         disk = self.model._one(id=data.disk_id)
+        partition = self.get_partition(disk, data.partition.number)
+        if data.partition.size not in (None, partition.size):
+            raise ValueError('edit_partition does not support changing size')
+        if data.partition.grub_device not in (None, partition.grub_device):
+            raise ValueError('edit_partition does not support changing '
+                             + 'grub_device')
+
         spec = {
-            "fstype": data.partition.format,
+            'fstype': data.partition.format,
-            "mount": data.partition.mount,
+            'mount': data.partition.mount,
-            "grub_device": data.partition.grub_device,
         }
         self.partition_disk_handler(disk, partition, spec)
         return await self.v2_GET()

subiquity/tests/api/test_api.py

@@ -228,7 +228,7 @@ class TestSimple(TestAPI):
         self.assertEqual(simple_add, manual_add)
 
 
-class TestWin10Start(TestAPI):
+class TestWin10(TestAPI):
     machine_config = 'examples/win10.json'
 
     @timeout(5)
@@ -262,12 +262,9 @@ class TestWin10Start(TestAPI):
         data = {
             'disk_id': disk_id,
             'partition': {
-                'size': -1,
                 'number': 1,
                 'format': 'ext3',
                 'mount': '/',
-                'grub_device': True,
-                'preserve': False,
             }
         }
         add_resp = await self.post('/storage/v2/add_partition', data)
@@ -324,7 +321,6 @@ class TestWin10Start(TestAPI):
         data = {
             'disk_id': disk_id,
             'partition': {
-                'size': -1,
                 'number': 4,
                 'mount': '/',
                 'format': 'ext4',
@@ -343,12 +339,10 @@ class TestWin10Start(TestAPI):
         data = {
             'disk_id': disk_id,
             'partition': {
-                'size': 0,
                 'number': 3,
                 'format': 'ext4',
                 'mount': '/',
                 'grub_device': False,
-                # 'preserve': False,
             }
         }
         resp = await self.post('/storage/v2/edit_partition', data)
@@ -382,6 +376,69 @@ class TestWin10Start(TestAPI):
         sda = first(resp['disks'], 'id', 'disk-sda')
         self.assertEqual('gpt', sda['ptable'])
 
+    @timeout(5)
+    async def test_add_rules(self):
+        disk_id = 'disk-sda'
+        await self.post('/storage/v2/reformat_disk', disk_id=disk_id)
+
+        bad_partitions = [
+            {'partition': {}},
+            {'partition': {'format': 'ext4'}},
+            {'partition': {'mount': '/'}},
+        ]
+        for partition in bad_partitions:
+            with self.assertRaises(ClientException):
+                data = {'disk_id': disk_id, 'partition': partition}
+                await self.post('/storage/v2/add_partition', data)
+
+        for size, expected in ((None, 85360377856), (20 << 30, 20 << 30)):
+            partition = {'format': 'ext4', 'mount': '/', 'size': size}
+            data = {'disk_id': disk_id, 'partition': partition}
+            resp = await self.post('/storage/v2/add_partition', data)
+            sda = first(resp['disks'], 'id', disk_id)
+            sda2 = first(sda['partitions'], 'number', 2)
+            self.assertEqual(expected, sda2['size'])
+            await self.post('/storage/v2/reformat_disk', disk_id=disk_id)
+
+
+                # required field number
+                #    optional fields wipe, mount, format
+                #    It is an error to modify other Partition fields.
+                # size: Optional[int] = None
+                # number: Optional[int] = None
+                # preserve: Optional[bool] = None
+                # wipe: Optional[str] = None
+                # annotations: Optional[List[str]] = []
+                # mount: Optional[str] = None
+                # format: Optional[str] = None
+                # grub_device: Optional[bool] = None
+    @timeout(5)
+    async def test_edit_rules(self):
+        disk_id = 'disk-sda'
+        data = {
+            'disk_id': disk_id,
+            'partition': {
+                'number': 3,
+                'format': 'ext4',
+                'mount': '/',
+                'wipe': None,
+            }
+        }
+        await self.post('/storage/v2/edit_partition', data)
+
+        data['partition']['size'] = 85240896512 // 2
+        with self.assertRaises(ClientException):
+            await self.post('/storage/v2/edit_partition', data)
+
+        data['partition']['size'] = None
+        data['partition']['grub_device'] = True
+        with self.assertRaises(ClientException):
+            await self.post('/storage/v2/edit_partition', data)
+
+        data['partition']['grub_device'] = None
+        data['partition']['format'] = 'btrfs'
+        await self.post('/storage/v2/edit_partition', data)
+
 
 class TestManyDisks(TestAPI):
     machine_config = 'examples/many-nics-and-disks.json'