diff options
| author | Olivier Gayot <[email protected]> | 2024-03-12 12:08:27 +0100 |
|---|---|---|
| committer | Olivier Gayot <[email protected]> | 2024-03-13 09:26:41 +0100 |
| commit | 4b3395bd71960406498cb8ee09db2f26fea9680c (patch) | |
| tree | fe5e7741c457ac11e9acc980dceab5be61dbc5b9 /tests/unittests/test_curthooks.py | |
| parent | cf8f7a48905109dbc28808db90fd63c3d6ec37f6 (diff) | |
nvme-o-tcp: factorize parsing of NVMe controllers
In three different places now, we have code parsing the storage
configuration to extract the NVMe controllers. This is too much.
This patch introduces a single implementation in curtin/block/nvme.py.
Signed-off-by: Olivier Gayot <[email protected]>
Diffstat (limited to 'tests/unittests/test_curthooks.py')
| -rw-r--r-- | tests/unittests/test_curthooks.py | 207 |
1 files changed, 61 insertions, 146 deletions
diff --git a/tests/unittests/test_curthooks.py b/tests/unittests/test_curthooks.py index d38c2399..526a3fe9 100644 --- a/tests/unittests/test_curthooks.py +++ b/tests/unittests/test_curthooks.py @@ -2032,126 +2032,52 @@ class TestCurthooksGrubDebconf(CiTestCase): class TestCurthooksNVMeOverTCP(CiTestCase): - def test_get_nvme_stas_controller_directives__no_nvme_controller(self): - self.assertFalse(curthooks.get_nvme_stas_controller_directives({ - "storage": { - "config": [ - {"type": "partition"}, - {"type": "mount"}, - {"type": "disk"}, - ], - }, - })) - - def test_get_nvme_stas_controller_directives__pcie_controller(self): - self.assertFalse(curthooks.get_nvme_stas_controller_directives({ - "storage": { - "config": [ - {"type": "nvme_controller", "transport": "pcie"}, - ], - }, - })) - - def test_get_nvme_stas_controller_directives__tcp_controller(self): - expected = {"controller = transport=tcp;traddr=1.2.3.4;trsvcid=1111"} - - result = curthooks.get_nvme_stas_controller_directives({ - "storage": { - "config": [ - { - "type": "nvme_controller", - "transport": "tcp", - "tcp_addr": "1.2.3.4", - "tcp_port": "1111", - }, - ], - }, - }) - self.assertEqual(expected, result) - - def test_get_nvme_stas_controller_directives__three_nvme_controllers(self): - expected = {"controller = transport=tcp;traddr=1.2.3.4;trsvcid=1111", - "controller = transport=tcp;traddr=4.5.6.7;trsvcid=1212"} - - result = curthooks.get_nvme_stas_controller_directives({ - "storage": { - "config": [ - { - "type": "nvme_controller", - "transport": "tcp", - "tcp_addr": "1.2.3.4", - "tcp_port": "1111", - }, { - "type": "nvme_controller", - "transport": "tcp", - "tcp_addr": "4.5.6.7", - "tcp_port": "1212", - }, { - "type": "nvme_controller", - "transport": "pcie", - }, - ], - }, - }) - self.assertEqual(expected, result) - - def test_get_nvme_stas_controller_directives__empty_conf(self): - self.assertFalse(curthooks.get_nvme_stas_controller_directives({})) - self.assertFalse(curthooks.get_nvme_stas_controller_directives( - {"storage": False})) - self.assertFalse(curthooks.get_nvme_stas_controller_directives( - {"storage": {}})) - self.assertFalse(curthooks.get_nvme_stas_controller_directives({ - "storage": { - "config": "disabled", - }, - })) - - def test_nvmeotcp_get_nvme_commands__no_nvme_controller(self): - self.assertFalse(curthooks.nvmeotcp_get_nvme_commands({ - "storage": { - "config": [ - {"type": "partition"}, - {"type": "mount"}, - {"type": "disk"}, - ], - }, - })) - - def test_nvmeotcp_get_nvme_commands__pcie_controller(self): - self.assertFalse(curthooks.nvmeotcp_get_nvme_commands({ - "storage": { - "config": [ - {"type": "nvme_controller", "transport": "pcie"}, - ], - }, - })) - - def test_nvmeotcp_get_nvme_commands__tcp_controller(self): - expected = [( + def test_no_nvme_controller(self): + with patch('curtin.block.nvme.get_nvme_controllers_from_config', + return_value=[]): + self.assertFalse( + curthooks.get_nvme_stas_controller_directives(None)) + self.assertFalse(curthooks.nvmeotcp_get_nvme_commands(None)) + + def test_pcie_controller(self): + controllers = [{'type': 'nvme_controller', 'transport': 'pcie'}] + with patch('curtin.block.nvme.get_nvme_controllers_from_config', + return_value=controllers): + self.assertFalse( + curthooks.get_nvme_stas_controller_directives(None)) + self.assertFalse(curthooks.nvmeotcp_get_nvme_commands(None)) + + def test_tcp_controller(self): + stas_expected = { + 'controller = transport=tcp;traddr=1.2.3.4;trsvcid=1111', + } + cmds_expected = [( "nvme", "connect-all", "--transport", "tcp", "--traddr", "1.2.3.4", "--trsvcid", "1111", ), ] - - result = curthooks.nvmeotcp_get_nvme_commands({ - "storage": { - "config": [ - { - "type": "nvme_controller", - "transport": "tcp", - "tcp_addr": "1.2.3.4", - "tcp_port": "1111", - }, - ], - }, - }) - self.assertEqual(expected, result) - - def test_nvmeotcp_get_nvme_commands__three_nvme_controllers(self): - expected = [( + controllers = [{ + "type": "nvme_controller", + "transport": "tcp", + "tcp_addr": "1.2.3.4", + "tcp_port": "1111", + }] + + with patch('curtin.block.nvme.get_nvme_controllers_from_config', + return_value=controllers): + stas_result = curthooks.get_nvme_stas_controller_directives(None) + cmds_result = curthooks.nvmeotcp_get_nvme_commands(None) + self.assertEqual(stas_expected, stas_result) + self.assertEqual(cmds_expected, cmds_result) + + def test_three_nvme_controllers(self): + stas_expected = { + "controller = transport=tcp;traddr=1.2.3.4;trsvcid=1111", + "controller = transport=tcp;traddr=4.5.6.7;trsvcid=1212", + } + cmds_expected = [( "nvme", "connect-all", "--transport", "tcp", "--traddr", "1.2.3.4", @@ -2163,40 +2089,29 @@ class TestCurthooksNVMeOverTCP(CiTestCase): "--trsvcid", "1212", ), ] - - result = curthooks.nvmeotcp_get_nvme_commands({ - "storage": { - "config": [ - { - "type": "nvme_controller", - "transport": "tcp", - "tcp_addr": "1.2.3.4", - "tcp_port": "1111", - }, { - "type": "nvme_controller", - "transport": "tcp", - "tcp_addr": "4.5.6.7", - "tcp_port": "1212", - }, { - "type": "nvme_controller", - "transport": "pcie", - }, - ], + controllers = [ + { + "type": "nvme_controller", + "transport": "tcp", + "tcp_addr": "1.2.3.4", + "tcp_port": "1111", + }, { + "type": "nvme_controller", + "transport": "tcp", + "tcp_addr": "4.5.6.7", + "tcp_port": "1212", + }, { + "type": "nvme_controller", + "transport": "pcie", }, - }) - self.assertEqual(expected, result) + ] - def test_nvmeotcp_get_nvme_commands__empty_conf(self): - self.assertFalse(curthooks.nvmeotcp_get_nvme_commands({})) - self.assertFalse(curthooks.nvmeotcp_get_nvme_commands( - {"storage": False})) - self.assertFalse(curthooks.nvmeotcp_get_nvme_commands( - {"storage": {}})) - self.assertFalse(curthooks.nvmeotcp_get_nvme_commands({ - "storage": { - "config": "disabled", - }, - })) + with patch('curtin.block.nvme.get_nvme_controllers_from_config', + return_value=controllers): + stas_result = curthooks.get_nvme_stas_controller_directives(None) + cmds_result = curthooks.nvmeotcp_get_nvme_commands(None) + self.assertEqual(stas_expected, stas_result) + self.assertEqual(cmds_expected, cmds_result) def test_nvmeotcp_get_ip_commands__ethernet_static(self): netcfg = """\ |