From 18f04d11728bfd66d335c447b996890abf56189a Mon Sep 17 00:00:00 2001 From: Dan Bungert Date: Tue, 5 Oct 2021 18:53:50 -0600 Subject: [PATCH] storage/v2: more explicit machine config init Use a context manager and an explicit machine config arg per test to help clarify the background machine assumptions. Restructure tests for the above, and to make them more targeted on a single idea per test. --- subiquity/tests/api/test_api.py | 823 +++++++++++++++++--------------- 1 file changed, 446 insertions(+), 377 deletions(-) diff --git a/subiquity/tests/api/test_api.py b/subiquity/tests/api/test_api.py index b44390534..5deff0d49 100755 --- a/subiquity/tests/api/test_api.py +++ b/subiquity/tests/api/test_api.py @@ -4,25 +4,19 @@ from aiohttp.client_exceptions import ClientResponseError import async_timeout import asyncio +import contextlib from functools import wraps import json -import logging import os -import sys import unittest +from urllib.parse import unquote from subiquitycore.utils import astart_command -logging.basicConfig(level=logging.DEBUG) socket_path = '.subiquity/socket' -ver = sys.version_info -if ver.major < 3 or ver.minor < 8: - raise Exception('skip asyncio testing') - - def find(items, key, value): for item in items: if item[key] == value: @@ -47,27 +41,24 @@ def json_print(json_data): print(json.dumps(json_data, indent=4)) -def loads(data): - return json.loads(data) if data else None - - -def dumps(data): - # if the data we're dumping is literally False, we want that to be 'false' - if data or type(data) is bool: - return json.dumps(data, separators=(',', ':')) - elif data is not None: - return '""' - else: - return data - - -class ClientException(Exception): - pass +class Server: + def __init__(self, session): + self.session = session + def loads(self, data): + if data == '' or data is None: # json.loads likes neither of these + return None + return json.loads(data) -class TestAPI(unittest.IsolatedAsyncioTestCase): - machine_config = 'examples/simple.json' - need_spawn_server = True + def dumps(self, data): + # if the data we're dumping is literally False, + # we want that to be 'false' + if data or type(data) is bool: + return json.dumps(data, separators=(',', ':')) + elif data is not None: + return '""' + else: + return data async def get(self, query, **kwargs): return await self.request('GET', query, **kwargs) @@ -76,23 +67,16 @@ async def post(self, query, data=None, **kwargs): return await self.request('POST', query, data, **kwargs) async def request(self, method, query, data=None, **kwargs): - params = {} - for key in kwargs: - params[key] = dumps(kwargs[key]) - data = dumps(data) - info = f'{method} {query}' - if params: - for i, key in enumerate(params): - joiner = '?' if i == 0 else '&' - info += f'{joiner}{key}={params[key]}' - print(info) + params = {k: self.dumps(v) for k, v in kwargs.items()} + data = self.dumps(data) async with self.session.request(method, f'http://a{query}', data=data, params=params) as resp: + print(unquote(str(resp.url))) resp.raise_for_status() content = await resp.content.read() - return loads(content.decode()) + return self.loads(content.decode()) - async def server_startup(self): + async def poll_startup(self): for _ in range(20): try: await self.get('/meta/status') @@ -104,418 +88,503 @@ async def server_startup(self): async def server_shutdown(self, immediate=True): try: await self.post('/shutdown', mode='POWEROFF', immediate=immediate) - raise Exception('expected ServerDisconnectedError') except aiohttp.client_exceptions.ServerDisconnectedError: return - async def spawn_server(self): + async def spawn(self, machine_config): env = os.environ.copy() env['SUBIQUITY_REPLAY_TIMESCALE'] = '100' cmd = 'python3 -m subiquity.cmd.server --dry-run --bootloader uefi' \ - + ' --machine-config ' + self.machine_config + + ' --machine-config ' + machine_config cmd = cmd.split(' ') self.proc = await astart_command(cmd, env=env) - self.server = asyncio.create_task(self.proc.communicate()) - - async def asyncSetUp(self): - if self.need_spawn_server: - await self.spawn_server() - - # setup client - conn = aiohttp.UnixConnector(path=socket_path) - self.session = aiohttp.ClientSession(connector=conn) - await self.server_startup() - - async def asyncTearDown(self): - if self.need_spawn_server: - await self.server_shutdown() - await self.session.close() - await self.server - try: - self.proc.kill() - except ProcessLookupError: - pass + self.server_task = asyncio.create_task(self.proc.communicate()) + + async def close(self): + await self.server_shutdown() + await self.server_task + try: + self.proc.kill() + except ProcessLookupError: + pass + + +class TestAPI(unittest.IsolatedAsyncioTestCase): + pass -# class TestDebug(TestAPI): -# need_spawn_server = False +@contextlib.asynccontextmanager +async def start_server(machine_config): + conn = aiohttp.UnixConnector(path=socket_path) + async with aiohttp.ClientSession(connector=conn) as session: + server = Server(session) + try: + await server.spawn(machine_config) + await server.poll_startup() + yield server + finally: + await server.close() -# @unittest.skip("useful for interactive debug only") -# async def test_v2_delete_debug(self): -# await self.get('/storage/v2') +class TestBitlocker(TestAPI): + @timeout(5) + async def test_has_bitlocker(self): + async with start_server('examples/win10.json') as inst: + resp = await inst.get('/storage/has_bitlocker') + self.assertEqual(1, len(resp)) -class TestSimple(TestAPI): @timeout(5) async def test_not_bitlocker(self): - resp = await self.get('/storage/has_bitlocker') - self.assertEqual(0, len(resp)) + async with start_server('examples/simple.json') as inst: + resp = await inst.get('/storage/has_bitlocker') + self.assertEqual(0, len(resp)) + +class TestFlow(TestAPI): @timeout(10) async def test_server_flow(self): - await self.post('/locale', 'en_US.UTF-8') - keyboard = { - 'layout': 'us', - 'variant': '', - 'toggle': None - } - await self.post('/keyboard', keyboard) - await self.post('/source', source_id='ubuntu-server') - await self.post('/network') - await self.post('/proxy', '') - await self.post('/mirror', 'http://us.archive.ubuntu.com/ubuntu') - resp = await self.get('/storage/guided') - disk_id = resp['disks'][0]['id'] - choice = {"disk_id": disk_id} - await self.post('/storage/v2/guided', choice=choice) - await self.post('/storage/v2') - await self.get('/meta/status', cur='WAITING') - await self.post('/meta/confirm', tty='/dev/tty1') - await self.get('/meta/status', cur='NEEDS_CONFIRMATION') - identity = { - 'realname': 'ubuntu', - 'username': 'ubuntu', - 'hostname': 'ubuntu-server', - 'crypted_password': '$6$exDY1mhS4KUYCE/2$zmn9ToZwTKLhCw.b4/' - + 'b.ZRTIZM30JZ4QrOQ2aOXJ8yk96xpcCof0kx' - + 'KwuX1kqLG/ygbJ1f8wxED22bTL4F46P0' - } - await self.post('/identity', identity) - ssh = { - 'install_server': False, - 'allow_pw': False, - 'authorized_keys': [] - } - await self.post('/ssh', ssh) - await self.post('/snaplist', []) - for state in 'RUNNING', 'POST_WAIT', 'POST_RUNNING', 'UU_RUNNING': - await self.get('/meta/status', cur=state) + async with start_server('examples/simple.json') as inst: + await inst.post('/locale', 'en_US.UTF-8') + keyboard = { + 'layout': 'us', + 'variant': '', + 'toggle': None + } + await inst.post('/keyboard', keyboard) + await inst.post('/source', source_id='ubuntu-server') + await inst.post('/network') + await inst.post('/proxy', '') + await inst.post('/mirror', 'http://us.archive.ubuntu.com/ubuntu') + resp = await inst.get('/storage/guided') + disk_id = resp['disks'][0]['id'] + choice = {"disk_id": disk_id} + await inst.post('/storage/v2/guided', choice=choice) + await inst.post('/storage/v2') + await inst.get('/meta/status', cur='WAITING') + await inst.post('/meta/confirm', tty='/dev/tty1') + await inst.get('/meta/status', cur='NEEDS_CONFIRMATION') + identity = { + 'realname': 'ubuntu', + 'username': 'ubuntu', + 'hostname': 'ubuntu-server', + 'crypted_password': '$6$exDY1mhS4KUYCE/2$zmn9ToZwTKLhCw.b4/' + + 'b.ZRTIZM30JZ4QrOQ2aOXJ8yk96xpcCof0kx' + + 'KwuX1kqLG/ygbJ1f8wxED22bTL4F46P0' + } + await inst.post('/identity', identity) + ssh = { + 'install_server': False, + 'allow_pw': False, + 'authorized_keys': [] + } + await inst.post('/ssh', ssh) + await inst.post('/snaplist', []) + for state in 'RUNNING', 'POST_WAIT', 'POST_RUNNING', 'UU_RUNNING': + await inst.get('/meta/status', cur=state) @timeout(5) - async def test_default(self): - choice = {'disk_id': 'disk-sda'} - resp = await self.post('/storage/guided', choice=choice) - self.assertEqual(7, len(resp['config'])) - self.assertEqual('/dev/sda', resp['config'][0]['path']) + async def test_v2_flow(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + orig_resp = await inst.get('/storage/v2') + sda = first(orig_resp['disks'], 'id', disk_id) + self.assertTrue(len(sda['partitions']) > 0) + + resp = await inst.post('/storage/v2/reformat_disk', + disk_id=disk_id) + sda = first(resp['disks'], 'id', disk_id) + self.assertEqual(0, len(sda['partitions'])) + + data = { + 'disk_id': disk_id, + 'partition': { + 'format': 'ext3', + 'mount': '/', + } + } + add_resp = await inst.post('/storage/v2/add_partition', data) + sda = first(add_resp['disks'], 'id', disk_id) + sda2 = first(sda['partitions'], 'number', 2) + self.assertEqual('ext3', sda2['format']) + + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 2, + 'format': 'ext4', + } + } + resp = await inst.post('/storage/v2/edit_partition', data) + expected = add_resp['disks'][0]['partitions'][1] + actual = resp['disks'][0]['partitions'][1] + for key in 'size', 'number', 'mount', 'grub_device': + self.assertEqual(expected[key], actual[key]) + self.assertEqual('ext4', + resp['disks'][0]['partitions'][1]['format']) + + resp = await inst.post('/storage/v2/delete_partition', data) + self.assertEqual(1, len(resp['disks'][0]['partitions'])) + + resp = await inst.post('/storage/v2/reset') + self.assertEqual(orig_resp, resp) + + choice = {'disk_id': disk_id} + guided_resp = await inst.post('/storage/v2/guided', choice=choice) + post_resp = await inst.post('/storage/v2') + # posting to the endpoint shouldn't change the answer + self.assertEqual(guided_resp, post_resp) + +class TestGuided(TestAPI): @timeout(5) async def test_guided_v2(self): - choice = {'disk_id': 'disk-sda'} - resp = await self.post('/storage/v2/guided', choice=choice) - self.assertEqual(1, len(resp['disks'])) - self.assertEqual('disk-sda', resp['disks'][0]['id']) + async with start_server('examples/simple.json') as inst: + choice = {'disk_id': 'disk-sda'} + resp = await inst.post('/storage/v2/guided', choice=choice) + self.assertEqual(1, len(resp['disks'])) + self.assertEqual('disk-sda', resp['disks'][0]['id']) + +class TestAdd(TestAPI): @timeout(5) async def test_v2_add_boot_partition(self): - self.maxDiff = None - disk_id = 'disk-sda' - - data = { - 'disk_id': disk_id, - 'partition': { - 'format': 'ext4', - 'mount': '/', + async with start_server('examples/simple.json') as inst: + disk_id = 'disk-sda' + + data = { + 'disk_id': disk_id, + 'partition': { + 'format': 'ext4', + 'mount': '/', + } } - } - simple_add = await self.post('/storage/v2/add_partition', data) - self.assertEqual(2, len(simple_add['disks'][0]['partitions'])) + single_add = await inst.post('/storage/v2/add_partition', data) + self.assertEqual(2, len(single_add['disks'][0]['partitions'])) - await self.post('/storage/v2/reset') + await inst.post('/storage/v2/reset') - await self.post('/storage/v2/add_boot_partition', disk_id=disk_id) - manual_add = await self.post('/storage/v2/add_partition', data) - self.assertEqual(simple_add, manual_add) + # these manual steps are expected to be equivalent to just adding + # the single partition and getting the automatic boot partition + await inst.post('/storage/v2/add_boot_partition', disk_id=disk_id) + manual_add = await inst.post('/storage/v2/add_partition', data) + self.assertEqual(single_add, manual_add) - -class TestWin10(TestAPI): - machine_config = 'examples/win10.json' + @timeout(5) + async def test_v2_free_for_partitions(self): + async with start_server('examples/simple.json') as inst: + disk_id = 'disk-sda' + resp = await inst.post('/storage/v2/add_boot_partition', + disk_id=disk_id) + sda = first(resp['disks'], 'id', disk_id) + orig_free = sda['free_for_partitions'] + + size_requested = 6 << 30 + expected_free = orig_free - size_requested + data = { + 'disk_id': disk_id, + 'partition': { + 'size': size_requested, + 'format': 'ext4', + 'mount': '/', + } + } + resp = await inst.post('/storage/v2/add_partition', data) + sda = first(resp['disks'], 'id', disk_id) + self.assertEqual(expected_free, sda['free_for_partitions']) @timeout(5) - async def test_bitlocker(self): - resp = await self.get('/storage/has_bitlocker') - self.assertEqual('BitLocker', resp[0]['partitions'][2]['format']) + async def test_add_format_required(self): + disk_id = 'disk-sda' + async with start_server('examples/simple.json') as inst: + bad_partitions = [ + {}, + {'mount': '/'}, + ] + for partition in bad_partitions: + data = {'disk_id': disk_id, 'partition': partition} + with self.assertRaises(ClientResponseError, + msg=f'data {data}'): + await inst.post('/storage/v2/add_partition', data) @timeout(5) - async def test_delete_bad(self): - data = { - 'disk_id': 'disk-sda', - 'partition': { - 'size': -1, - 'number': 5, + async def test_add_default_size_handling(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) + expected_total = sda['free_for_partitions'] + + data = { + 'disk_id': disk_id, + 'partition': { + 'format': 'ext4', + 'mount': '/', + } } - } - with self.assertRaises(ClientResponseError): - await self.post('/storage/v2/delete_partition', data) + resp = await inst.post('/storage/v2/add_partition', data) + sda = first(resp['disks'], 'id', disk_id) + sda1 = first(sda['partitions'], 'number', 1) + sda2 = first(sda['partitions'], 'number', 2) + self.assertEqual(expected_total, sda1['size'] + sda2['size']) + +class TestDelete(TestAPI): @timeout(5) - async def test_v2_flow(self): - disk_id = 'disk-sda' - orig_resp = await self.get('/storage/v2') - self.assertEqual(1, len(orig_resp['disks'])) - self.assertEqual('disk-sda', orig_resp['disks'][0]['id']) - self.assertEqual(4, len(orig_resp['disks'][0]['partitions'])) - - resp = await self.post('/storage/v2/reformat_disk', disk_id=disk_id) - self.assertEqual(0, len(resp['disks'][0]['partitions'])) - - data = { - 'disk_id': disk_id, - 'partition': { - 'number': 1, - 'format': 'ext3', - 'mount': '/', + async def test_v2_delete_without_reformat(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + data = { + 'disk_id': disk_id, + 'partition': {'number': 1} } - } - add_resp = await self.post('/storage/v2/add_partition', data) - self.assertEqual(2, len(add_resp['disks'][0]['partitions'])) - self.assertEqual('ext3', - add_resp['disks'][0]['partitions'][1]['format']) - - data['partition']['number'] = 2 - data['partition']['format'] = 'ext4' - resp = await self.post('/storage/v2/edit_partition', data) - expected = add_resp['disks'][0]['partitions'][1] - actual = resp['disks'][0]['partitions'][1] - for key in 'size', 'number', 'mount', 'grub_device': - self.assertEqual(expected[key], actual[key]) - self.assertEqual('ext4', resp['disks'][0]['partitions'][1]['format']) - - resp = await self.post('/storage/v2/delete_partition', data) - self.assertEqual(1, len(resp['disks'][0]['partitions'])) - - resp = await self.post('/storage/v2/reset') - self.assertEqual(orig_resp, resp) - - choice = {'disk_id': disk_id} - orig_resp = await self.post('/storage/v2/guided', choice=choice) - await self.post('/storage/v2') - resp = await self.get('/storage/v2') - # posting to the endpoint shouldn't change the answer - self.assertEqual(orig_resp, resp) + with self.assertRaises(ClientResponseError): + await inst.post('/storage/v2/delete_partition', data) @timeout(5) - async def test_v2_free_for_partitions(self): - disk_id = 'disk-sda' - resp = await self.post('/storage/v2/reformat_disk', disk_id=disk_id) - - data = { - 'disk_id': disk_id, - 'partition': { - 'size': resp['disks'][0]['size'] // 2, - 'number': 1, - 'format': 'ext4', - 'mount': '/', + async def test_v2_delete_with_reformat(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + await inst.post('/storage/v2/reformat_disk', disk_id=disk_id) + data = { + 'disk_id': disk_id, + 'partition': { + 'mount': '/', + 'format': 'ext4', + } + } + await inst.post('/storage/v2/add_partition', data) + data = { + 'disk_id': disk_id, + 'partition': {'number': 1} } - } - resp = await self.post('/storage/v2/add_partition', data) - self.assertEqual(2, len(resp['disks'][0]['partitions'])) - self.assertEqual('ext4', resp['disks'][0]['partitions'][1]['format']) - self.assertEqual(42410704896, resp['disks'][0]['free_for_partitions']) + await inst.post('/storage/v2/delete_partition', data) @timeout(5) - async def test_v2_delete_requires_reformat(self): - disk_id = 'disk-sda' - data = { - 'disk_id': disk_id, - 'partition': { - 'number': 4, - 'mount': '/', - 'format': 'ext4', + async def test_delete_nonexistant(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + await inst.post('/storage/v2/reformat_disk', disk_id=disk_id) + data = { + 'disk_id': disk_id, + 'partition': {'number': 1} } - } - await self.post('/storage/v2/edit_partition', data) - with self.assertRaises(ClientResponseError): - await self.post('/storage/v2/delete_partition', data) + with self.assertRaises(ClientResponseError): + await inst.post('/storage/v2/delete_partition', data) + +class TestEdit(TestAPI): @timeout(5) - async def test_v2_reuse(self): - orig_resp = await self.get('/storage/v2') + async def test_edit_no_change_size(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + resp = await inst.get('/storage/v2') - disk_id = 'disk-sda' - data = { - 'disk_id': disk_id, - 'partition': { - 'number': 3, - 'format': 'ext4', - 'mount': '/', + sda = first(resp['disks'], 'id', disk_id) + sda3 = first(sda['partitions'], 'number', 3) + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 3, + 'size': sda3['size'] - 1 << 30 + } } - } - resp = await self.post('/storage/v2/edit_partition', data) - orig_disk = orig_resp['disks'][0] - disk = resp['disks'][0] - self.assertEqual('/dev/sda', disk['path']) - - esp = disk['partitions'][0] - self.assertIsNone(esp['wipe']) - self.assertEqual('/boot/efi', esp['mount']) - self.assertEqual('vfat', esp['format']) - self.assertTrue(esp['grub_device']) - - part = disk['partitions'][1] - expected = orig_disk['partitions'][1] - self.assertEqual(expected, part) - - root = disk['partitions'][2] - self.assertEqual('superblock', root['wipe']) - self.assertEqual('/', root['mount']) - self.assertEqual('ext4', root['format']) - self.assertFalse(root['grub_device']) - - part = disk['partitions'][3] - expected = orig_disk['partitions'][3] - self.assertEqual(expected, part) + with self.assertRaises(ClientResponseError): + await inst.post('/storage/v2/edit_partition', data) @timeout(5) - async def test_v2_gpt(self): - resp = await self.get('/storage/v2') - sda = first(resp['disks'], 'id', 'disk-sda') - self.assertEqual('gpt', sda['ptable']) + async def test_edit_no_change_grub(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 3, + 'grub_device': True, + } + } + with self.assertRaises(ClientResponseError): + await inst.post('/storage/v2/edit_partition', data) @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(ClientResponseError): - data = {'disk_id': disk_id, 'partition': partition} - await self.post('/storage/v2/add_partition', data) + async def test_edit_format(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 3, + 'format': 'btrfs', + } + } + resp = await inst.post('/storage/v2/edit_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) + sda3 = first(sda['partitions'], 'number', 3) + self.assertEqual('btrfs', sda3['format']) @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, + async def test_edit_mount(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 3, + 'mount': '/', + } } - } - await self.post('/storage/v2/edit_partition', data) + resp = await inst.post('/storage/v2/edit_partition', data) - data['partition']['size'] = 85240896512 // 2 - with self.assertRaises(ClientResponseError): - await self.post('/storage/v2/edit_partition', data) - - data['partition']['size'] = None - data['partition']['grub_device'] = True - with self.assertRaises(ClientResponseError): - 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) + sda = first(resp['disks'], 'id', disk_id) + sda3 = first(sda['partitions'], 'number', 3) + self.assertEqual('/', sda3['mount']) @timeout(5) - async def test_todos_simple(self): - disk_id = 'disk-sda' - resp = await self.post('/storage/v2/reformat_disk', disk_id=disk_id) - self.assertTrue(resp['need_root']) - self.assertTrue(resp['need_boot']) - - data = { - 'disk_id': disk_id, - 'partition': { - 'format': 'ext4', - 'mount': '/', + async def test_edit_format_and_mount(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 3, + 'format': 'btrfs', + 'mount': '/', + } } - } - resp = await self.post('/storage/v2/add_partition', data) - self.assertFalse(resp['need_root']) - self.assertFalse(resp['need_boot']) + resp = await inst.post('/storage/v2/edit_partition', data) + + sda = first(resp['disks'], 'id', disk_id) + sda3 = first(sda['partitions'], 'number', 3) + self.assertEqual('btrfs', sda3['format']) + self.assertEqual('/', sda3['mount']) @timeout(5) - async def test_todos_manual(self): - disk_id = 'disk-sda' - resp = await self.post('/storage/v2/reformat_disk', disk_id=disk_id) - self.assertTrue(resp['need_root']) - self.assertTrue(resp['need_boot']) - - resp = await self.post('/storage/v2/add_boot_partition', - disk_id=disk_id) - self.assertTrue(resp['need_root']) - self.assertFalse(resp['need_boot']) - - data = { - 'disk_id': disk_id, - 'partition': { - 'format': 'ext4', - 'mount': '/', + async def test_v2_reuse(self): + async with start_server('examples/win10.json') as inst: + disk_id = 'disk-sda' + resp = await inst.get('/storage/v2') + orig_sda = first(resp['disks'], 'id', disk_id) + + data = { + 'disk_id': disk_id, + 'partition': { + 'number': 3, + 'format': 'ext4', + 'mount': '/', + } } - } - resp = await self.post('/storage/v2/add_partition', data) - self.assertFalse(resp['need_root']) - self.assertFalse(resp['need_boot']) + resp = await inst.post('/storage/v2/edit_partition', data) + sda = first(resp['disks'], 'id', disk_id) + sda1 = first(sda['partitions'], 'number', 1) + self.assertIsNone(sda1['wipe']) + self.assertEqual('/boot/efi', sda1['mount']) + self.assertEqual('vfat', sda1['format']) + self.assertTrue(sda1['grub_device']) + + sda2 = first(sda['partitions'], 'number', 2) + orig_sda2 = first(orig_sda['partitions'], 'number', 2) + self.assertEqual(orig_sda2, sda2) + + sda3 = first(sda['partitions'], 'number', 3) + self.assertEqual('superblock', sda3['wipe']) + self.assertEqual('/', sda3['mount']) + self.assertEqual('ext4', sda3['format']) + self.assertFalse(sda3['grub_device']) + + sda4 = first(sda['partitions'], 'number', 4) + orig_sda4 = first(orig_sda['partitions'], 'number', 4) + self.assertEqual(orig_sda4, sda4) + +class TestPartitionTableTypes(TestAPI): @timeout(5) - async def test_todos_guided(self): - disk_id = 'disk-sda' - resp = await self.post('/storage/v2/reformat_disk', disk_id=disk_id) - self.assertTrue(resp['need_root']) - self.assertTrue(resp['need_boot']) + async def test_ptable_gpt(self): + async with start_server('examples/win10.json') as inst: + resp = await inst.get('/storage/v2') + sda = first(resp['disks'], 'id', 'disk-sda') + self.assertEqual('gpt', sda['ptable']) - choice = {'disk_id': disk_id} - resp = await self.post('/storage/v2/guided', choice=choice) - self.assertFalse(resp['need_root']) - self.assertFalse(resp['need_boot']) + @timeout(5) + async def test_ptable_msdos(self): + async with start_server('examples/many-nics-and-disks.json') as inst: + resp = await inst.get('/storage/v2') + sda = first(resp['disks'], 'id', 'disk-sda') + self.assertEqual('msdos', sda['ptable']) @timeout(5) - async def test_edit_partial(self): - disk_id = 'disk-sda' - await self.post('/storage/v2/reformat_disk', disk_id=disk_id) + async def test_ptable_none(self): + async with start_server('examples/simple.json') as inst: + resp = await inst.get('/storage/v2') + sda = first(resp['disks'], 'id', 'disk-sda') + self.assertEqual(None, sda['ptable']) + - data = { - 'disk_id': disk_id, - 'partition': { - 'format': 'ext4', - 'mount': '/', +class TestTodos(TestAPI): # server indicators of required client actions + @timeout(5) + async def test_todos_simple(self): + async with start_server('examples/simple.json') as inst: + disk_id = 'disk-sda' + resp = await inst.post('/storage/v2/reformat_disk', + disk_id=disk_id) + self.assertTrue(resp['need_root']) + self.assertTrue(resp['need_boot']) + + data = { + 'disk_id': disk_id, + 'partition': { + 'format': 'ext4', + 'mount': '/', + } } - } - await self.post('/storage/v2/add_partition', data) + resp = await inst.post('/storage/v2/add_partition', data) + self.assertFalse(resp['need_root']) + self.assertFalse(resp['need_boot']) - data['partition'].update({ - 'number': 2, - 'format': None, - 'mount': '/home' - }) - resp = await self.post('/storage/v2/edit_partition', data) + @timeout(5) + async def test_todos_manual(self): + async with start_server('examples/simple.json') as inst: + disk_id = 'disk-sda' + resp = await inst.post('/storage/v2/reformat_disk', + disk_id=disk_id) + self.assertTrue(resp['need_root']) + self.assertTrue(resp['need_boot']) + + resp = await inst.post('/storage/v2/add_boot_partition', + disk_id=disk_id) + self.assertTrue(resp['need_root']) + self.assertFalse(resp['need_boot']) + + data = { + 'disk_id': disk_id, + 'partition': { + 'format': 'ext4', + 'mount': '/', + } + } + resp = await inst.post('/storage/v2/add_partition', data) + self.assertFalse(resp['need_root']) + self.assertFalse(resp['need_boot']) - sda = first(resp['disks'], 'id', disk_id) - sda2 = first(sda['partitions'], 'number', 2) - self.assertEqual('ext4', sda2['format']) - self.assertEqual('/home', sda2['mount']) + @timeout(5) + async def test_todos_guided(self): + async with start_server('examples/simple.json') as inst: + disk_id = 'disk-sda' + resp = await inst.post('/storage/v2/reformat_disk', + disk_id=disk_id) + self.assertTrue(resp['need_root']) + self.assertTrue(resp['need_boot']) + choice = {'disk_id': disk_id} + resp = await inst.post('/storage/v2/guided', choice=choice) + self.assertFalse(resp['need_root']) + self.assertFalse(resp['need_boot']) -class TestManyDisks(TestAPI): - machine_config = 'examples/many-nics-and-disks.json' +class TestInfo(TestAPI): @timeout(5) - async def test_v2_msdos(self): - disk_id = 'disk-sda' - resp = await self.get('/storage/v2') - sda = first(resp['disks'], 'id', disk_id) - self.assertEqual('msdos', sda['ptable']) - - sda5 = first(sda['partitions'], 'number', 5) - sda5.update({'format': 'ext4', 'mount': '/'}) - data = { - 'disk_id': disk_id, - 'partition': sda5 - } - await self.post('/storage/v2/edit_partition', data) + async def test_path(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('/dev/sda', sda['path'])