Add an initial "pub lish" command.

utils/tests/pub/test_pub.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 /**
  * Test infrastructure for testing pub. Unlike typical unit tests, most pub
  * tests are integration tests that stage some stuff on the file system, run
  * pub, and then validate the results. This library provides an API to build
  * tests like that.
  */
 library test_pub;
 
 import 'dart:io';
 import 'dart:isolate';
 import 'dart:json';
 import 'dart:math';
 import 'dart:uri';
 
+import '../../../pkg/oauth2/lib/oauth2.dart' as oauth2;
 import '../../../pkg/unittest/lib/unittest.dart';
 import '../../lib/file_system.dart' as fs;
 import '../../pub/git_source.dart';
 import '../../pub/hosted_source.dart';
 import '../../pub/io.dart';
 import '../../pub/sdk_source.dart';
 import '../../pub/utils.dart';
 import '../../pub/yaml/yaml.dart';
 
 /**
@@ skipping 328 matching lines @@
       contents.add(packageCacheDir(name, version));
     }
   });
   return dir(cachePath, [
     dir('hosted', [
       async(port.transform((p) => dir('localhost%58$p', contents)))
     ])
   ]);
 }
 
+/// Describes the file in the system cache that contains the client's OAuth2
+/// credentials. The URL "/token" on [server] will be used as the token
+/// endpoint for refreshing the access token.
+Descriptor credentialsFile(
+    ScheduledServer server,
+    String accessToken,
+    {String refreshToken,
+     Date expiration}) {
+  return async(server.url.transform((url) {
+    return dir(cachePath, [
+      file('credentials.json', new oauth2.Credentials(
+          accessToken,
+          refreshToken,
+          url.resolve('/token'),
+          ['https://www.googleapis.com/auth/userinfo.email'],
+          expiration).toJson())
+    ]);
+  }));
+}
+
 /**
  * Describes the application directory, containing only a pubspec specifying the
  * given [dependencies].
  */
 DirectoryDescriptor appDir(List dependencies) =>
   dir(appPath, [appPubspec(dependencies)]);
 
 /**
  * Converts a list of dependencies as passed to [package] into a hash as used in
  * a pubspec.
@@ skipping 76 matching lines @@
  * The list of events that are scheduled to run as part of the test case.
  */
 List<_ScheduledEvent> _scheduled;
 
 /**
  * The list of events that are scheduled to run after the test case, even if it
  * failed.
  */
 List<_ScheduledEvent> _scheduledCleanup;
 
+/// The list of events that are scheduled to run after the test case only if it
+/// failed.
+List<_ScheduledEvent> _scheduledOnException;
+
 /**
  * Set to true when the current batch of scheduled events should be aborted.
  */
 bool _abortScheduled = false;
 
 /**
  * Runs all the scheduled events for a test case. This should only be called
  * once per test case.
  */
 void run() {
   var createdSandboxDir;
 
   var asyncDone = expectAsync0(() {});
 
   Future cleanup() {
     return _runScheduled(createdSandboxDir, _scheduledCleanup).chain((_) {
       _scheduled = null;
+      _scheduledCleanup = null;
+      _scheduledOnException = null;
       if (createdSandboxDir != null) return deleteDir(createdSandboxDir);
       return new Future.immediate(null);
     });
   }
 
   final future = _setUpSandbox().chain((sandboxDir) {
     createdSandboxDir = sandboxDir;
     return _runScheduled(sandboxDir, _scheduled);
   });
 
   future.handleException((error) {
     // If an error occurs during testing, delete the sandbox, throw the error so
     // that the test framework sees it, then finally call asyncDone so that the
     // test framework knows we're done doing asynchronous stuff.
-    cleanup().then((_) => registerException(error, future.stackTrace));
+    var future = _runScheduled(createdSandboxDir, _scheduledOnException)
+        .chain((_) => cleanup());
+    future.handleException((e) {
+      print("Exception while cleaning up: $e");
+      print(future.stackTrace);
+      registerException(error, future.stackTrace);
+      return true;
+    });
+    future.then((_) => registerException(error, future.stackTrace));
     return true;
   });
 
   future.chain((_) => cleanup()).then((_) {
     asyncDone();
   });
 }
 
 /// Get the path to the root "util/test/pub" directory containing the pub tests.
 String get testDirectory {
   var dir = new Path.fromNative(new Options().script);
   while (dir.filename != 'pub') dir = dir.directoryPath;
 
   return new File(dir.toNativePath()).fullPathSync();
 }
 
 /**
  * Schedules a call to the Pub command-line utility. Runs Pub with [args] and
  * validates that its results match [output], [error], and [exitCode].
  */
 void schedulePub({List<String> args, Pattern output, Pattern error,
-    int exitCode: 0}) {
+    Future<Uri> tokenEndpoint, int exitCode: 0}) {
   _schedule((sandboxDir) {
-    String pathInSandbox(path) => join(getFullPath(sandboxDir), path);
-
-    return ensureDir(pathInSandbox(appPath)).chain((_) {
-      // Find a Dart executable we can use to spawn. Use the same one that was
-      // used to run this script itself.
-      var dartBin = new Options().executable;
-
-      // If the executable looks like a path, get its full path. That way we
-      // can still find it when we spawn it with a different working directory.
-      if (dartBin.contains(Platform.pathSeparator)) {
-        dartBin = new File(dartBin).fullPathSync();
-      }
-
-      // Find the main pub entrypoint.
-      var pubPath = fs.joinPaths(testDirectory, '../../pub/pub.dart');
-
-      var dartArgs =
-          ['--enable-type-checks', '--enable-asserts', pubPath, '--trace'];
-      dartArgs.addAll(args);
-
-      var environment = {
-        'PUB_CACHE': pathInSandbox(cachePath),
-        'DART_SDK': pathInSandbox(sdkPath)
-      };
-
-      return runProcess(dartBin, dartArgs, workingDir: pathInSandbox(appPath),
-          environment: environment);
-    }).transform((result) {
+    return _doPub(runProcess, sandboxDir, args, tokenEndpoint)
+        .transform((result) {
       var failures = [];
 
       _validateOutput(failures, 'stdout', output, result.stdout);
       _validateOutput(failures, 'stderr', error, result.stderr);
 
       if (result.exitCode != exitCode) {
         failures.add(
             'Pub returned exit code ${result.exitCode}, expected $exitCode.');
       }
 
@@ skipping 15 matching lines @@
 /**
  * A shorthand for [schedulePub] and [run] when no validation needs to be done
  * after Pub has been run.
  */
 void runPub({List<String> args, Pattern output, Pattern error,
     int exitCode: 0}) {
   schedulePub(args: args, output: output, error: error, exitCode: exitCode);
   run();
 }
 
+/// Starts a Pub process and returns a [ScheduledProcess] that supports
+/// interaction with that process.
+ScheduledProcess startPub({List<String> args}) {
+  var process = _scheduleValue((sandboxDir) =>
+      _doPub(startProcess, sandboxDir, args));
+  return new ScheduledProcess("pub", process);
+}
+
+/// Like [startPub], but runs `pub lish` in particular with [server] used both
+/// as the OAuth2 server (with "/token" as the token endpoint) and as the
+/// package server.
+ScheduledProcess startPubLish(ScheduledServer server, {List<String> args}) {
+  var process = _scheduleValue((sandboxDir) {
+    return server.url.chain((url) {
+      var tokenEndpoint = url.resolve('/token');
+      if (args == null) args = [];
+      args = flatten(['lish', '--server', url.toString(), args]);
+      return _doPub(startProcess, sandboxDir, args, tokenEndpoint);
+    });
+  });
+  return new ScheduledProcess("pub lish", process);
+}
+
+/// Calls [fn] with appropriately modified arguments to run a pub process. [fn]
+/// should have the same signature as [startProcess], except that the returned
+/// [Future] may have a type other than [Process].
+Future _doPub(Function fn, sandboxDir, List<String> args, Uri tokenEndpoint) {
+  String pathInSandbox(path) => join(getFullPath(sandboxDir), path);
+
+  return ensureDir(pathInSandbox(appPath)).chain((_) {
+    // Find a Dart executable we can use to spawn. Use the same one that was
+    // used to run this script itself.
+    var dartBin = new Options().executable;
+
+    // If the executable looks like a path, get its full path. That way we
+    // can still find it when we spawn it with a different working directory.
+    if (dartBin.contains(Platform.pathSeparator)) {
+      dartBin = new File(dartBin).fullPathSync();
+    }
+
+    // Find the main pub entrypoint.
+    var pubPath = fs.joinPaths(testDirectory, '../../pub/pub.dart');
+
+    var dartArgs =
+        ['--enable-type-checks', '--enable-asserts', pubPath, '--trace'];
+    dartArgs.addAll(args);
+
+    var environment = {
+      'PUB_CACHE': pathInSandbox(cachePath),
+      'DART_SDK': pathInSandbox(sdkPath)
+    };
+    if (tokenEndpoint != null) {
+      environment['_PUB_TEST_TOKEN_ENDPOINT'] = tokenEndpoint.toString();
+    }
+
+    return fn(dartBin, dartArgs, workingDir: pathInSandbox(appPath),
+        environment: environment);
+  });
+}
+
 /**
  * Skips the current test if Git is not installed. This validates that the
  * current test is running on a buildbot in which case we expect git to be
  * installed. If we are not running on the buildbot, we will instead see if git
  * is installed and skip the test if not. This way, users don't need to have git
  * installed to run the tests locally (unless they actually care about the pub
  * git tests).
  */
 void ensureGit() {
   _schedule((_) {
@@ skipping 421 matching lines @@
   /**
    * Schedules changes to be committed to the Git repository.
    */
   void scheduleCommit() => _schedule((dir) => this.commit(dir));
 
   /**
    * Return a Future that completes to the commit in the git repository referred
    * to by [ref] at the current point in the scheduled test run.
    */
   Future<String> revParse(String ref) {
-    var completer = new Completer<String>();
-    _schedule((parentDir) {
+    return _scheduleValue((parentDir) {
       return super.create(parentDir).chain((rootDir) {
         return _runGit(['rev-parse', ref], rootDir);
-      }).transform((output) {
-        completer.complete(output[0]);
-        return null;
-      });
+      }).transform((output) => output[0]);
     });
-    return completer.future;
   }
 
   /// Schedule a Git command to run in this repository.
   void scheduleGit(List<String> args) {
     _schedule((parentDir) {
       var gitDir = new Directory(join(parentDir, name));
       return _runGit(args, gitDir);
     });
   }
 
@@ skipping 140 matching lines @@
   InputStream load(List<String> path) {
     if (path.isEmpty) {
       throw "Can't load the contents of $name: it doesn't exist.";
     } else {
       throw "Can't load ${Strings.join(path, '/')} from within $name: $name "
         "doesn't exist.";
     }
   }
 }
 
+/// A class representing a [Process] that is scheduled to run in the course of
+/// the test. This class allows actions on the process to be scheduled
+/// synchronously. All operations on this class are scheduled.
+///
+/// Before running the test, either [shouldExit] or [kill] must be called on
+/// this to ensure that the process terminates when expected.
+///
+/// If the test fails, this will automatically print out any remaining stdout
+/// and stderr from the process to aid debugging.
+class ScheduledProcess {
+  /// The name of the process. Used for error reporting.
+  final String name;
+
+  /// The process that's scheduled to run.
+  final Future<Process> _process;
+
+  /// A [StringInputStream] wrapping the stdout of the process that's scheduled
+  /// to run.
+  final Future<StringInputStream> _stdout;
+
+  /// A [StringInputStream] wrapping the stderr of the process that's scheduled
+  /// to run.
+  final Future<StringInputStream> _stderr;
+
+  /// The exit code of the process that's scheduled to run. This will naturally
+  /// only complete once the process has terminated.
+  Future<int> get _exitCode => _exitCodeCompleter.future;
+
+  /// The completer for [_exitCode].
+  final Completer<int> _exitCodeCompleter = new Completer();
+
+  /// Whether the user has scheduled the end of this process by calling either
+  /// [shouldExit] or [kill].
+  bool _endScheduled = false;
+
+  /// Whether the process is expected to terminate at this point.
+  bool _endExpected = false;
+
+  /// Wraps a [Process] [Future] in a scheduled process.
+  ScheduledProcess(this.name, Future<Process> process)
+    : _process = process,
+      _stdout = process.transform((p) => new StringInputStream(p.stdout)),
+      _stderr = process.transform((p) => new StringInputStream(p.stderr)) {
+
+    _schedule((_) {
+      if (!_endScheduled) {
+        throw new StateError("Scheduled process $name must have shouldExit() "
+            "or kill() called before the test is run.");
+      }
+
+      return _process.transform((p) {
+        p.onExit = (c) {
+          if (_endExpected) {
+            _exitCodeCompleter.complete(c);
+            return;
+          }
+
+          // Sleep for half a second in case _endExpected is set in the next

[Bob Nystrom, 2012/11/27 21:00:53]

Why 500ms?

[nweiz, 2012/11/27 22:07:34]

It seemed like a time increment that would be large enough to allow the
previously scheduled event to finish up (assuming it's done with most of its IO)
while being short enough to avoid overlapping too much with future callbacks and
having a false positive due to _endExpected being set later.

+          // scheduled event.
+          sleep(500).then((_) {
+            if (_endExpected) {
+              _exitCodeCompleter.complete(c);
+              return;
+            }
+
+            _printStreams().then((_) {
+              registerException(new ExpectException("Process $name ended "
+                  "earlier than scheduled with exit code $c"));
+            });
+          });
+        };
+      });
+    });
+
+    _scheduleOnException((_) {
+      if (!_process.hasValue) return;
+
+      if (!_exitCode.hasValue) {
+        print("\nKilling process $name prematurely.");
+        _endExpected = true;
+        _process.value.kill();
+      }
+
+      return _printStreams();
+    });
+
+    _scheduleCleanup((_) {
+      if (!_process.hasValue) return;
+      // Ensure that the process is dead and we aren't waiting on any IO.
+      var process = _process.value;
+      process.kill();
+      process.stdout.close();
+      process.stderr.close();
+    });
+  }
+
+  /// Reads the next line of stdout from the process.
+  Future<String> nextLine() {
+    return _scheduleValue((_) {
+      return timeout(_stdout.chain(readLine), 5000,
+          "waiting for the next stdout line from process $name");
+    });
+  }
+
+  /// Reads the next line of stderr from the process.
+  Future<String> nextErrLine() {
+    return _scheduleValue((_) {
+      return timeout(_stderr.chain(readLine), 5000,
+          "waiting for the next stderr line from process $name");
+    });
+  }
+
+  /// Writes [line] to the process as stdin.
+  void writeLine(String line) {
+    _schedule((_) => _process.transform((p) => p.stdin.writeString('$line\n')));
+  }
+
+  /// Kills the process, and waits until it's dead.
+  void kill() {
+    _endScheduled = true;
+    _schedule((_) {
+      _endExpected = true;
+      return _process.chain((p) {
+        p.kill();
+        return timeout(_exitCode, 5000, "waiting for process $name to die");
+      });
+    });
+  }
+
+  /// Waits for the process to exit, and verifies that the exit code matches
+  /// [expectedExitCode] (if given).
+  void shouldExit([int expectedExitCode]) {
+    _endScheduled = true;
+    _schedule((_) {
+      _endExpected = true;
+      return timeout(_exitCode, 5000, "waiting for process $name to exit")
+          .transform((exitCode) {
+        if (expectedExitCode != null) expect(exitCode, equals(expectedExitCode));

[Bob Nystrom, 2012/11/27 21:00:53]

Long line.

[nweiz, 2012/11/27 22:07:34]

Done.

+      });
+    });
+  }
+
+  /// Prints the remaining data in the process's stdout and stderr streams.
+  /// Prints nothing if the straems are empty.
+  Future _printStreams() {
+    Future printStream(String streamName, StringInputStream stream) {
+      return consumeStringInputStream(stream).transform((output) {
+        if (output.isEmpty) return;
+
+        print('\nProcess $name $streamName:');
+        for (var line in output.trim().split("\n")) {
+          print('| $line');
+        }
+        return;
+      });
+    }
+
+    return printStream('stdout', _stdout.value)
+        .chain((_) => printStream('stderr', _stderr.value));
+  }
+}
+
+/// A class representing an [HttpServer] that's scheduled to run in the course
+/// of the test. This class allows the server's request handling to be scheduled
+/// synchronously. All operations on this class are scheduled.
+class ScheduledServer {
+  /// The wrapped server.
+  final Future<HttpServer> _server;
+
+  ScheduledServer._(this._server);
+
+  /// Creates a new server listening on an automatically-allocated port on
+  /// localhost.
+  factory ScheduledServer() {
+    return new ScheduledServer._(_scheduleValue((_) {
+      var server = new HttpServer();
+      server.defaultRequestHandler = _unexpectedRequest;
+      server.listen("127.0.0.1", 0);
+      _scheduleCleanup((_) => server.close());
+      return new Future.immediate(server);
+    }));
+  }
+
+  /// The port on which the server is listening.
+  Future<int> get port => _server.transform((s) => s.port);
+
+  /// The base URL of the server, including its port.
+  Future<Uri> get url =>
+    port.transform((p) => new Uri.fromString("http://localhost:$p"));
+
+  /// Assert that the next request has the given [method] and [path], and pass
+  /// it to [handler] to handle. If [handler] returns a [Future], wait until
+  /// it's completed to continue the schedule.
+  void handle(String method, String path,
+      Future handler(HttpRequest request, HttpResponse response)) {
+    _schedule((_) {

[Bob Nystrom, 2012/11/27 21:00:53]

Could do:

_schedule((_) => timeout(...));

And get rid of a level of indentation.

[nweiz, 2012/11/27 22:07:34]

That's allowed? :o

+      return timeout(_server.chain((server) {
+        var completer = new Completer();
+        server.defaultRequestHandler = (request, response) {
+          server.defaultRequestHandler = _unexpectedRequest;
+          expect(request.method, equals(method));
+          expect(request.path, equals(path));
+
+          var future = handler(request, response);
+          if (future == null) future = new Future.immediate(null);
+          chainToCompleter(future, completer);
+        };
+        return completer.future;
+      }), 5000, "waiting for $method $path");
+    });
+  }
+
+  /// Raises an error complaining of an unexpected request.
+  static void _unexpectedRequest(HttpRequest request, HttpResponse response) {
+    fail('Unexpected ${request.method} request to ${request.path}.');
+  }
+}
+
 /**
  * Takes a simple data structure (composed of [Map]s, [List]s, scalar objects,
  * and [Future]s) and recursively resolves all the [Future]s contained within.
  * Completes with the fully resolved structure.
  */
 Future _awaitObject(object) {
   // Unroll nested futures.
   if (object is Future) return object.chain(_awaitObject);
   if (object is Collection) return Futures.wait(object.map(_awaitObject));
   if (object is! Map) return new Future.immediate(object);
@@ skipping 13 matching lines @@
 }
 
 /**
  * Schedules a callback to be called as part of the test case.
  */
 void _schedule(_ScheduledEvent event) {
   if (_scheduled == null) _scheduled = [];
   _scheduled.add(event);
 }
 
-/**
- * Schedules a callback to be called after Pub is run with [runPub], even if it
- * fails.
- */
+/// Like [_schedule], but pipes the return value of [event] to a returned
+/// [Future].
+Future _scheduleValue(_ScheduledEvent event) {
+  var completer = new Completer();
+  _schedule((parentDir) {

[Bob Nystrom, 2012/11/27 21:00:53]

"parentDir" -> "value".

[nweiz, 2012/11/27 22:07:34]

This isn't the value that's being returned. It actually is the parent directory
of the test.

+    chainToCompleter(event(parentDir), completer);
+    return completer.future;
+  });
+  return completer.future;
+}
+
+/// Schedules a callback to be called after the test case has completed, even if
+/// it failed.
 void _scheduleCleanup(_ScheduledEvent event) {
   if (_scheduledCleanup == null) _scheduledCleanup = [];
   _scheduledCleanup.add(event);
 }
+
+/// Schedules a callback to be called after the test case has completed, but
+/// only if it failed.
+void _scheduleOnException(_ScheduledEvent event) {
+  if (_scheduledOnException == null) _scheduledOnException = [];
+  _scheduledOnException.add(event);
+}
+
+/// Like [expect], but for [Future]s that complete as part of the scheduled
+/// test. This is necessary to ensure that the exception thrown by the
+/// expectation failing is handled by the scheduler.
+///
+/// Note that [matcher] matches against the completed value of [actual], so
+/// calling [completion] is unnecessary.
+Matcher expectLater(Future actual, matcher, {String reason,
+    FailureHandler failureHandler, bool verbose: false}) {
+  _schedule((_) {
+    return actual.transform((value) {
+      expect(value, matcher, reason: reason, failureHandler: failureHandler,
+          verbose: false);
+    });
+  });
+}