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
|
import os
import gc
import unittest
import copy
import mock
import mozharness.base.log as log
from mozharness.base.log import ERROR
import mozharness.base.script as script
from mozharness.mozilla.blob_upload import BlobUploadMixin, \
blobupload_config_options
class CleanupObj(script.ScriptMixin, log.LogMixin):
def __init__(self):
super(CleanupObj, self).__init__()
self.log_obj = None
self.config = {'log_level': ERROR}
def cleanup():
gc.collect()
c = CleanupObj()
for f in ('test_logs', 'test_dir', 'tmpfile_stdout', 'tmpfile_stderr'):
c.rmtree(f)
class BlobUploadScript(BlobUploadMixin, script.BaseScript):
config_options = copy.deepcopy(blobupload_config_options)
def __init__(self, **kwargs):
self.abs_dirs = None
self.set_buildbot_property = mock.Mock()
super(BlobUploadScript, self).__init__(
config_options=self.config_options,
**kwargs
)
def query_python_path(self, binary="python"):
if binary == "blobberc.py":
return mock.Mock(return_value='/path/to/blobberc').return_value
elif binary == "python":
return mock.Mock(return_value='/path/to/python').return_value
def query_abs_dirs(self):
if self.abs_dirs:
return self.abs_dirs
abs_dirs = super(BlobUploadScript, self).query_abs_dirs()
dirs = {}
dirs['abs_blob_upload_dir'] = os.path.join(abs_dirs['abs_work_dir'],
'blobber_upload_dir')
abs_dirs.update(dirs)
self.abs_dirs = abs_dirs
return self.abs_dirs
def run_command(self, command):
self.command = command
# TestBlobUploadMechanism {{{1
class TestBlobUploadMechanism(unittest.TestCase):
# I need a log watcher helper function, here and in test_log.
def setUp(self):
cleanup()
self.s = None
def tearDown(self):
# Close the logfile handles, or windows can't remove the logs
if hasattr(self, 's') and isinstance(self.s, object):
del(self.s)
cleanup()
def test_blob_upload_mechanism(self):
self.s = BlobUploadScript(config={'log_type': 'multi',
'blob_upload_branch': 'test-branch',
'default_blob_upload_servers':
['http://blob_server.me'],
'blob_uploader_auth_file':
os.path.abspath(__file__)},
initial_config_file='test/test.json')
content = "Hello world!"
parent_dir = self.s.query_abs_dirs()['abs_blob_upload_dir']
if not os.path.isdir(parent_dir):
self.s.mkdir_p(parent_dir)
file_name = os.path.join(parent_dir, 'test_mock_blob_file')
self.s.write_to_file(file_name, content)
self.s.upload_blobber_files()
self.assertTrue(self.s.set_buildbot_property.called)
expected_result = ['/path/to/python', '/path/to/blobberc', '-u',
'http://blob_server.me', '-a',
os.path.abspath(__file__), '-b', 'test-branch', '-d']
expected_result.append(self.s.query_abs_dirs()['abs_blob_upload_dir'])
expected_result += [
'--output-manifest',
os.path.join(self.s.query_abs_dirs()['abs_work_dir'], "uploaded_files.json")
]
self.assertEqual(expected_result, self.s.command)
# main {{{1
if __name__ == '__main__':
unittest.main()
|
