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
|
from __future__ import absolute_import
import os.path
import subprocess
import tempfile
import unittest2
import mock
import ansible_mitogen.target
import testlib
LOGGER_NAME = ansible_mitogen.target.LOG.name
class NamedTemporaryDirectory(object):
def __init__(self, **kwargs):
self.kwargs = kwargs
def __enter__(self):
self.path = tempfile.mkdtemp(**self.kwargs)
return self.path
def __exit__(self, _1, _2, _3):
subprocess.check_call(['rm', '-rf', self.path])
class FindGoodTempDirTest(testlib.TestCase):
func = staticmethod(ansible_mitogen.target.find_good_temp_dir)
def test_expands_usernames(self):
with NamedTemporaryDirectory(
prefix='.ansible_mitogen_test',
dir=os.environ['HOME']
) as tmpdir:
path = self.func(['~'])
self.assertTrue(path.startswith(os.environ['HOME']))
def test_expands_vars(self):
with NamedTemporaryDirectory(
prefix='.ansible_mitogen_test',
dir=os.environ['HOME']
) as tmpdir:
os.environ['somevar'] = 'xyz'
path = self.func([tmpdir + '/$somevar'])
self.assertTrue(path.startswith('%s/%s' % (tmpdir, 'xyz')))
@mock.patch('ansible_mitogen.target.is_good_temp_dir')
def test_no_good_candidate(self, is_good_temp_dir):
is_good_temp_dir.return_value = False
e = self.assertRaises(IOError,
lambda: self.func([])
)
self.assertTrue(str(e).startswith('Unable to find a useable'))
class ApplyModeSpecTest(unittest2.TestCase):
func = staticmethod(ansible_mitogen.target.apply_mode_spec)
def test_simple(self):
spec = 'u+rwx,go=x'
self.assertEquals(int('0711', 8), self.func(spec, 0))
spec = 'g-rw'
self.assertEquals(int('0717', 8), self.func(spec, int('0777', 8)))
class IsGoodTempDirTest(unittest2.TestCase):
func = staticmethod(ansible_mitogen.target.is_good_temp_dir)
def test_creates(self):
with NamedTemporaryDirectory() as temp_path:
bleh = os.path.join(temp_path, 'bleh')
self.assertFalse(os.path.exists(bleh))
self.assertTrue(self.func(bleh))
self.assertTrue(os.path.exists(bleh))
def test_file_exists(self):
with NamedTemporaryDirectory() as temp_path:
bleh = os.path.join(temp_path, 'bleh')
with open(bleh, 'w') as fp:
fp.write('derp')
self.assertTrue(os.path.isfile(bleh))
self.assertFalse(self.func(bleh))
self.assertEquals(open(bleh).read(), 'derp')
@unittest2.skipIf(os.geteuid() == 0, 'root can always write to directories')
def test_unwriteable(self):
with NamedTemporaryDirectory() as temp_path:
os.chmod(temp_path, 0)
self.assertFalse(self.func(temp_path))
os.chmod(temp_path, int('0700', 8))
@mock.patch('os.chmod')
def test_weird_filesystem(self, os_chmod):
os_chmod.side_effect = OSError('nope')
with NamedTemporaryDirectory() as temp_path:
self.assertFalse(self.func(temp_path))
@mock.patch('os.access')
def test_noexec(self, os_access):
os_access.return_value = False
with NamedTemporaryDirectory() as temp_path:
self.assertFalse(self.func(temp_path))
if __name__ == '__main__':
unittest2.main()
|
