Merge branch 'master' of http://github.com/sskirby/hydra into sskirby

.gitignore

@@ -17,5 +17,6 @@ tmtags
 coverage
 rdoc
 pkg
+tags
 
 ## PROJECT::SPECIFIC

hydra.gemspec

@@ -51,6 +51,7 @@ Gem::Specification.new do |s|
      "lib/hydra/spec/hydra_formatter.rb",
      "lib/hydra/ssh.rb",
      "lib/hydra/stdio.rb",
+     "lib/hydra/sync.rb",
      "lib/hydra/tasks.rb",
      "lib/hydra/trace.rb",
      "lib/hydra/worker.rb",
@@ -73,6 +74,7 @@ Gem::Specification.new do |s|
      "test/pipe_test.rb",
      "test/runner_test.rb",
      "test/ssh_test.rb",
+     "test/sync_test.rb",
      "test/test_helper.rb",
      "test/worker_test.rb"
   ]
@@ -97,6 +99,7 @@ Gem::Specification.new do |s|
      "test/test_helper.rb",
      "test/master_test.rb",
      "test/runner_test.rb",
+     "test/sync_test.rb",
      "test/worker_test.rb"
   ]
 

lib/hydra.rb

@@ -7,6 +7,7 @@ require 'hydra/safe_fork'
 require 'hydra/runner'
 require 'hydra/worker'
 require 'hydra/master'
+require 'hydra/sync'
 require 'hydra/listener/abstract'
 require 'hydra/listener/minimal_output'
 require 'hydra/listener/report_generator'

lib/hydra/master.rb

@@ -14,6 +14,7 @@ module Hydra #:nodoc:
     include Hydra::Messages::Master
     include Open3
     traceable('MASTER')
+
     # Create a new Master
     #
     # Options:
@@ -32,6 +33,12 @@ module Hydra #:nodoc:
     # * :autosort
     #   * Set to false to disable automatic sorting by historical run-time per file
     def initialize(opts = { })
+      trap("SIGINT") do
+        puts "Testing halted by user.  Untested files:"
+        puts @incomplete_files.join("\n")
+        exit
+      end 
+
       opts.stringify_keys!
       config_file = opts.delete('config') { nil }
       if config_file
@@ -64,6 +71,7 @@ module Hydra #:nodoc:
       @verbose = opts.fetch('verbose') { false }
       @autosort = opts.fetch('autosort') { true }
       @sync = opts.fetch('sync') { nil }
+      @environment = opts.fetch('environment') { 'test' }
 
       if @autosort
         sort_files_from_report 
@@ -151,39 +159,16 @@ module Hydra #:nodoc:
     end
 
     def boot_ssh_worker(worker)
+      sync = Sync.new(worker, @sync, @verbose)
+
       runners = worker.fetch('runners') { raise "You must specify the number of runners"  }
-      connect = worker.fetch('connect') { raise "You must specify an SSH connection target" }
-      ssh_opts = worker.fetch('ssh_opts') { "" }
-      directory = worker.fetch('directory') { raise "You must specify a remote directory" }
       command = worker.fetch('command') { 
-        "ruby -e \"require 'rubygems'; require 'hydra'; Hydra::Worker.new(:io => Hydra::Stdio.new, :runners => #{runners}, :verbose => #{@verbose});\""
+        "RAILS_ENV=#{@environment} ruby -e \"require 'rubygems'; require 'hydra'; Hydra::Worker.new(:io => Hydra::Stdio.new, :runners => #{runners}, :verbose => #{@verbose});\""
       }
 
-      if @sync
-        @sync.stringify_keys!
-        trace "Synchronizing with #{connect}\n\t#{@sync.inspect}"
-        local_dir = @sync.fetch('directory') { 
-          raise "You must specify a synchronization directory"
-        }
-        exclude_paths = @sync.fetch('exclude') { [] }
-        exclude_opts = exclude_paths.inject(''){|memo, path| memo += "--exclude=#{path} "}
-
-        rsync_command = [
-          'rsync',
-          '-avz',
-          '--delete',
-          exclude_opts,
-          File.expand_path(local_dir)+'/',
-          "-e \"ssh #{ssh_opts}\"",
-          "#{connect}:#{directory}"
-        ].join(" ")
-        trace rsync_command
-        trace `#{rsync_command}`
-      end
-
       trace "Booting SSH worker" 
-      ssh = Hydra::SSH.new("#{ssh_opts} #{connect}", directory, command)
+      ssh = Hydra::SSH.new("#{sync.ssh_opts} #{sync.connect}", sync.remote_dir, command)
-      return { :io => ssh, :idle => false, :type => :ssh }
+      return { :io => ssh, :idle => false, :type => :ssh, :connect => sync.connect }
     end
 
     def shutdown_all_workers

lib/hydra/sync.rb

@@ -0,0 +1,99 @@
+require 'yaml'
+module Hydra #:nodoc:
+  # Hydra class responsible for delegate work down to workers.
+  #
+  # The Sync is run once for each remote worker.
+  class Sync
+    traceable('SYNC')
+    self.class.traceable('SYNC MANY')
+
+    attr_reader :connect, :ssh_opts, :remote_dir
+
+    # Create a new Sync instance to rsync source from the local machine to a remote worker
+    #
+    # Arguments:
+    # * :worker_opts
+    #   * A hash of the configuration options for a worker.
+    # * :sync
+    #   * A hash of settings specifically for copying the source directory to be tested
+    #     to the remote worked
+    # * :verbose
+    #   * Set to true to see lots of Hydra output (for debugging)
+    def initialize(worker_opts, sync_opts, verbose = false)
+      worker_opts ||= {}
+      worker_opts.stringify_keys!
+      @verbose = verbose
+      @connect = worker_opts.fetch('connect') { raise "You must specify an SSH connection target" }
+      @ssh_opts = worker_opts.fetch('ssh_opts') { "" }
+      @remote_dir = worker_opts.fetch('directory') { raise "You must specify a remote directory" }
+
+      return unless sync_opts
+      sync_opts.stringify_keys!
+      @local_dir = sync_opts.fetch('directory') { raise "You must specify a synchronization directory" }
+      @exclude_paths = sync_opts.fetch('exclude') { [] }
+
+      trace "Initialized"
+      trace "  Worker: (#{worker_opts.inspect})"
+      trace "  Sync:   (#{sync_opts.inspect})"
+
+      sync
+    end
+
+    def sync
+      #trace "Synchronizing with #{connect}\n\t#{sync_opts.inspect}"
+      exclude_opts = @exclude_paths.inject(''){|memo, path| memo += "--exclude=#{path} "}
+
+      rsync_command = [
+        'rsync',
+        '-avz',
+        '--delete',
+        exclude_opts,
+        File.expand_path(@local_dir)+'/',
+        "-e \"ssh #{@ssh_opts}\"",
+        "#{@connect}:#{@remote_dir}"
+      ].join(" ")
+      trace rsync_command
+      trace `#{rsync_command}`
+    end
+
+    def self.sync_many opts
+      opts.stringify_keys!
+      config_file = opts.delete('config') { nil }
+      if config_file
+        opts.merge!(YAML.load_file(config_file).stringify_keys!)
+      end
+      @verbose = opts.fetch('verbose') { false }
+      @sync = opts.fetch('sync') { {} }
+
+      workers_opts = opts.fetch('workers') { [] }
+      @remote_worker_opts = []
+      workers_opts.each do |worker_opts|
+        worker_opts.stringify_keys!
+        if worker_opts['type'].to_s == 'ssh'
+          @remote_worker_opts << worker_opts
+        end
+      end
+
+      trace "Initialized"
+      trace "  Sync:   (#{@sync.inspect})"
+      trace "  Workers: (#{@remote_worker_opts.inspect})"
+
+      Thread.abort_on_exception = true
+      trace "Processing workers"
+      @listeners = []
+      @remote_worker_opts.each do |worker_opts|
+        @listeners << Thread.new do
+          begin
+            trace "Syncing #{worker_opts.inspect}"
+            Sync.new worker_opts, @sync, @verbose
+          rescue 
+            trace "Syncing failed [#{worker_opts.inspect}]"
+          end
+        end
+      end
+      
+      @listeners.each{|l| l.join}
+    end
+
+  end
+end

lib/hydra/tasks.rb

@@ -177,6 +177,43 @@ module Hydra #:nodoc:
     end
   end
 
+  # Define a sync task that uses hydra to rsync the source tree under test to remote workers.
+  #
+  # This task is very useful to run before a remote db:reset task to make sure the db/schema.rb
+  # file is up to date on the remote workers.
+  #
+  #   Hydra::SyncTask.new('hydra:sync') do |t|
+  #     t.verbose = false # optionally set to true for lots of debug messages
+  #   end  
+  class SyncTask < Hydra::Task
+
+    # Create a new SyncTestTask
+    def initialize(name = :sync)
+      @name = name
+      @verbose = false
+
+      yield self if block_given?
+
+      @config = find_config_file
+
+      @opts = {
+        :verbose => @verbose
+      }
+      @opts.merge!(:config => @config) if @config
+
+      define
+    end
+
+    private
+    # Create the rake task defined by this HydraSyncTask
+    def define
+      desc "Hydra Tests" + (@name == :hydra ? "" : " for #{@name}")
+      task @name do
+        Hydra::Sync.sync_many(@opts)
+      end
+    end
+  end
+
   # Setup a task that will be run across all remote workers
   #   Hydra::RemoteTask.new('db:reset')
   #
@@ -202,32 +239,59 @@ module Hydra #:nodoc:
       desc "Run #{@name} remotely on all workers"
       task "hydra:remote:#{@name}" do
         config = YAML.load_file(@config)
+        environment = config.fetch('environment') { 'test' }
         workers = config.fetch('workers') { [] }
         workers = workers.select{|w| w['type'] == 'ssh'}
+
+        $stdout.write "==== Hydra Running #{@name} ====\n"
+        Thread.abort_on_exception = true
+        @listeners = []
+        @results = {}
         workers.each do |worker|
+          @listeners << Thread.new do
+            begin
+              @results[worker] = if run_task(worker, environment)
+                "==== #{@name} passed on #{worker['connect']} ====\n"
+              else
+                "==== #{@name} failed on #{worker['connect']} ====\nPlease see above for more details.\n"
+              end
+            rescue 
+              @results[worker] = "==== #{@name} failed for #{worker['connect']} ====\n#{$!.inspect}\n#{$!.backtrace.join("\n")}"
+            end
+          end
+        end
+        @listeners.each{|l| l.join}
+        $stdout.write "\n==== Hydra Running #{@name} COMPLETE ====\n\n"
+        $stdout.write @results.values.join("\n")
+      end
+    end
+
+    def run_task worker, environment
       $stdout.write "==== Hydra Running #{@name} on #{worker['connect']} ====\n"
       ssh_opts = worker.fetch('ssh_opts') { '' }
       writer, reader, error = popen3("ssh -tt #{ssh_opts} #{worker['connect']} ")
       writer.write("cd #{worker['directory']}\n")
       writer.write "echo BEGIN HYDRA\n"
-          writer.write("RAILS_ENV=test rake #{@name}\n")
+      writer.write("RAILS_ENV=#{environment} rake #{@name}\n")
       writer.write "echo END HYDRA\n"
       writer.write("exit\n")
       writer.close
       ignoring = true
+      passed = true
       while line = reader.gets
         line.chomp!
+        if line =~ /^rake aborted!$/
+          passed = false
+        end
         if line =~ /echo END HYDRA$/
           ignoring = true
         end
-            $stdout.write "#{line}\n" unless ignoring
+        $stdout.write "#{worker['connect']}: #{line}\n" unless ignoring
         if line == 'BEGIN HYDRA'
           ignoring = false
         end
       end
-          $stdout.write "\n==== Hydra Running #{@name} COMPLETE ====\n\n"
+      passed
-        end
-      end
     end
   end
 

test/sync_test.rb

@@ -0,0 +1,113 @@
+require File.join(File.dirname(__FILE__), 'test_helper')
+
+class SyncTest < Test::Unit::TestCase
+  context "with a file to test and a destination to verify" do
+    setup do
+      # avoid having other tests interfering with us
+      sleep(0.2)
+      #FileUtils.rm_f(target_file)
+    end
+
+    teardown do
+      #FileUtils.rm_f(target_file)
+    end
+
+    should "synchronize a test file over ssh with rsync" do
+      local = File.join(Dir.tmpdir, 'hydra', 'local')
+      remote = File.join(Dir.tmpdir, 'hydra', 'remote')
+      sync_test = File.join(File.dirname(__FILE__), 'fixtures', 'sync_test.rb')
+      [local, remote].each{|f| FileUtils.rm_rf f; FileUtils.mkdir_p f}
+
+      # setup the folders:
+      # local:
+      #   - test_a
+      #   - test_c
+      # remote:
+      #   - test_b
+      #
+      # add test_c to exludes
+      FileUtils.cp(sync_test, File.join(local, 'test_a.rb'))
+      FileUtils.cp(sync_test, File.join(local, 'test_c.rb'))
+      FileUtils.cp(sync_test, File.join(remote, 'test_b.rb'))
+
+      # ensure a is not on remote
+      assert !File.exists?(File.join(remote, 'test_a.rb')), "A should not be on remote"
+      # ensure c is not on remote
+      assert !File.exists?(File.join(remote, 'test_c.rb')), "C should not be on remote"
+      # ensure b is on remote
+      assert File.exists?(File.join(remote, 'test_b.rb')),  "B should be on remote"
+
+      Hydra::Sync.new(
+        {
+          :type => :ssh,
+          :connect => 'localhost',
+          :directory => remote,
+          :runners => 1
+        },
+        {
+          :directory => local,
+          :exclude => ['test_c.rb']
+        }
+      )
+      # ensure a is copied
+      assert File.exists?(File.join(remote, 'test_a.rb')), "A was not copied"
+      # ensure c is not copied
+      assert !File.exists?(File.join(remote, 'test_c.rb')), "C was copied, should be excluded"
+      # ensure b is deleted
+      assert !File.exists?(File.join(remote, 'test_b.rb')), "B was not deleted"
+    end
+
+    should "synchronize a test file over ssh with rsync to multiple workers" do
+      local = File.join(Dir.tmpdir, 'hydra', 'local')
+      remote_a = File.join(Dir.tmpdir, 'hydra', 'remote_a')
+      remote_b = File.join(Dir.tmpdir, 'hydra', 'remote_b')
+      sync_test = File.join(File.dirname(__FILE__), 'fixtures', 'sync_test.rb')
+      [local, remote_a, remote_b].each{|f| FileUtils.rm_rf f; FileUtils.mkdir_p f}
+
+      # setup the folders:
+      # local:
+      #   - test_a
+      # remote_a:
+      #   - test_b
+      # remote_b:
+      #   - test_c
+      #
+      # add test_c to exludes
+      FileUtils.cp(sync_test, File.join(local, 'test_a.rb'))
+      FileUtils.cp(sync_test, File.join(remote_a, 'test_b.rb'))
+      FileUtils.cp(sync_test, File.join(remote_b, 'test_c.rb'))
+
+      # ensure a is not on remotes
+      assert !File.exists?(File.join(remote_a, 'test_a.rb')), "A should not be on remote_a"
+      assert !File.exists?(File.join(remote_b, 'test_a.rb')), "A should not be on remote_b"
+      # ensure b is on remote_a
+      assert File.exists?(File.join(remote_a, 'test_b.rb')),  "B should be on remote_a"
+      # ensure c is on remote_b
+      assert File.exists?(File.join(remote_b, 'test_c.rb')),  "C should be on remote_b"
+
+      Hydra::Sync.sync_many(
+        :workers => [{
+          :type => :ssh,
+          :connect => 'localhost',
+          :directory => remote_a,
+          :runners => 1
+        },
+        {
+          :type => :ssh,
+          :connect => 'localhost',
+          :directory => remote_b,
+          :runners => 1
+        }],
+        :sync => {
+          :directory => local
+        }
+      )
+      # ensure a is copied to both remotes
+      assert File.exists?(File.join(remote_a, 'test_a.rb')), "A was not copied to remote_a"
+      assert File.exists?(File.join(remote_b, 'test_a.rb')), "A was not copied to remote_b"
+      # ensure b and c are deleted from remotes
+      assert !File.exists?(File.join(remote_a, 'test_b.rb')), "B was not deleted from remote_a"
+      assert !File.exists?(File.join(remote_b, 'test_c.rb')), "C was not deleted from remote_b"
+    end
+  end
+end