File: loop_device_test.py

package info (click to toggle)
kiwi 10.2.36-1
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 7,664 kB
  • sloc: python: 69,179; sh: 4,228; xml: 3,383; ansic: 391; makefile: 353
file content (132 lines) | stat: -rw-r--r-- 4,550 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import logging
from unittest.mock import (
    patch, call
)
from pytest import (
    raises, fixture
)

from kiwi.storage.loop_device import LoopDevice

from kiwi.exceptions import (
    KiwiLoopSetupError,
    KiwiCommandError
)


class TestLoopDevice:
    @fixture(autouse=True)
    def inject_fixtures(self, caplog):
        self._caplog = caplog

    @patch('os.path.exists')
    def setup(self, mock_exists):
        mock_exists.return_value = False
        self.loop = LoopDevice('loop-file', 20, 4096)

    @patch('os.path.exists')
    def setup_method(self, cls, mock_exists):
        self.setup()

    def test_loop_setup_invalid(self):
        with raises(KiwiLoopSetupError):
            LoopDevice('loop-file-does-not-exist-and-no-size-given')

    def test_get_device(self):
        assert self.loop.get_device() == ''

    def test_is_loop(self):
        assert self.loop.is_loop() is True

    @patch('os.path.exists')
    @patch('kiwi.storage.loop_device.Command.run')
    @patch('kiwi.storage.loop_device.CommandCapabilities.has_option_in_help')
    def test_create(
        self, mock_has_option_in_help, mock_command, mock_exists
    ):
        mock_has_option_in_help.return_value = True
        mock_exists.return_value = False
        self.loop.create()
        call = mock_command.call_args_list[0]
        assert mock_command.call_args_list[0] == \
            call([
                'qemu-img', 'create', 'loop-file', '20M'
            ])
        call = mock_command.call_args_list[1]
        assert mock_command.call_args_list[1] == \
            call([
                'losetup', '--sector-size', '4096',
                '-f', '--show', 'loop-file'
            ])
        mock_has_option_in_help.return_value = False
        mock_command.reset_mock()
        self.loop.create()
        assert mock_command.call_args_list[1] == \
            call([
                'losetup', '--logical-blocksize', '4096',
                '-f', '--show', 'loop-file'
            ])
        self.loop.node_name = None

    @patch('kiwi.storage.loop_device.Command.run')
    @patch('os.path.exists')
    @patch('pathlib.Path.is_block_device')
    @patch('time.sleep')
    def test_context_manager_exit_loop_released(
        self, mock_time_sleep, mock_is_block_device,
        mock_os_path_exists, mock_command_run
    ):
        is_block_device = [False, True]
        mock_os_path_exists.return_value = True
        mock_command_run.side_effect = KiwiCommandError('error')

        def Command_run(params):
            # raise on first command which is 'losetup -f ...'
            if params[1] == '-f':
                raise KiwiCommandError('issue')

        def Path_is_block_device():
            return is_block_device.pop()

        mock_is_block_device.side_effect = Path_is_block_device
        mock_command_run.side_effect = Command_run

        with self._caplog.at_level(logging.ERROR):
            with LoopDevice('loop-file', 20) as loop_provider:
                loop_provider.node_name = '/dev/loop0'
                with raises(KiwiCommandError):
                    loop_provider.create(overwrite=False)
            assert len(mock_is_block_device.call_args_list) == 2
            assert mock_command_run.call_args_list == [
                call(['losetup', '-f', '--show', 'loop-file']),
                call(['losetup', '-d', '/dev/loop0'])
            ]

    @patch('kiwi.storage.loop_device.Command.run')
    @patch('os.path.exists')
    @patch('pathlib.Path.is_block_device')
    @patch('time.sleep')
    def test_context_manager_exit_loop_not_released(
        self, mock_time_sleep, mock_is_block_device,
        mock_os_path_exists, mock_command_run
    ):
        mock_os_path_exists.return_value = True
        mock_command_run.side_effect = KiwiCommandError('error')
        mock_is_block_device.return_value = True

        def Command_run(params):
            # raise on first command which is 'losetup -f ...'
            if params[1] == '-f':
                raise KiwiCommandError('issue')

        mock_command_run.side_effect = Command_run

        with self._caplog.at_level(logging.ERROR):
            with LoopDevice('loop-file', 20) as loop_provider:
                loop_provider.node_name = '/dev/loop0'
                with raises(KiwiCommandError):
                    loop_provider.create(overwrite=False)
            assert mock_command_run.call_args_list == [
                call(['losetup', '-f', '--show', 'loop-file']),
                call(['losetup', '-d', '/dev/loop0'])
            ]