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
133
|
import os.path
import shutil
import tempfile
from cloudinit.config import cc_mounts
from .. import helpers as test_helpers
try:
from unittest import mock
except ImportError:
import mock
class TestSanitizeDevname(test_helpers.FilesystemMockingTestCase):
def setUp(self):
super(TestSanitizeDevname, self).setUp()
self.new_root = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.new_root)
self.patchOS(self.new_root)
def _touch(self, path):
path = os.path.join(self.new_root, path.lstrip('/'))
basedir = os.path.dirname(path)
if not os.path.exists(basedir):
os.makedirs(basedir)
open(path, 'a').close()
def _makedirs(self, directory):
directory = os.path.join(self.new_root, directory.lstrip('/'))
if not os.path.exists(directory):
os.makedirs(directory)
def mock_existence_of_disk(self, disk_path):
self._touch(disk_path)
self._makedirs(os.path.join('/sys/block', disk_path.split('/')[-1]))
def mock_existence_of_partition(self, disk_path, partition_number):
self.mock_existence_of_disk(disk_path)
self._touch(disk_path + str(partition_number))
disk_name = disk_path.split('/')[-1]
self._makedirs(os.path.join('/sys/block',
disk_name,
disk_name + str(partition_number)))
def test_existent_full_disk_path_is_returned(self):
disk_path = '/dev/sda'
self.mock_existence_of_disk(disk_path)
self.assertEqual(disk_path,
cc_mounts.sanitize_devname(disk_path,
lambda x: None,
mock.Mock()))
def test_existent_disk_name_returns_full_path(self):
disk_name = 'sda'
disk_path = '/dev/' + disk_name
self.mock_existence_of_disk(disk_path)
self.assertEqual(disk_path,
cc_mounts.sanitize_devname(disk_name,
lambda x: None,
mock.Mock()))
def test_existent_meta_disk_is_returned(self):
actual_disk_path = '/dev/sda'
self.mock_existence_of_disk(actual_disk_path)
self.assertEqual(
actual_disk_path,
cc_mounts.sanitize_devname('ephemeral0',
lambda x: actual_disk_path,
mock.Mock()))
def test_existent_meta_partition_is_returned(self):
disk_name, partition_part = '/dev/sda', '1'
actual_partition_path = disk_name + partition_part
self.mock_existence_of_partition(disk_name, partition_part)
self.assertEqual(
actual_partition_path,
cc_mounts.sanitize_devname('ephemeral0.1',
lambda x: disk_name,
mock.Mock()))
def test_existent_meta_partition_with_p_is_returned(self):
disk_name, partition_part = '/dev/sda', 'p1'
actual_partition_path = disk_name + partition_part
self.mock_existence_of_partition(disk_name, partition_part)
self.assertEqual(
actual_partition_path,
cc_mounts.sanitize_devname('ephemeral0.1',
lambda x: disk_name,
mock.Mock()))
def test_first_partition_returned_if_existent_disk_is_partitioned(self):
disk_name, partition_part = '/dev/sda', '1'
actual_partition_path = disk_name + partition_part
self.mock_existence_of_partition(disk_name, partition_part)
self.assertEqual(
actual_partition_path,
cc_mounts.sanitize_devname('ephemeral0',
lambda x: disk_name,
mock.Mock()))
def test_nth_partition_returned_if_requested(self):
disk_name, partition_part = '/dev/sda', '3'
actual_partition_path = disk_name + partition_part
self.mock_existence_of_partition(disk_name, partition_part)
self.assertEqual(
actual_partition_path,
cc_mounts.sanitize_devname('ephemeral0.3',
lambda x: disk_name,
mock.Mock()))
def test_transformer_returning_none_returns_none(self):
self.assertIsNone(
cc_mounts.sanitize_devname(
'ephemeral0', lambda x: None, mock.Mock()))
def test_missing_device_returns_none(self):
self.assertIsNone(
cc_mounts.sanitize_devname('/dev/sda', None, mock.Mock()))
def test_missing_sys_returns_none(self):
disk_path = '/dev/sda'
self._makedirs(disk_path)
self.assertIsNone(
cc_mounts.sanitize_devname(disk_path, None, mock.Mock()))
def test_existent_disk_but_missing_partition_returns_none(self):
disk_path = '/dev/sda'
self.mock_existence_of_disk(disk_path)
self.assertIsNone(
cc_mounts.sanitize_devname(
'ephemeral0.1', lambda x: disk_path, mock.Mock()))
|
