Moved acceptance tests into tests/acceptance

tests/acceptance/fakes/dest/close_after_hello.rb

@@ -0,0 +1,22 @@
+#!/usr/bin/env ruby
+
+# Wait for a sender connection, send a correct hello, then disconnect.
+# Simulate a server which crashes after sending the hello.  We then
+# reopen the server socket to check that the sender retries: since the
+# command-line has gone away, and can't feed an error back to the
+# user, we have to keep trying.
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client = server.accept( "Timed out waiting for a connection" )
+client.write_hello
+client.close
+
+new_client = server.accept( "Timed out waiting for a reconnection" )
+new_client.close
+
+server.close
+
+exit 0

tests/acceptance/fakes/dest/error_on_write.rb

@@ -0,0 +1,21 @@
+#!/usr/bin/env ruby
+# encoding: utf-8
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client = server.accept
+
+client.write_hello
+handle = client.read_request[:handle]
+client.write_error( handle )
+
+
+client2 = server.accept( "Timed out waiting for a reconnection" )
+
+client.close
+client2.close
+server.close
+
+exit(0)

tests/acceptance/fakes/dest/hang_after_connect.rb

@@ -0,0 +1,21 @@
+#!/usr/bin/env ruby
+
+# Will open a server, accept a single connection, then sleep for 5
+# seconds.  After that time, the client should have disconnected,
+# which we can can't effectively check.
+#
+# This allows the test runner to check that the command-line sees the
+# right error message after the timeout time.
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client = server.accept( "Client didn't make a connection" )
+
+# Sleep for one second past the timeout (a bit of slop in case ruby
+# doesn't launch things quickly)
+sleep(FlexNBD::MS_HELLO_TIME_SECS + 1)
+
+client.close
+server.close

tests/acceptance/fakes/dest/hang_after_write.rb

@@ -0,0 +1,28 @@
+#!/usr/bin/env ruby
+# encoding: utf-8
+
+# Open a socket, say hello, receive a write, then sleep for >
+# MS_REQUEST_LIMIT_SECS seconds. This should tell the source that the
+# write has gone MIA, and we expect a reconnect.
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client1 = server.accept( server )
+client1.write_hello
+client1.read_request
+
+t = Thread.start do
+  client2 = server.accept( "Timed out waiting for a reconnection",
+                        FlexNBD::MS_REQUEST_LIMIT_SECS + 2 )
+  client2.close
+end
+
+sleep( FlexNBD::MS_REQUEST_LIMIT_SECS + 2 )
+client1.close
+
+t.join
+
+server.close
+exit(0)

tests/acceptance/fakes/dest/hello_wrong_magic.rb

@@ -0,0 +1,25 @@
+#!/usr/bin/env ruby
+
+# Simulate a destination which sends the wrong magic.
+# We expect the sender to disconnect and reconnect.
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client1 = server.accept
+
+# Launch a second thread so that we can spot the reconnection attempt
+# as soon as it happens, or alternatively die a flaming death on
+# timeout.
+t = Thread.new do
+  client2 = server.accept( "Timed out waiting for a reconnection",
+                           FlexNBD::MS_RETRY_DELAY_SECS + 1 )
+  client2.close
+end
+
+client1.write_hello( :magic => :wrong )
+
+t.join
+
+exit 0

tests/acceptance/fakes/dest/hello_wrong_size.rb

@@ -0,0 +1,23 @@
+#!/usr/bin/env ruby
+
+# Simulate a server which has a disc of the wrong size attached: send
+# a valid NBD hello with a random size, then check that we have see an
+# EOF on read.
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client = server.accept
+
+t = Thread.new do
+  client2 = server.accept( "Timed out waiting for a reconnection",
+                    FlexNBD::MS_RETRY_DELAY_SECS + 1 )
+  client2.close
+end
+
+client.write_hello( :size => :wrong )
+
+t.join
+
+exit 0

tests/acceptance/fakes/dest/reject_acl.rb

@@ -0,0 +1,13 @@
+#!/usr/bin/env ruby
+
+# Accept a connection, then immediately close it.  This simulates an ACL rejection.
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+server.accept.close
+
+server.close
+
+exit(0)

tests/acceptance/fakes/source/close_after_connect.rb

@@ -0,0 +1,24 @@
+#!/usr/bin/env ruby
+
+# Connects to the destination server, then immediately disconnects,
+# simulating a source crash.
+#
+# It then connects again, to check that the destination is still
+# listening.
+
+require 'flexnbd/fake_source'
+include FlexNBD::FakeSource
+
+addr, port = *ARGV
+
+
+connect( addr, port, "Failed to connect" ).close
+  # Sleep to be sure we don't try to connect too soon. That wouldn't
+  # be a problem for the destination, but it would prevent us from
+  # determining success or failure here in the case where we try to
+  # reconnect before the destination has tidied up after the first
+  # thread went away.
+sleep(0.5)
+connect( addr, port, "Failed to reconnect" ).close
+
+exit 0

tests/acceptance/fakes/source/close_after_hello.rb

@@ -0,0 +1,22 @@
+#!/usr/bin/env ruby
+
+# Connect, read the hello, then immediately disconnect.  This
+# simulates a sender which dislikes something in the hello message - a
+# wrong size, for instance.
+
+# After the disconnect, we reconnect to be sure that the destination
+# is still alive.
+
+require 'flexnbd/fake_source'
+include FlexNBD::FakeSource
+
+addr, port = *ARGV
+
+
+client_sock = connect( addr, port, "Timed out connecting." )
+read_hello( client_sock )
+client_sock.close
+sleep(0.2)
+connect( addr, port, "Timed out reconnecting." )
+
+exit(0)

tests/acceptance/fakes/source/connect_during_hello.rb

@@ -0,0 +1,19 @@
+#!/usr/bin/env ruby
+
+# Connect to the destination, then hang.  Connect a second time to the
+# destination.  This will trigger the destination's thread clearer.
+
+require 'flexnbd/fake_source'
+include FlexNBD::FakeSource
+
+addr, port = *ARGV
+
+# client_sock1 is a connection the destination is expecting.
+client_sock1 = connect( addr, port, "Timed out connecting" )
+sleep(0.25)
+client_sock2 = connect( addr, port, "Timed out connecting a second time" )
+
+# This is the expected source crashing after connect
+client_sock1.close
+# And this is just a tidy-up.
+client_sock2.close

tests/acceptance/fakes/source/connect_from_banned_ip.rb

@@ -0,0 +1,23 @@
+#!/usr/bin/env ruby
+# encoding: utf-8
+
+# We connect from a local address which should be blocked, sleep for a
+# bit, then try to read from the socket. We should get an instant EOF
+# as we've been cut off by the destination.
+
+require 'timeout'
+require 'flexnbd/fake_source'
+include FlexNBD::FakeSource
+
+addr, port = *ARGV
+sock = connect( addr, port, "Timed out connecting", "127.0.0.6" )
+sleep( 0.25 )
+Timeout.timeout( 2 ) do
+  fail "Not disconnected" if sock.read(1)
+end
+
+sock.close
+exit(0)
+
+
+

tests/acceptance/fakes/source/hang_after_hello.rb

@@ -0,0 +1,39 @@
+#!/usr/bin/env ruby
+
+# Simulate the hello message going astray, or the source hanging after
+# receiving it.
+#
+# We then connect again, to confirm that the destination is still
+# listening for an incoming migration.
+
+addr, port = *ARGV
+require "flexnbd/fake_source"
+include FlexNBD::FakeSource
+
+client_sock = connect( addr, port, "Timed out connecting" )
+read_hello( client_sock )
+
+# Now we do two things:
+
+# - In the parent process, we sleep for CLIENT_MAX_WAIT_SECS+5, which
+#   will make the destination give up and close the connection.
+# - In the child process, we sleep for CLIENT_MAX_WAIT_SECS+1, which
+#   should be able to reconnect despite the parent process not having
+#   closed its end yet.
+
+kidpid = fork do
+  client_sock.close
+  new_sock = nil
+  sleep( FlexNBD::CLIENT_MAX_WAIT_SECS + 1 )
+  new_sock = connect( addr, port, "Timed out reconnecting." )
+  read_hello( new_sock )
+  exit 0
+end
+
+# Sleep for longer than the child, to give the flexnbd process a bit
+# of slop
+sleep( FlexNBD::CLIENT_MAX_WAIT_SECS + 3 )
+client_sock.close
+
+_,status = Process.waitpid2( kidpid )
+exit status.exitstatus

tests/acceptance/fakes/source/write_out_of_range.rb

@@ -0,0 +1,21 @@
+#!/usr/bin/env ruby
+# encoding: utf-8
+
+# Connect, read the hello then make a write request with an impossible
+# (from,len) pair.  We expect an error response, and not to be
+# disconnected.
+
+require 'flexnbd/fake_source'
+include FlexNBD::FakeSource
+
+addr, port = *ARGV
+client_sock = connect( addr, port, "Timed out connecting" )
+read_hello( client_sock )
+write_write_request( client_sock, 1 << 31, 1 << 31, "myhandle" )
+response = read_response( client_sock )
+
+fail "Not an error" if response[:error] == 0
+fail "Wrong handle" unless "myhandle" == response[:handle]
+
+client_sock.close
+exit(0)

tests/acceptance/flexnbd.rb

@@ -0,0 +1,454 @@
+require 'socket'
+require 'thread'
+require 'open3'
+require 'timeout'
+require 'rexml/document'
+require 'rexml/streamlistener'
+
+Thread.abort_on_exception = true
+
+
+class Executor
+  attr_reader :pid
+
+  def run( cmd )
+    @pid = fork do exec cmd end
+  end
+end # class Executor
+
+
+class ValgrindExecutor
+  attr_reader :pid
+
+  def run( cmd )
+    @pid = fork do exec "valgrind --track-origins=yes #{cmd}" end
+  end
+end # class ValgrindExecutor
+
+
+class ValgrindKillingExecutor
+  attr_reader :pid
+
+  class Error
+    attr_accessor :what, :kind, :pid
+    attr_reader :backtrace
+    def initialize
+      @backtrace=[]
+      @what = ""
+      @kind = ""
+      @pid = ""
+    end
+
+    def add_frame
+      @backtrace << {}
+    end
+
+    def add_fn(fn)
+      @backtrace.last[:fn] = fn
+    end
+
+    def add_file(file)
+      @backtrace.last[:file] = file
+    end
+
+    def add_line(line)
+      @backtrace.last[:line] = line
+    end
+
+    def to_s
+      ([@what + " (#{@kind}) in #{@pid}"] + @backtrace.map{|h| "#{h[:file]}:#{h[:line]} #{h[:fn]}" }).join("\n")
+    end
+
+  end # class Error
+
+
+  class ErrorListener
+    include REXML::StreamListener
+    def initialize( killer )
+      @killer = killer
+      @error = Error.new
+      @found = false
+    end
+
+    def text( text )
+      @text = text
+    end
+
+    def tag_start(tag, attrs)
+      case tag.to_s
+      when "error"
+        @found = true
+      when "frame"
+        @error.add_frame
+      end
+    end
+
+    def tag_end(tag)
+      case tag.to_s
+      when "what"
+        @error.what = @text if @found
+        @text = ""
+      when "kind"
+        @error.kind = @text if @found
+      when "file"
+        @error.add_file( @text ) if @found
+      when "fn"
+        @error.add_fn( @text ) if @found
+      when "line"
+        @error.add_line( @text ) if @found
+      when "error", "stack"
+        @killer.call( @error )
+      when "pid"
+        @error.pid=@text
+      end
+    end
+  end # class ErrorListener
+
+
+  class DebugErrorListener < ErrorListener
+    def text( txt )
+      print txt
+      super( txt )
+    end
+
+    def tag_start( tag, attrs )
+      print "<#{tag}>"
+      super( tag, attrs )
+    end
+
+    def tag_end( tag )
+      print "</#{tag}>"
+      super( tag )
+    end
+  end
+
+
+  def initialize
+    @pid = nil
+  end
+
+  def run( cmd )
+    @io_r, io_w = IO.pipe
+    @pid = fork do exec( "valgrind --xml=yes --xml-fd=#{io_w.fileno} " + cmd ) end
+    launch_watch_thread( @pid, @io_r )
+    @pid
+  end
+
+
+  def call( err )
+    Process.kill( "KILL", @pid )
+    $stderr.puts "*"*72
+    $stderr.puts "* Valgrind error spotted:"
+    $stderr.puts err.to_s.split("\n").map{|s| "  #{s}"}
+    $stderr.puts "*"*72
+    exit(1)
+  end
+
+
+  private
+
+  def pick_listener
+    ENV['DEBUG'] ? DebugErrorListener : ErrorListener
+  end
+
+  def launch_watch_thread(pid, io_r)
+    Thread.start do
+      io_source = REXML::IOSource.new( io_r )
+      listener = pick_listener.new( self )
+      REXML::Document.parse_stream( io_source, listener )
+    end
+  end
+
+
+end # class ValgrindExecutor
+
+
+# Noddy test class to exercise FlexNBD from the outside for testing.
+#
+class FlexNBD
+  attr_reader :bin, :ctrl, :pid, :ip, :port
+
+  class << self
+    def counter
+      Dir['tmp/*'].select{|f| File.file?(f)}.length + 1
+    end
+  end
+
+  def pick_executor
+    kls = if ENV['VALGRIND']
+      if ENV['VALGRIND'] =~ /kill/
+        ValgrindKillingExecutor
+      else
+        ValgrindExecutor
+      end
+    else
+      Executor
+    end
+  end
+
+
+  def initialize(bin, ip, port)
+    @bin  = bin
+    @debug = (ENV['DEBUG'] && `#{@bin} serve --help` =~ /--verbose/) ? "--verbose" : ""
+    raise "#{bin} not executable" unless File.executable?(bin)
+    @executor = pick_executor.new
+    @ctrl = "/tmp/.flexnbd.ctrl.#{Time.now.to_i}.#{rand}"
+    @ip = ip
+    @port = port
+    @kill = []
+  end
+
+
+  def debug?
+    !@debug.empty? || ENV['DEBUG']
+  end
+
+  def debug( msg )
+    $stderr.puts msg if debug?
+  end
+
+
+  def serve_cmd( file, acl )
+    "#{bin} serve "\
+      "--addr #{ip} "\
+      "--port #{port} "\
+      "--file #{file} "\
+      "--sock #{ctrl} "\
+      "#{@debug} "\
+      "#{acl.join(' ')}"
+  end
+
+
+  def listen_cmd( file, acl )
+    "#{bin} listen "\
+      "--addr #{ip} "\
+      "--port #{port} "\
+      "--file #{file} "\
+      "--sock #{ctrl} "\
+      "#{@debug} "\
+      "#{acl.join(' ')}"
+  end
+
+
+  def read_cmd( offset, length )
+    "#{bin} read "\
+      "--addr #{ip} "\
+      "--port #{port} "\
+      "--from #{offset} "\
+      "#{@debug} "\
+      "--size #{length}"
+  end
+
+
+  def write_cmd( offset, data )
+    "#{bin} write "\
+      "--addr #{ip} "\
+      "--port #{port} "\
+      "--from #{offset} "\
+      "#{@debug} "\
+      "--size #{data.length}"
+  end
+
+
+  def mirror_cmd(dest_ip, dest_port)
+    "#{@bin} mirror "\
+      "--addr #{dest_ip} "\
+      "--port #{dest_port} "\
+      "--sock #{ctrl} "\
+      "#{@debug} "
+  end
+
+
+  def status_cmd
+    "#{@bin} status "\
+      "--sock #{ctrl} "\
+      "#{@debug}"
+  end
+
+  def acl_cmd( *acl )
+    "#{@bin} acl " \
+      "--sock #{ctrl} "\
+      "#{@debug} "\
+      "#{acl.join " "}"
+  end
+
+
+  def run_serve_cmd(cmd)
+    File.unlink(ctrl) if File.exists?(ctrl)
+    debug( cmd )
+
+    @pid = @executor.run( cmd )
+    start_wait_thread( @pid )
+
+    while !File.socket?(ctrl)
+      pid, status = Process.wait2(@pid, Process::WNOHANG)
+      raise "server did not start (#{cmd})" if pid
+      sleep 0.1
+    end
+    at_exit { kill }
+  end
+  private :run_serve_cmd
+
+
+  def serve( file, *acl)
+    run_serve_cmd( serve_cmd( file, acl ) )
+  end
+
+  def listen(file, *acl)
+    run_serve_cmd( listen_cmd( file, acl ) )
+  end
+
+
+  def start_wait_thread( pid )
+    @wait_thread = Thread.start do
+      _, status = Process.waitpid2( pid )
+
+      if @kill
+        fail "flexnbd quit with a bad status: #{status.exitstatus}" unless
+          @kill.include? status.exitstatus
+      else
+        $stderr.puts "flexnbd #{self.pid} quit"
+        fail "flexnbd #{self.pid} quit early with status #{status.to_i}"
+      end
+    end
+  end
+
+
+  def can_die(*status)
+    status = [0] if status.empty?
+    @kill += status
+  end
+
+  def kill
+    # At this point, to a certain degree we don't care what the exit
+    # status is
+    can_die(1)
+    if @pid
+      begin
+        Process.kill("INT", @pid)
+      rescue Errno::ESRCH => e
+        # already dead.  Presumably this means it went away after a
+        # can_die() call.
+      end
+    end
+    @wait_thread.join if @wait_thread
+  end
+
+  def read(offset, length)
+    cmd = read_cmd( offset, length )
+    debug( cmd )
+
+    IO.popen(cmd) do |fh|
+      return fh.read
+    end
+    raise IOError.new "NBD read failed" unless $?.success?
+    out
+  end
+
+  def write(offset, data)
+    cmd = write_cmd( offset, data )
+    debug( cmd )
+
+    IO.popen(cmd, "w") do |fh|
+      fh.write(data)
+    end
+    raise IOError.new "NBD write failed" unless $?.success?
+    nil
+  end
+
+
+  def join
+    @wait_thread.join
+  end
+
+
+  def mirror_unchecked( dest_ip, dest_port, bandwidth=nil, action=nil, timeout=nil )
+    cmd = mirror_cmd( dest_ip, dest_port)
+    debug( cmd )
+
+    maybe_timeout( cmd, timeout )
+  end
+
+
+  def maybe_timeout(cmd, timeout=nil )
+    stdout, stderr = "",""
+    run = Proc.new do
+      Open3.popen3( cmd ) do |io_in, io_out, io_err|
+        io_in.close
+        stdout.replace io_out.read
+        stderr.replace io_err.read
+      end
+    end
+
+    if timeout
+      Timeout.timeout(timeout, &run)
+    else
+      run.call
+    end
+
+    [stdout, stderr]
+  end
+
+
+  def mirror(dest_ip, dest_port, bandwidth=nil, action=nil)
+    stdout, stderr = mirror_unchecked( dest_ip, dest_port, bandwidth, action )
+    raise IOError.new( "Migrate command failed\n" + stderr) unless $?.success?
+
+    stdout
+  end
+
+
+  def acl(*acl)
+    cmd = acl_cmd( *acl )
+    debug( cmd )
+
+    maybe_timeout( cmd, 2 )
+  end
+
+
+  def status( timeout = nil )
+    cmd = status_cmd()
+    debug( cmd )
+
+    o,e = maybe_timeout( cmd, timeout )
+
+    [parse_status(o), e]
+  end
+
+
+  def launched?
+    !!@pid
+  end
+
+
+  protected
+  def control_command(*args)
+    raise "Server not running" unless @pid
+    args = args.compact
+    UNIXSocket.open(@ctrl) do |u|
+      u.write(args.join("\n") + "\n")
+      code, message = u.readline.split(": ", 2)
+      return [code, message]
+    end
+  end
+
+
+  def parse_status( status )
+    hsh = {}
+
+    status.split(" ").each do |part|
+      next if part.strip.empty?
+      a,b = part.split("=")
+      b.strip!
+      b = true if b == "true"
+      b = false if b == "false"
+
+      hsh[a.strip] = b
+    end
+
+    hsh
+  end
+
+
+end
+

tests/acceptance/flexnbd/constants.rb

@@ -0,0 +1,36 @@
+# encoding: utf-8
+
+module FlexNBD
+
+  # eeevil is his one and only name...
+  def self.read_constants
+    parents = []
+    current = File.expand_path(".")
+    while current != "/"
+      parents << current
+      current = File.expand_path( File.join( current, ".." ) )
+    end
+
+    source_root = parents.find do |dirname|
+      File.directory?( File.join( dirname, "src" ) )
+    end
+
+    fail "No source root!" unless source_root
+
+    headers = Dir[File.join( source_root, "src", "*.h" ) ]
+
+    headers.each do |header_filename|
+      txt_lines = File.readlines( header_filename )
+      txt_lines.each do |line|
+        if line =~ /^#\s*define\s+([A-Z0-9_]+)\s+(\d+)\s*$/
+          const_set($1, $2.to_i)
+        end
+      end
+    end
+
+  end
+
+  read_constants()
+end # module FlexNBD
+
+

tests/acceptance/flexnbd/fake_dest.rb

@@ -0,0 +1,127 @@
+# encoding: utf-8
+
+require 'socket'
+require 'timeout'
+
+require 'flexnbd/constants'
+
+module FlexNBD
+  class FakeDest
+
+    class Client
+      def initialize( sock )
+        @sock = sock
+      end
+
+
+      def write_hello( opts = {} )
+        self.class.write_hello( @sock, opts )
+      end
+
+      def read_request()
+        self.class.read_request( @sock )
+      end
+
+      def write_error( handle )
+        self.class.write_error( @sock, handle )
+      end
+
+      def close
+        @sock.close
+      end
+
+
+      def self.write_hello( client_sock, opts={} )
+        client_sock.write( "NBDMAGIC" )
+
+        if opts[:magic] == :wrong
+          client_sock.write( "\x00\x00\x42\x02\x81\x86\x12\x52" )
+        else
+          client_sock.write( "\x00\x00\x42\x02\x81\x86\x12\x53" )
+        end
+
+        if opts[:size] == :wrong
+          8.times do client_sock.write rand(256).chr end
+        else
+          client_sock.write( "\x00\x00\x00\x00\x00\x00\x10\x00" )
+        end
+
+        client_sock.write( "\x00" * 128 )
+      end
+
+
+
+      def self.read_request( client_sock )
+        req = client_sock.read(28)
+
+        magic_s  = req[0  ... 4 ]
+        type_s   = req[4  ... 8 ]
+        handle_s = req[8  ... 16]
+        from_s   = req[16 ... 24]
+        len_s    = req[24 ... 28]
+
+        {
+          :magic  => magic_s,
+          :type   => type_s.unpack("N").first,
+          :handle => handle_s,
+          :from   => parse_be64( from_s ),
+          :len    => len_s.unpack( "N").first
+        }
+      end
+
+
+      def self.parse_be64(str)
+        raise "String is the wrong length: 8 bytes expected (#{str.length} received)" unless
+          str.length == 8
+
+        top, bottom = str.unpack("NN")
+        (top << 32) + bottom
+      end
+
+
+      def self.write_error( client_sock, handle )
+        client_sock.write( "\x67\x44\x66\x98")
+        client_sock.write( "\x00\x00\x00\x01")
+        client_sock.write( handle )
+      end
+
+
+    end # class Client
+
+
+    def initialize( addr, port )
+      @sock = self.class.serve( addr, port )
+    end
+
+    def accept( err_msg = "Timed out waiting for a connection", timeout = 2)
+      Client.new( self.class.accept( @sock, err_msg, timeout ) )
+    end
+
+    def close
+      @sock.close
+    end
+
+
+    def self.serve( addr, port )
+      TCPServer.new( addr, port )
+    end
+
+
+    def self.accept( sock, err_msg, timeout )
+      client_sock = nil
+
+      begin
+        Timeout.timeout(timeout) do
+          client_sock = sock.accept
+        end
+      rescue Timeout::Error
+        $stderr.puts err_msg
+        exit 1
+      end
+
+      client_sock
+    end
+
+
+  end # module FakeDest
+end # module FlexNBD

tests/acceptance/flexnbd/fake_source.rb

@@ -0,0 +1,65 @@
+# encoding: utf-8
+
+require 'socket'
+require "timeout"
+require 'flexnbd/constants'
+
+module FlexNBD
+  module FakeSource
+
+    def connect( addr, port, err_msg, source_addr=nil, source_port=0 )
+      timing_out( 2, err_msg ) do
+        TCPSocket.new( addr, port, source_addr, source_port )
+      end
+    end
+
+
+    def read_hello( client_sock )
+      timing_out( FlexNBD::MS_HELLO_TIME_SECS,
+                  "Timed out waiting for hello." ) do
+          fail "No hello." unless (hello = client_sock.read( 152 )) &&
+            hello.length==152
+        hello
+      end
+    end
+
+
+    def write_write_request( client_sock, from, len, handle )
+      fail "Bad handle" unless handle.length == 8
+
+      client_sock.write( "\x25\x60\x95\x13" )
+      client_sock.write( "\x00\x00\x00\x01" )
+      client_sock.write( handle )
+      client_sock.write( "\x0"*4 )
+      client_sock.write( [from].pack( 'N' ) )
+      client_sock.write( [len ].pack( 'N' ) )
+    end
+
+
+    def read_response( client_sock )
+      magic = client_sock.read(4)
+      error_s = client_sock.read(4)
+      handle = client_sock.read(8)
+
+      {
+        :magic => magic,
+        :error => error_s.unpack("N"),
+        :handle => handle
+      }
+    end
+
+
+    def timing_out( time, msg )
+      begin
+        Timeout.timeout( time ) do
+          yield
+        end
+      rescue Timeout::Error
+        $stderr.puts msg
+        exit 1
+      end
+    end
+
+
+  end # module FakeSource
+end # module FlexNBD

tests/acceptance/nbd_scenarios

@@ -0,0 +1,364 @@
+#!/usr/bin/ruby
+
+require 'test/unit'
+require 'flexnbd'
+require 'test_file_writer'
+
+class Environment
+  attr_reader( :blocksize, :filename1, :filename2, :ip,
+               :port1, :port2, :nbd1, :nbd2, :file1, :file2 )
+
+  def initialize
+    @blocksize = 1024
+    @filename1 = "/tmp/.flexnbd.test.#{$$}.#{Time.now.to_i}.1"
+    @filename2 = "/tmp/.flexnbd.test.#{$$}.#{Time.now.to_i}.2"
+    @ip = "127.0.0.1"
+    @available_ports = [*40000..41000] - listening_ports
+    @port1 = @available_ports.shift
+    @port2 = @available_ports.shift
+    @nbd1 = FlexNBD.new("../../build/flexnbd", @ip, @port1)
+    @nbd2 = FlexNBD.new("../../build/flexnbd", @ip, @port2)
+
+    @fake_pid = nil
+  end
+
+
+  def serve1(*acl)
+    @nbd1.serve(@filename1, *acl)
+  end
+
+  def serve2(*acl)
+    @nbd2.serve(@filename2, *acl)
+  end
+
+
+  def listen1( *acl )
+    @nbd1.listen( @filename1, *(acl.empty? ? @acl1: acl) )
+  end
+
+  def listen2( *acl )
+    @nbd2.listen( @filename2, *acl )
+  end
+
+
+  def acl1( *acl )
+    @nbd1.acl( *acl )
+  end
+
+  def acl2( *acl )
+    @nbd2.acl( *acl )
+  end
+
+
+  def status1
+    @nbd1.status.first
+  end
+
+  def status2
+    @nbd2.status.first
+  end
+
+
+
+  def mirror12
+    @nbd1.mirror( @nbd2.ip, @nbd2.port )
+  end
+
+  def mirror12_unchecked
+    @nbd1.mirror_unchecked( @nbd2.ip, @nbd2.port, nil, nil, 10 )
+  end
+
+
+  def writefile1(data)
+    @file1 = TestFileWriter.new(@filename1, @blocksize).write(data)
+  end
+
+  def writefile2(data)
+    @file2 = TestFileWriter.new(@filename2, @blocksize).write(data)
+  end
+
+
+
+  def listening_ports
+    `netstat -ltn`.
+      split("\n").
+      map { |x| x.split(/\s+/) }[2..-1].
+      map { |l| l[3].split(":")[-1].to_i }
+  end
+
+
+  def cleanup
+    if @fake_pid
+      begin
+        Process.waitpid2( @fake_pid )
+      rescue Errno::ESRCH
+      end
+    end
+
+
+    @nbd1.kill
+    @nbd2.kill
+
+    [@filename1, @filename2].each do |f|
+      File.unlink(f) if File.exists?(f)
+    end
+  end
+
+
+  def run_fake( name, addr, port )
+    fakedir = File.join( File.dirname( __FILE__ ), "fakes" )
+    fake = Dir[File.join( fakedir, name ) + "*"].sort.find { |fn|
+      File.executable?( fn )
+    }
+
+    raise "no fake executable" unless fake
+    raise "no addr" unless addr
+    raise "no port" unless port
+    @fake_pid = fork do
+      exec fake + " " + addr.to_s + " " + port.to_s
+    end
+    sleep(0.5)
+  end
+
+
+  def fake_reports_success
+    _,status = Process.waitpid2( @fake_pid )
+    @fake_pid = nil
+    status.success?
+  end
+
+
+end # class Environment
+
+
+class NBDScenarios < Test::Unit::TestCase
+  def setup
+    @env = Environment.new
+  end
+
+  def teardown
+    @env.nbd1.can_die(0)
+    @env.nbd2.can_die(0)
+    @env.cleanup
+  end
+
+
+  def test_read1
+    @env.writefile1("f"*64)
+    @env.serve1
+
+    [0, 12, 63].each do |num|
+
+      assert_equal(
+        @env.nbd1.read(num*@env.blocksize, @env.blocksize),
+        @env.file1.read(num*@env.blocksize, @env.blocksize)
+      )
+    end
+
+    [124, 1200, 10028, 25488].each do |num|
+      assert_equal(@env.nbd1.read(num, 4), @env.file1.read(num, 4))
+    end
+  end
+
+  # Check that we're not
+  #
+  def test_writeread1
+    @env.writefile1("0"*64)
+    @env.serve1
+
+    [0, 12, 63].each do |num|
+      data = "X"*@env.blocksize
+      @env.nbd1.write(num*@env.blocksize, data)
+      assert_equal(data, @env.file1.read(num*@env.blocksize, data.size))
+      assert_equal(data, @env.nbd1.read(num*@env.blocksize, data.size))
+    end
+  end
+
+  # Check that we're not overstepping or understepping where our writes end
+  # up.
+  #
+  def test_writeread2
+    @env.writefile1("0"*1024)
+    @env.serve1
+
+    d0 = "\0"*@env.blocksize
+    d1 = "X"*@env.blocksize
+    (0..63).each do |num|
+      @env.nbd1.write(num*@env.blocksize*2, d1)
+    end
+    (0..63).each do |num|
+      assert_equal(d0, @env.nbd1.read(((2*num)+1)*@env.blocksize, d0.size))
+    end
+  end
+
+
+  def test_mirror
+    @env.writefile1( "f"*4 )
+    @env.serve1
+
+    @env.writefile2( "0"*4 )
+    @env.listen2
+
+    @env.nbd1.can_die
+    stdout, stderr = @env.mirror12
+
+    @env.nbd1.join
+
+    assert_equal(@env.file1.read_original( 0, @env.blocksize ),
+                 @env.file2.read( 0, @env.blocksize ) )
+    assert @env.status2['has_control'], "destination didn't take control"
+  end
+
+end
+
+
+class NBDConnectSourceFailureScenarios < Test::Unit::TestCase
+  def setup
+    @env = Environment.new
+    @env.writefile1( "f" * 4 )
+    @env.serve1
+
+  end
+
+  def teardown
+    @env.nbd1.can_die(0)
+    @env.cleanup
+  end
+
+
+  def test_failure_to_connect_reported_in_mirror_cmd_response
+    stdout, stderr = @env.mirror12_unchecked
+    assert_match( /failed to connect/, stderr )
+  end
+
+
+  def test_destination_hangs_after_connect_reports_error_at_source
+    run_fake( "dest/hang_after_connect" )
+
+    stdout, stderr = @env.mirror12_unchecked
+    assert_match( /Remote server failed to respond/, stderr )
+    assert_success
+  end
+
+
+  def test_destination_rejects_connection_reports_error_at_source
+    run_fake( "dest/reject_acl" )
+
+    stdout, stderr = @env.mirror12_unchecked
+    assert_match /Mirror was rejected/, stderr
+    assert_success
+  end
+
+  def test_wrong_size_causes_disconnect
+    run_fake( "dest/hello_wrong_size" )
+    stdout, stderr = @env.mirror12_unchecked
+    assert_match /Remote size does not match local size/, stderr
+    assert_success
+  end
+
+
+  def test_wrong_magic_causes_disconnect
+    run_fake( "dest/hello_wrong_magic" )
+    stdout, stderr = @env.mirror12_unchecked
+    assert_match /Mirror was rejected/, stderr
+    assert_success "dest/hello_wrong_magic fake failed"
+  end
+
+
+  def test_disconnect_after_hello_causes_retry
+    run_fake( "dest/close_after_hello" )
+    stdout, stderr = @env.mirror12_unchecked
+    assert_match( /Mirror started/, stdout )
+
+    assert_success
+  end
+
+
+  def test_write_times_out_causes_retry
+    run_fake( "dest/hang_after_write" )
+    stdout, stderr = @env.mirror12_unchecked
+
+    assert_success
+  end
+
+
+  def test_rejected_write_causes_retry
+    run_fake( "dest/error_on_write" )
+    stdout, stderr = @env.mirror12_unchecked
+    assert_success
+  end
+
+
+  private
+  def run_fake(name)
+    @env.run_fake( name, @env.ip, @env.port2 )
+  end
+
+  def assert_success( msg=nil )
+    assert @env.fake_reports_success, msg || "Fake failed"
+  end
+
+
+end
+
+
+class NBDConnectDestFailureScenarios < Test::Unit::TestCase
+
+  def setup
+    @env = Environment.new
+    @env.writefile1( "0" * 4 )
+    @env.listen1
+  end
+
+  def teardown
+    @env.cleanup
+  end
+
+
+  def test_hello_blocked_by_disconnect_causes_error_not_fatal
+    run_fake( "source/close_after_connect" )
+    assert_no_control
+  end
+
+
+  def test_hello_goes_astray_causes_timeout_error
+    run_fake( "source/hang_after_hello" )
+    assert_no_control
+  end
+
+
+  def test_disconnect_after_hello_causes_error_not_fatal
+    run_fake( "source/close_after_hello" )
+    assert_no_control
+  end
+
+
+  def test_double_connect_during_hello
+    run_fake( "source/connect_during_hello" )
+  end
+
+
+  def test_acl_rejection
+    @env.acl1("127.0.0.1")
+    run_fake( "source/connect_from_banned_ip")
+  end
+
+
+  def test_bad_write
+    run_fake( "source/write_out_of_range" )
+  end
+
+
+  private
+  def run_fake( name )
+    @env.run_fake( name, @env.ip, @env.port1 )
+    assert @env.fake_reports_success, "#{name} failed."
+  end
+
+  def assert_no_control
+    status, stderr = @env.status1
+    assert !status['has_control'],  "Thought it had control"
+  end
+
+
+end # class NBDConnectDestFailureScenarios

tests/acceptance/test_file_writer.rb

@@ -0,0 +1,123 @@
+# Noddy test class for writing files to disc in predictable patterns
+# in order to test FlexNBD.
+#
+class TestFileWriter
+  def initialize(filename, blocksize)
+    @fh = File.open(filename, "w+")
+    @blocksize = blocksize
+    @pattern = ""
+  end
+  
+  # We write in fixed block sizes, given by "blocksize" 
+  #    _ means skip a block
+  #    0 means write a block full of zeroes
+  #    f means write a block with the file offset packed every 4 bytes
+  #    
+  def write(data)
+    @pattern += data
+    
+    data.split("").each do |code|
+      if code == "_"
+        @fh.seek(@blocksize, IO::SEEK_CUR)
+      else
+        @fh.write(data(code))
+      end
+    end
+    @fh.flush
+    self
+  end
+  
+ 
+  # Returns what the data ought to be at the given offset and length
+  #
+  def read_original( off, len )
+    patterns = @pattern.split( "" )
+    patterns.zip( (0...patterns.length).to_a ).
+      map { |blk, blk_off|
+      data(blk, blk_off)
+    }.join[off...(off+len)]
+  end
+  
+  # Read what's actually in the file
+  #
+  def read(off, len)
+    @fh.seek(off, IO::SEEK_SET)
+    @fh.read(len)
+  end
+  
+  def untouched?(offset, len)
+    read(offset, len) == read_original(offset, len)
+  end
+  
+  def close
+    @fh.close
+    nil
+  end
+  
+  protected
+  
+  def data(code, at=@fh.tell)
+    case code
+      when "0", "_"
+        "\0" * @blocksize
+      when "X"
+        "X" * @blocksize
+      when "f"
+        r = ""
+        (@blocksize/4).times do
+          r += [at].pack("I")
+          at += 4
+        end
+        r
+      else
+        raise "Unknown character '#{block}'"
+    end    
+  end
+  
+end
+
+if __FILE__==$0
+  require 'tempfile'
+  require 'test/unit'
+
+  class TestFileWriterTest < Test::Unit::TestCase
+    def test_read_original_zeros
+      Tempfile.open("test_read_original_zeros") do |tempfile|
+        tempfile.close
+        file = TestFileWriter.new( tempfile.path, 4096 )
+        file.write( "0" )
+        assert_equal file.read( 0, 4096 ), file.read_original( 0, 4096 )
+        assert( file.untouched?(0,4096) , "Untouched file was touched." )
+      end
+    end
+
+    def test_read_original_offsets
+      Tempfile.open("test_read_original_offsets") do |tempfile|
+        tempfile.close
+        file = TestFileWriter.new( tempfile.path, 4096 )
+        file.write( "f" )
+        assert_equal file.read( 0, 4096 ), file.read_original( 0, 4096 )
+        assert( file.untouched?(0,4096) , "Untouched file was touched." )
+      end
+    end
+
+    def test_file_size
+      Tempfile.open("test_file_size") do |tempfile|
+        tempfile.close
+        file = TestFileWriter.new( tempfile.path, 4096 )
+        file.write( "f" )
+        assert_equal 4096, File.stat( tempfile.path ).size
+      end
+    end
+
+    def test_read_original_size
+      Tempfile.open("test_read_original_offsets") do |tempfile|
+        tempfile.close
+        file = TestFileWriter.new( tempfile.path, 4)
+        file.write( "f"*4 )
+        assert_equal 4, file.read_original(0, 4).length
+      end
+    end
+  end
+end
+