testing/mozharness/test/test_mozilla_blob_upload.py

   1 import os
   2 import gc
   3 import unittest
   4 import copy
   5 import mock
   6
   7 import mozharness.base.log as log
   8 from mozharness.base.log import ERROR
   9 import mozharness.base.script as script
  10 from mozharness.mozilla.blob_upload import BlobUploadMixin, \
  11     blobupload_config_options
  12
  13 class 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
  20 def 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
  27 class 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
  59 class 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
 102 if __name__ == '__main__':
 103     unittest.main()