1import os
2import gc
3import unittest
4import copy
5import mock
6
7import mozharness.base.log as log
8from mozharness.base.log import ERROR
9import mozharness.base.script as script
10from mozharness.mozilla.blob_upload import BlobUploadMixin, \
11    blobupload_config_options
12
13class CleanupObj(script.ScriptMixin, log.LogMixin):
14    def __init__(self):
15        super(CleanupObj, self).__init__()
16        self.log_obj = None
17        self.config = {'log_level': ERROR}
18
19
20def cleanup():
21    gc.collect()
22    c = CleanupObj()
23    for f in ('test_logs', 'test_dir', 'tmpfile_stdout', 'tmpfile_stderr'):
24        c.rmtree(f)
25
26
27class BlobUploadScript(BlobUploadMixin, script.BaseScript):
28    config_options = copy.deepcopy(blobupload_config_options)
29    def __init__(self, **kwargs):
30        self.abs_dirs = None
31        self.set_buildbot_property = mock.Mock()
32        super(BlobUploadScript, self).__init__(
33                config_options=self.config_options,
34                **kwargs
35        )
36
37    def query_python_path(self, binary="python"):
38        if binary == "blobberc.py":
39            return mock.Mock(return_value='/path/to/blobberc').return_value
40        elif binary == "python":
41            return mock.Mock(return_value='/path/to/python').return_value
42
43    def query_abs_dirs(self):
44        if self.abs_dirs:
45            return self.abs_dirs
46        abs_dirs = super(BlobUploadScript, self).query_abs_dirs()
47        dirs = {}
48        dirs['abs_blob_upload_dir'] = os.path.join(abs_dirs['abs_work_dir'],
49                                                   'blobber_upload_dir')
50        abs_dirs.update(dirs)
51        self.abs_dirs = abs_dirs
52
53        return self.abs_dirs
54
55    def run_command(self, command):
56        self.command = command
57
58# TestBlobUploadMechanism {{{1
59class TestBlobUploadMechanism(unittest.TestCase):
60    # I need a log watcher helper function, here and in test_log.
61    def setUp(self):
62        cleanup()
63        self.s = None
64
65    def tearDown(self):
66        # Close the logfile handles, or windows can't remove the logs
67        if hasattr(self, 's') and isinstance(self.s, object):
68            del(self.s)
69        cleanup()
70
71    def test_blob_upload_mechanism(self):
72        self.s = BlobUploadScript(config={'log_type': 'multi',
73                                          'blob_upload_branch': 'test-branch',
74                                          'default_blob_upload_servers':
75                                            ['http://blob_server.me'],
76                                          'blob_uploader_auth_file':
77                                            os.path.abspath(__file__)},
78                                  initial_config_file='test/test.json')
79
80        content = "Hello world!"
81        parent_dir = self.s.query_abs_dirs()['abs_blob_upload_dir']
82        if not os.path.isdir(parent_dir):
83            self.s.mkdir_p(parent_dir)
84
85        file_name = os.path.join(parent_dir, 'test_mock_blob_file')
86        self.s.write_to_file(file_name, content)
87        self.s.upload_blobber_files()
88        self.assertTrue(self.s.set_buildbot_property.called)
89
90        expected_result = ['/path/to/python', '/path/to/blobberc', '-u',
91                           'http://blob_server.me', '-a',
92                           os.path.abspath(__file__), '-b', 'test-branch', '-d']
93        expected_result.append(self.s.query_abs_dirs()['abs_blob_upload_dir'])
94        expected_result += [
95                '--output-manifest',
96                os.path.join(self.s.query_abs_dirs()['abs_work_dir'], "uploaded_files.json")
97        ]
98        self.assertEqual(expected_result, self.s.command)
99
100
101# main {{{1
102if __name__ == '__main__':
103    unittest.main()
104