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