Add support for ${ISOLATED_OUTDIR} for swarming.py reproduce.

Fixes #250.

[email protected]
BUG=

Review URL: https://codereview.chromium.org/1722653004

client/swarming.py

@@ -36,6 +36,7 @@
 import auth
 import isolated_format
 import isolateserver
+import run_isolated
 
 
 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -1333,6 +1334,9 @@ def CMDreproduce(parser, args):
   You can pass further additional arguments to the target command by passing
   them after --.
   """
+  parser.add_option(
+      '--output-dir', metavar='DIR', default='',
+      help='Directory that will have results stored into')
   options, args = parser.parse_args(args)
   extra_args = []
   if not args:
@@ -1381,6 +1385,11 @@ def CMDreproduce(parser, args):
       if bundle.relative_cwd:
         workdir = os.path.join(workdir, bundle.relative_cwd)
       command.extend(properties.get('extra_args') or [])
+    # https://github.com/luci/luci-py/blob/master/appengine/swarming/doc/Magic-Values.md
+    new_command = run_isolated.process_command(command, options.output_dir)
+    if not options.output_dir and new_command != command:
+      parser.error('The task has outputs, you must use --output-dir')
+    command = new_command
   else:
     command = properties['command']
   try:

client/tests/isolateserver_mock.py

@@ -36,9 +36,7 @@ def validate(cls, ticket, message):
     return json.loads(a.groups()[0])
 
 
-class IsolateServerHandler(BaseHTTPServer.BaseHTTPRequestHandler):
-  """An extremely minimal implementation of the isolate server API v1.0."""
-
+class MockHandler(BaseHTTPServer.BaseHTTPRequestHandler):
   def _json(self, data):
     """Sends a JSON response."""
     self.send_response(200)
@@ -65,26 +63,14 @@ def _drop_body(self):
       self.rfile.read(chunk)
       size -= chunk
 
-  def do_GET(self):
-    logging.info('GET %s', self.path)
-    if self.path in ('/on/load', '/on/quit'):
-      self._octet_stream('')
-    elif self.path == '/auth/api/v1/server/oauth_config':
-      self._json({
-          'client_id': 'c',
-          'client_not_so_secret': 's',
-          'primary_url': self.server.url})
-    elif self.path == '/auth/api/v1/accounts/self':
-      self._json({'identity': 'user:joe', 'xsrf_token': 'foo'})
-    elif self.path.startswith('/_ah/api/isolateservice/v1/retrieve'):
-      namespace, h = self.path[len(
-          '/_ah/api/isolateservice/v1/retrieve'):].split('/', 1)
-      self._octet_stream(self.server.contents[namespace][h])
-    else:
-      raise NotImplementedError(self.path)
+  def log_message(self, fmt, *args):
+    logging.info(
+        '%s - - [%s] %s', self.address_string(), self.log_date_time_string(),
+        fmt % args)
 
-  ### Utility Functions Adapted from endpoint_handlers_api
-  # TODO(cmassaro): inherit these directly?
+
+class IsolateServerHandler(MockHandler):
+  """An extremely minimal implementation of the isolate server API v1.0."""
 
   def _should_push_to_gs(self, isolated, size):
     max_memcache = 500 * 1024
@@ -120,7 +106,22 @@ def _storage_helper(self, body, gs=False):
 
   ### Mocked HTTP Methods
 
+  def do_GET(self):
+    logging.info('GET %s', self.path)
+    if self.path in ('/on/load', '/on/quit'):
+      self._octet_stream('')
+    elif self.path == '/auth/api/v1/server/oauth_config':
+      self._json({
+          'client_id': 'c',
+          'client_not_so_secret': 's',
+          'primary_url': self.server.url})
+    elif self.path == '/auth/api/v1/accounts/self':
+      self._json({'identity': 'user:joe', 'xsrf_token': 'foo'})
+    else:
+      raise NotImplementedError(self.path)
+
   def do_POST(self):
+    logging.info('POST %s', self.path)
     body = self._read_body()
     if self.path.startswith('/_ah/api/isolateservice/v1/preupload'):
       response = {'items': []}
@@ -157,7 +158,10 @@ def append_entry(entry, index, li):
     elif self.path.startswith('/_ah/api/isolateservice/v1/retrieve'):
       request = json.loads(body)
       namespace = request['namespace']['namespace']
-      data = self.server.contents[namespace][request['digest']]
+      data = self.server.contents[namespace].get(request['digest'])
+      if data is None:
+        logging.error(
+            'Failed to retrieve %s / %s', namespace, request['digest'])
       self._json({'content': data})
     elif self.path.startswith('/_ah/api/isolateservice/v1/server_details'):
       self._json({'server_version': 'such a good version'})
@@ -177,26 +181,47 @@ def do_PUT(self):
     else:
       raise NotImplementedError(self.path)
 
-  def log_message(self, fmt, *args):
-    logging.info(
-        '%s - - [%s] %s', self.address_string(), self.log_date_time_string(),
-        fmt % args)
 
+class MockServer(object):
+  _HANDLER_CLS = None
 
-class MockIsolateServer(object):
   def __init__(self):
     self._closed = False
     self._server = BaseHTTPServer.HTTPServer(
-        ('127.0.0.1', 0), IsolateServerHandler)
-    self._server.contents = {}
-    self._server.discard_content = False
+        ('127.0.0.1', 0), self._HANDLER_CLS)
     self._server.url = self.url = 'http://localhost:%d' % (
         self._server.server_port)
     self._thread = threading.Thread(target=self._run, name='httpd')
     self._thread.daemon = True
     self._thread.start()
     logging.info('%s', self.url)
 
+  def close(self):
+    self.close_start()
+    self.close_end()
+
+  def close_start(self):
+    assert not self._closed
+    self._closed = True
+    urllib2.urlopen(self.url + '/on/quit')
+
+  def close_end(self):
+    assert self._closed
+    self._thread.join()
+
+  def _run(self):
+    while not self._closed:
+      self._server.handle_request()
+
+
+class MockIsolateServer(MockServer):
+  _HANDLER_CLS = IsolateServerHandler
+
+  def __init__(self):
+    super(MockIsolateServer, self).__init__()
+    self._server.contents = {}
+    self._server.discard_content = False
+
   def discard_content(self):
     """Stops saving content in memory. Used to test large files."""
     self._server.discard_content = True
@@ -220,20 +245,3 @@ def add_content(self, namespace, content):
     self._server.contents.setdefault(namespace, {})[h] = base64.b64encode(
         content)
     return h
-
-  def close(self):
-    self.close_start()
-    self.close_end()
-
-  def close_start(self):
-    assert not self._closed
-    self._closed = True
-    urllib2.urlopen(self.url + '/on/quit')
-
-  def close_end(self):
-    assert self._closed
-    self._thread.join()
-
-  def _run(self):
-    while not self._closed:
-      self._server.handle_request()

client/tests/swarming_test.py

@@ -8,6 +8,7 @@
 import json
 import logging
 import os
+import re
 import StringIO
 import subprocess
 import sys
@@ -153,28 +154,56 @@ def wait(self, timeout=None):
     return super(NonBlockingEvent, self).wait(0)
 
 
-class NetTestCase(net_utils.TestCase):
-  """Base class that defines the url_open mock."""
+class SwarmingServerHandler(isolateserver_mock.MockHandler):
+  """An extremely minimal implementation of the swarming server API v1.0."""
+
+  def do_GET(self):
+    logging.info('S GET %s', self.path)
+    if self.path in ('/on/load', '/on/quit'):
+      self._octet_stream('')
+    elif self.path == '/auth/api/v1/server/oauth_config':
+      self._json({
+          'client_id': 'c',
+          'client_not_so_secret': 's',
+          'primary_url': self.server.url})
+    elif self.path == '/auth/api/v1/accounts/self':
+      self._json({'identity': 'user:joe', 'xsrf_token': 'foo'})
+    else:
+      m = re.match(r'/_ah/api/swarming/v1/task/(\d+)/request', self.path)
+      if m:
+        logging.info('%s', m.group(1))
+        self._json(self.server.tasks[int(m.group(1))])
+      else:
+        self._json( {'a': 'b'})
+        #raise NotImplementedError(self.path)
+
+  def do_POST(self):
+    logging.info('POST %s', self.path)
+    raise NotImplementedError(self.path)
+
+
+class MockSwarmingServer(isolateserver_mock.MockServer):
+  _HANDLER_CLS = SwarmingServerHandler
+
+  def __init__(self):
+    super(MockSwarmingServer, self).__init__()
+    self._server.tasks = {}
+
+
+class Common(object):
   def setUp(self):
-    super(NetTestCase, self).setUp()
     self._tempdir = None
     self.mock(auth, 'ensure_logged_in', lambda _: None)
-    self.mock(time, 'sleep', lambda _: None)
-    self.mock(subprocess, 'call', lambda *_: self.fail())
-    self.mock(threading, 'Event', NonBlockingEvent)
     self.mock(sys, 'stdout', StringIO.StringIO())
     self.mock(sys, 'stderr', StringIO.StringIO())
     self.mock(logging_utils, 'prepare_logging', lambda *args: None)
     self.mock(logging_utils, 'set_console_level', lambda *args: None)
 
   def tearDown(self):
-    try:
-      if self._tempdir:
-        file_path.rmtree(self._tempdir)
-      if not self.has_failed():
-        self._check_output('', '')
-    finally:
-      super(NetTestCase, self).tearDown()
+    if self._tempdir:
+      file_path.rmtree(self._tempdir)
+    if not self.has_failed():
+      self._check_output('', '')
 
   @property
   def tempdir(self):
@@ -194,27 +223,80 @@ def _check_output(self, out, err):
     self.mock(sys, 'stderr', StringIO.StringIO())
 
 
-class TestIsolated(auto_stub.TestCase):
+class NetTestCase(net_utils.TestCase, Common):
+  """Base class that defines the url_open mock."""
+  def setUp(self):
+    net_utils.TestCase.setUp(self)
+    Common.setUp(self)
+    self.mock(time, 'sleep', lambda _: None)
+    self.mock(subprocess, 'call', lambda *_: self.fail())
+    self.mock(threading, 'Event', NonBlockingEvent)
+
+
+class TestIsolated(auto_stub.TestCase, Common):
   """Test functions with isolated_ prefix."""
   def setUp(self):
-    super(TestIsolated, self).setUp()
-    self._server = None
+    auto_stub.TestCase.setUp(self)
+    Common.setUp(self)
+    self._isolate = isolateserver_mock.MockIsolateServer()
+    self._swarming = MockSwarmingServer()
 
   def tearDown(self):
     try:
-      if self._server:
-        self._server.close_start()
-      if self._server:
-        self._server.close_end()
+      self._isolate.close_start()
+      self._swarming.close_start()
+      self._isolate.close_end()
+      self._swarming.close_end()
     finally:
-      super(TestIsolated, self).tearDown()
+      Common.tearDown(self)
+      auto_stub.TestCase.tearDown(self)
 
-  @property
-  def server(self):
-    """Creates the Isolate Server mock on first reference."""
-    if not self._server:
-      self._server = isolateserver_mock.MockIsolateServer()
-    return self._server
+  def test_reproduce_isolated(self):
+    old_cwd = os.getcwd()
+    try:
+      os.chdir(self.tempdir)
+
+      def call(cmd, env, cwd):
+        self.assertEqual([sys.executable, u'main.py', u'foo', '--bar'], cmd)
+        self.assertEqual(None, env)
+        self.assertEqual(unicode(os.path.abspath('work')), cwd)
+        return 0
+
+      self.mock(subprocess, 'call', call)
+
+      main_hash = self._isolate.add_content_compressed(
+          'default-gzip', 'not executed')
+      isolated = {
+        'files': {
+          'main.py': {
+            'h': main_hash,
+            's': 12,
+            'm': 0700,
+          },
+        },
+        'command': ['python', 'main.py'],
+      }
+      isolated_hash = self._isolate.add_content_compressed(
+          'default-gzip', json.dumps(isolated))
+      self._swarming._server.tasks[123] = {
+        'properties': {
+          'inputs_ref': {
+            'isolatedserver': self._isolate.url,
+            'namespace': 'default-gzip',
+            'isolated': isolated_hash,
+          },
+          'extra_args': ['foo'],
+        },
+      }
+      ret = main(
+          [
+            'reproduce', '--swarming', self._swarming.url, '123', '--',
+            '--bar',
+          ])
+      self._check_output('', '')
+      self.assertEqual(0, ret)
+    finally:
+      os.chdir(old_cwd)
 
 
 class TestSwarmingTrigger(NetTestCase):