142
143
class TestSubp(TestCase):
145
stdin2err = ['bash', '-c', 'cat >&2']
147
utf8_invalid = b'ab\xaadef'
148
utf8_valid = b'start \xc3\xa9 end'
149
utf8_valid_2 = b'd\xc3\xa9j\xc8\xa7'
151
def printf_cmd(self, *args):
152
# bash's printf supports \xaa. So does /usr/bin/printf
153
# but by using bash, we remove dependency on another program.
154
return(['bash', '-c', 'printf "$@"', 'printf'] + list(args))
144
156
def test_subp_handles_utf8(self):
145
157
# The given bytes contain utf-8 accented characters as seen in e.g.
146
158
# the "deja dup" package in Ubuntu.
147
input_bytes = b'd\xc3\xa9j\xc8\xa7'
148
cmd = ['echo', '-n', input_bytes]
159
cmd = self.printf_cmd(self.utf8_valid_2)
149
160
(out, _err) = util.subp(cmd, capture=True)
150
self.assertEqual(out, input_bytes.decode('utf-8'))
161
self.assertEqual(out, self.utf8_valid_2.decode('utf-8'))
163
def test_subp_respects_decode_false(self):
164
(out, err) = util.subp(self.stdin2out, capture=True, decode=False,
165
data=self.utf8_valid)
166
self.assertTrue(isinstance(out, bytes))
167
self.assertTrue(isinstance(err, bytes))
168
self.assertEqual(out, self.utf8_valid)
170
def test_subp_decode_ignore(self):
171
# this executes a string that writes invalid utf-8 to stdout
172
(out, _err) = util.subp(self.printf_cmd('abc\\xaadef'),
173
capture=True, decode='ignore')
174
self.assertEqual(out, 'abcdef')
176
def test_subp_decode_strict_valid_utf8(self):
177
(out, _err) = util.subp(self.stdin2out, capture=True,
178
decode='strict', data=self.utf8_valid)
179
self.assertEqual(out, self.utf8_valid.decode('utf-8'))
181
def test_subp_decode_invalid_utf8_replaces(self):
182
(out, _err) = util.subp(self.stdin2out, capture=True,
183
data=self.utf8_invalid)
184
expected = self.utf8_invalid.decode('utf-8', errors='replace')
185
self.assertEqual(out, expected)
187
def test_subp_decode_strict_raises(self):
189
kwargs = {'args': self.stdin2out, 'capture': True,
190
'decode': 'strict', 'data': self.utf8_invalid}
191
self.assertRaises(UnicodeDecodeError, util.subp, *args, **kwargs)
193
def test_subp_capture_stderr(self):
194
data = b'hello world'
195
(out, err) = util.subp(self.stdin2err, capture=True,
196
decode=False, data=data)
197
self.assertEqual(err, data)
198
self.assertEqual(out, b'')
200
def test_returns_none_if_no_capture(self):
201
(out, err) = util.subp(self.stdin2out, data=b'')
202
self.assertEqual(err, None)
203
self.assertEqual(out, None)
206
class TestHuman2Bytes(TestCase):
207
GB = 1024 * 1024 * 1024
210
def test_float_equal_int_is_allowed(self):
211
self.assertEqual(1000, util.human2bytes(1000.0))
213
def test_float_in_string_nonequal_int_raises_type_error(self):
214
self.assertRaises(ValueError, util.human2bytes, "1000.4B")
216
def test_float_nonequal_int_raises_type_error(self):
217
self.assertRaises(ValueError, util.human2bytes, 1000.4)
219
def test_int_gets_int(self):
220
self.assertEqual(100, util.human2bytes(100))
222
def test_no_suffix_is_bytes(self):
223
self.assertEqual(100, util.human2bytes("100"))
225
def test_suffix_M(self):
226
self.assertEqual(100 * self.MB, util.human2bytes("100M"))
228
def test_suffix_B(self):
229
self.assertEqual(100, util.human2bytes("100B"))
231
def test_suffix_G(self):
232
self.assertEqual(int(10 * self.GB), util.human2bytes("10G"))
234
def test_float_in_string(self):
235
self.assertEqual(int(3.5 * self.GB), util.human2bytes("3.5G"))
237
def test_GB_equals_G(self):
238
self.assertEqual(util.human2bytes("3GB"), util.human2bytes("3G"))
241
class TestSetUnExecutable(TestCase):
247
if os.path.exists(self.tmpf):
251
shutil.rmtree(self.tmpd)
254
def tempfile(self, data=None):
255
fp, self.tmpf = tempfile.mkstemp()
261
def test_change_needed_returns_original_mode(self):
262
tmpf = self.tempfile()
263
os.chmod(tmpf, 0o755)
264
ret = util.set_unexecutable(tmpf)
265
self.assertEqual(ret, 0o0755)
267
def test_no_change_needed_returns_none(self):
268
tmpf = self.tempfile()
269
os.chmod(tmpf, 0o600)
270
ret = util.set_unexecutable(tmpf)
271
self.assertEqual(ret, None)
273
def test_change_does_as_expected(self):
274
tmpf = self.tempfile()
275
os.chmod(tmpf, 0o755)
276
ret = util.set_unexecutable(tmpf)
277
self.assertEqual(ret, 0o0755)
278
self.assertEqual(stat.S_IMODE(os.stat(tmpf).st_mode), 0o0644)
280
def test_strict_no_exists_raises_exception(self):
281
self.tmpd = tempfile.mkdtemp()
282
bogus = os.path.join(self.tmpd, 'bogus')
283
self.assertRaises(ValueError, util.set_unexecutable, bogus, True)
152
285
# vi: ts=4 expandtab syntax=python