# Use of this source code is governed under the Apache License, Version 2.0 that
# can be found in the LICENSE file.
+import getpass
import hashlib
import json
import logging
import os
+import shutil
import StringIO
import sys
+import tempfile
import threading
import unittest
from depot_tools import auto_stub
import swarming
+import test_utils
from utils import net
self.requests = []
self.mock(swarming.net, 'url_open', self._url_open)
self.mock(swarming.time, 'sleep', lambda x: None)
+ self.mock(swarming.subprocess, 'call', lambda *_: self.fail())
self.mock(sys, 'stdout', StringIO.StringIO())
self.mock(sys, 'stderr', StringIO.StringIO())
'')
-def chromium_tasks(retrieval_url):
+def chromium_tasks(retrieval_url, file_hash):
return [
{
u'action': [
u'python', u'run_isolated.zip',
- u'--hash', FILE_HASH,
+ u'--hash', file_hash,
+ u'--namespace', u'default-gzip',
u'--isolate-server', retrieval_url,
],
u'decorate_output': False,
env,
working_dir,
isolate_server,
- profile):
+ profile,
+ test_case_name=TEST_NAME,
+ file_hash=FILE_HASH):
expected = {
u'cleanup': u'root',
u'configurations': [
u'encoding': u'UTF-8',
u'env_vars': env.copy(),
u'restart_on_failure': True,
- u'test_case_name': TEST_NAME,
- u'tests': chromium_tasks(isolate_server),
+ u'test_case_name': test_case_name,
+ u'tests': chromium_tasks(isolate_server, file_hash),
u'working_dir': unicode(working_dir),
}
if shards > 1:
dimensions = {'os': 'Windows'}
manifest = swarming.Manifest(
isolate_server='http://localhost:8081',
+ namespace='default-gzip',
isolated_hash=FILE_HASH,
task_name=TEST_NAME,
shards=2,
dimensions = {'os': 'Linux'}
manifest = swarming.Manifest(
isolate_server='http://localhost:8081',
+ namespace='default-gzip',
isolated_hash=FILE_HASH,
task_name=TEST_NAME,
shards=1,
dimensions = {'os': 'Linux'}
manifest = swarming.Manifest(
isolate_server='http://localhost:8081',
+ namespace='default-gzip',
isolated_hash=FILE_HASH,
task_name=TEST_NAME,
shards=1,
result = swarming.process_manifest(
swarming='http://localhost:8082',
isolate_server='http://localhost:8081',
+ namespace='default',
isolated_hash=FILE_HASH,
task_name=TEST_NAME,
shards=1,
result = swarming.process_manifest(
swarming='http://localhost:8082',
isolate_server='http://localhost:8081',
+ namespace='default',
isolated_hash=FILE_HASH,
task_name=TEST_NAME,
shards=1,
algo=ALGO)
self.assertEqual(0, result)
+ def test_isolated_to_hash(self):
+ calls = []
+ self.mock(swarming.subprocess, 'call', lambda *c: calls.append(c))
+ content = '{}'
+ expected_hash = hashlib.sha1(content).hexdigest()
+ handle, isolated = tempfile.mkstemp(
+ prefix='swarming_test_', suffix='.isolated')
+ os.close(handle)
+ try:
+ with open(isolated, 'w') as f:
+ f.write(content)
+ hash_value, is_file = swarming.isolated_to_hash(
+ 'http://localhost:1', 'default', isolated, hashlib.sha1, False)
+ finally:
+ os.remove(isolated)
+ self.assertEqual(expected_hash, hash_value)
+ self.assertEqual(True, is_file)
+ expected_calls = [
+ (
+ [
+ sys.executable,
+ os.path.join(ROOT_DIR, 'isolate.py'),
+ 'archive',
+ '--isolate-server', 'http://localhost:1',
+ '--namespace', 'default',
+ '--isolated',
+ isolated,
+ ],
+ False,
+ ),
+ ]
+ self.assertEqual(expected_calls, calls)
+ self._check_output('Archiving: %s\n' % isolated, '')
+
class MainTest(TestCase):
+ def setUp(self):
+ super(MainTest, self).setUp()
+ self._tmpdir = None
+
+ def tearDown(self):
+ try:
+ if self._tmpdir:
+ shutil.rmtree(self._tmpdir)
+ finally:
+ super(MainTest, self).tearDown()
+
+ @property
+ def tmpdir(self):
+ if not self._tmpdir:
+ self._tmpdir = tempfile.mkdtemp(prefix='swarming')
+ return self._tmpdir
+
+ def test_run_hash(self):
+ self.mock(swarming.isolateserver, 'get_storage',
+ lambda *_: MockedStorage(warm_cache=False))
+
+ task_name = '%s/foo=bar_os=Mac/1111111111111111111111111111111111111111' % (
+ getpass.getuser())
+ j = generate_expected_json(
+ shards=1,
+ dimensions={'foo': 'bar', 'os': 'Mac'},
+ env={},
+ working_dir='swarm_tests',
+ isolate_server='https://host2',
+ profile=False,
+ test_case_name=task_name)
+ j['data'] = [['http://localhost:8081/fetch_url', 'swarm_data.zip']]
+ data = {
+ 'request': json.dumps(j, sort_keys=True, separators=(',',':')),
+ }
+ self.requests = [
+ (
+ 'https://host1/test',
+ {'data': data},
+ # The actual output is ignored as long as it is valid json.
+ StringIO.StringIO('{}'),
+ ),
+ ]
+ ret = main([
+ 'trigger',
+ '--swarming', 'https://host1',
+ '--isolate-server', 'https://host2',
+ '--shards', '1',
+ '--priority', '101',
+ '--dimension', 'foo', 'bar',
+ '--dimension', 'os', 'Mac',
+ FILE_HASH,
+ ])
+ actual = sys.stdout.getvalue()
+ self.assertEqual(0, ret, (actual, sys.stderr.getvalue()))
+ self._check_output('Triggered task: %s\n' % task_name, '')
+
+ def test_run_isolated(self):
+ self.mock(swarming.isolateserver, 'get_storage',
+ lambda *_: MockedStorage(warm_cache=False))
+ calls = []
+ self.mock(swarming.subprocess, 'call', lambda *c: calls.append(c))
+
+ isolated = os.path.join(self.tmpdir, 'zaz.isolated')
+ content = '{}'
+ with open(isolated, 'wb') as f:
+ f.write(content)
+
+ isolated_hash = ALGO(content).hexdigest()
+ task_name = 'zaz/foo=bar_os=Mac/%s' % isolated_hash
+ j = generate_expected_json(
+ shards=1,
+ dimensions={'foo': 'bar', 'os': 'Mac'},
+ env={},
+ working_dir='swarm_tests',
+ isolate_server='https://host2',
+ profile=False,
+ test_case_name=task_name,
+ file_hash=isolated_hash)
+ j['data'] = [['http://localhost:8081/fetch_url', 'swarm_data.zip']]
+ data = {
+ 'request': json.dumps(j, sort_keys=True, separators=(',',':')),
+ }
+ self.requests = [
+ (
+ 'https://host1/test',
+ {'data': data},
+ # The actual output is ignored as long as it is valid json.
+ StringIO.StringIO('{}'),
+ ),
+ ]
+ ret = main([
+ 'trigger',
+ '--swarming', 'https://host1',
+ '--isolate-server', 'https://host2',
+ '--shards', '1',
+ '--priority', '101',
+ '--dimension', 'foo', 'bar',
+ '--dimension', 'os', 'Mac',
+ isolated,
+ ])
+ actual = sys.stdout.getvalue()
+ self.assertEqual(0, ret, (actual, sys.stderr.getvalue()))
+ expected = [
+ (
+ [
+ sys.executable,
+ os.path.join(ROOT_DIR, 'isolate.py'), 'archive',
+ '--isolate-server', 'https://host2',
+ '--namespace' ,'default-gzip',
+ '--isolated', isolated,
+ ],
+ 0),
+ ]
+ self.assertEqual(expected, calls)
+ expected = 'Archiving: %s\nTriggered task: %s\n' % (isolated, task_name)
+ self._check_output(expected, '')
+
def test_trigger_no_request(self):
with self.assertRaises(SystemExit):
main([
'trigger', '--swarming', 'https://host',
- '--isolate-server', 'https://host',
+ '--isolate-server', 'https://host', '-T', 'foo',
])
self._check_output(
'',
'swarming.py: error: Must pass one .isolated file or its hash (sha1).'
'\n')
+ def test_trigger_no_env_vars(self):
+ with self.assertRaises(SystemExit):
+ main(['trigger'])
+ self._check_output(
+ '',
+ 'Usage: swarming.py trigger [options] (hash|isolated)\n\n'
+ 'swarming.py: error: --swarming is required.'
+ '\n')
+
+ def test_trigger_no_swarming_env_var(self):
+ with self.assertRaises(SystemExit):
+ with test_utils.EnvVars({'ISOLATE_SERVER': 'https://host'}):
+ main(['trigger', '-T' 'foo', 'foo.isolated'])
+ self._check_output(
+ '',
+ 'Usage: swarming.py trigger [options] (hash|isolated)\n\n'
+ 'swarming.py: error: --swarming is required.'
+ '\n')
+
+ def test_trigger_no_isolate_env_var(self):
+ with self.assertRaises(SystemExit):
+ with test_utils.EnvVars({'SWARMING_SERVER': 'https://host'}):
+ main(['trigger', 'T', 'foo', 'foo.isolated'])
+ self._check_output(
+ '',
+ 'Usage: swarming.py trigger [options] (hash|isolated)\n\n'
+ 'swarming.py: error: Use one of --indir or --isolate-server.'
+ '\n')
+
+ def test_trigger_env_var(self):
+ with self.assertRaises(SystemExit):
+ with test_utils.EnvVars({'ISOLATE_SERVER': 'https://host',
+ 'SWARMING_SERVER': 'https://host'}):
+ main(['trigger', '-T', 'foo'])
+ self._check_output(
+ '',
+ 'Usage: swarming.py trigger [options] (hash|isolated)\n\n'
+ 'swarming.py: error: Must pass one .isolated file or its hash (sha1).'
+ '\n')
+
+ def test_trigger_no_task(self):
+ with self.assertRaises(SystemExit):
+ main([
+ 'trigger', '--swarming', 'https://host',
+ '--isolate-server', 'https://host', 'foo.isolated',
+ ])
+ self._check_output(
+ '',
+ 'Usage: swarming.py trigger [options] (hash|isolated)\n\n'
+ 'swarming.py: error: Please at least specify the dimension of the '
+ 'swarming bot OS with --dimension os <something>.\n')
+
def test_trigger_env(self):
self.mock(swarming.isolateserver, 'get_storage',
lambda *_: MockedStorage(warm_cache=False))
self.assertEqual(0, ret, (actual, sys.stderr.getvalue()))
+def clear_env_vars():
+ for e in ('ISOLATE_SERVER', 'SWARMING_SERVER'):
+ os.environ.pop(e, None)
+
+
if __name__ == '__main__':
logging.basicConfig(
level=logging.DEBUG if '-v' in sys.argv else logging.ERROR)
if '-v' in sys.argv:
unittest.TestCase.maxDiff = None
+ clear_env_vars()
unittest.main()
projects / platform / framework / web / crosswalk.git / blobdiff