# -*- encoding: binary -*-
require './test/setup'
require 'stringio'
require 'tempfile'
require 'fileutils'

class TestMogileFS__MogileFS < TestMogileFS
  def setup
    @klass = MogileFS::MogileFS
    super
  end

  def test_initialize
    assert_equal 'test', @client.domain

    assert_raises ArgumentError do
      MogileFS::MogileFS.new :hosts => ['kaa:6001']
    end
  end

  def test_get_file_data_http
    tmp = Tempfile.new('accept')
    accept = File.open(tmp.path, "ab")
    svr = Proc.new do |serv, port|
      client, _ = serv.accept
      client.sync = true
      readed = client.recv(4096, 0)
      assert(readed =~ \
            %r{\AGET /dev[12]/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
      accept.syswrite('.')
      client.send("HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\ndata!", 0)
      client.close
    end
    t1 = TempServer.new(svr)
    t2 = TempServer.new(svr)
    path1 = "http://127.0.0.1:#{t1.port}/dev1/0/000/000/0000000062.fid"
    path2 = "http://127.0.0.1:#{t2.port}/dev2/0/000/000/0000000062.fid"

    @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }

    assert_equal 'data!', @client.get_file_data('key')
    assert_equal 1, accept.stat.size
    ensure
      TempServer.destroy_all!
  end

  def test_get_file_data_http_not_found_failover
    tmp = Tempfile.new('accept')
    accept = File.open(tmp.path, 'ab')
    svr1 = Proc.new do |serv, port|
      client, _ = serv.accept
      client.sync = true
      readed = client.recv(4096, 0)
      assert(readed =~ \
            %r{\AGET /dev1/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
      accept.syswrite('.')
      client.send("HTTP/1.0 404 Not Found\r\n\r\ndata!", 0)
      client.close
    end

    svr2 = Proc.new do |serv, port|
      client, _ = serv.accept
      client.sync = true
      readed = client.recv(4096, 0)
      assert(readed =~ \
            %r{\AGET /dev2/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
      accept.syswrite('.')
      client.send("HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\ndata!", 0)
      client.close
    end

    t1 = TempServer.new(svr1)
    t2 = TempServer.new(svr2)
    path1 = "http://127.0.0.1:#{t1.port}/dev1/0/000/000/0000000062.fid"
    path2 = "http://127.0.0.1:#{t2.port}/dev2/0/000/000/0000000062.fid"
    @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }

    assert_equal 'data!', @client.get_file_data('key')
    assert_equal 2, accept.stat.size
    ensure
      TempServer.destroy_all!
  end

  def test_get_file_data_http_block
    tmpfp = Tempfile.new('test_mogilefs.open_data')
    nr = nr_chunks
    chunk_size = 1024 * 1024
    expect_size = nr * chunk_size
    header = "HTTP/1.0 200 OK\r\n" \
             "Content-Length: #{expect_size}\r\n\r\n"
    assert_equal header.size, tmpfp.syswrite(header)
    nr.times { assert_equal chunk_size, tmpfp.syswrite(' ' * chunk_size) }
    assert_equal expect_size + header.size, File.size(tmpfp.path)
    tmpfp.sysseek(0)

    accept = Tempfile.new('accept')
    svr = Proc.new do |serv, port|
      client, _ = serv.accept
      client.sync = true
      accept.syswrite('.')
      readed = client.recv(4096, 0)
      assert(readed =~ \
            %r{\AGET /dev[12]/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
      MogileFS.io.copy_stream(tmpfp, client)
      client.close
      exit 0
    end
    t1 = TempServer.new(svr)
    t2 = TempServer.new(svr)
    path1 = "http://127.0.0.1:#{t1.port}/dev1/0/000/000/0000000062.fid"
    path2 = "http://127.0.0.1:#{t2.port}/dev2/0/000/000/0000000062.fid"

    @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }

    data = Tempfile.new('test_mogilefs.dest_data')
    read_nr = nr = 0
    @client.get_file_data('key') do |fp|
      buf = ''
      loop do
        begin
          fp.sysread(16384, buf)
          read_nr = buf.size
          nr += read_nr
          assert_equal read_nr, data.syswrite(buf), "partial write"
        rescue Errno::EAGAIN
          retry
        rescue EOFError
          break
        end
      end
    end
    assert_equal expect_size, nr, "size mismatch"
    assert_equal 1, accept.stat.size
  end

  def test_get_paths
    path1 = 'http://rur-1/dev1/0/000/000/0000000062.fid'
    path2 = 'http://rur-2/dev2/0/000/000/0000000062.fid'

    @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }

    expected = [ path1, path2 ]

    assert_equal expected, @client.get_paths('key').sort
  end

  def test_get_uris
    path1 = 'http://rur-1/dev1/0/000/000/0000000062.fid'
    path2 = 'http://rur-2/dev2/0/000/000/0000000062.fid'

    @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }

    expected = [ URI.parse(path1), URI.parse(path2) ]

    assert_equal expected, @client.get_uris('key')
  end


  def test_get_paths_unknown_key
    @backend.get_paths = ['unknown_key', '']

    assert_raises MogileFS::Backend::UnknownKeyError do
      assert_equal nil, @client.get_paths('key')
    end
  end

  def test_delete_existing
    @backend.delete = { }
    assert_nothing_raised do
      @client.delete 'no_such_key'
    end
  end

  def test_delete_nonexisting
    @backend.delete = 'unknown_key', ''
    assert_raises MogileFS::Backend::UnknownKeyError do
      @client.delete('no_such_key')
    end
  end

  def test_delete_readonly
    @client.readonly = true
    assert_raises MogileFS::ReadOnlyError do
      @client.delete 'no_such_key'
    end
  end

  def test_each_key
    @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_2',
                           'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }
    @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_4',
                           'key_1' => 'new_key_3', 'key_2' => 'new_key_4' }
    @backend.list_keys = { 'key_count' => 0, 'next_after' => 'new_key_4' }
    keys = []
    @client.each_key 'new' do |key|
      keys << key
    end

    assert_equal %w[new_key_1 new_key_2 new_key_3 new_key_4], keys
  end

  def test_list_keys
    @backend.list_keys = { 'key_count' => '2', 'next_after' => 'new_key_2',
                           'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }

    keys, next_after = @client.list_keys 'new'
    assert_equal ['new_key_1', 'new_key_2'], keys.sort
    assert_equal 'new_key_2', next_after
  end

  def test_new_file_http
    @client.readonly = true
    assert_raises MogileFS::ReadOnlyError do
      @client.new_file 'new_key', 'test'
    end
  end

  def test_new_file_readonly
    @client.readonly = true
    assert_raises MogileFS::ReadOnlyError do
      @client.new_file 'new_key', 'test'
    end
  end

  def test_store_file_small_http
    received = Tempfile.new('received')
    to_store = Tempfile.new('small')
    to_store.syswrite('data')

    expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"
    t = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      client.sync = true
      received.syswrite(client.read(expected.bytesize))
      client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'devid' => '1',
      'path' => "http://127.0.0.1:#{t.port}/path",
    }
    nr = @client.store_file 'new_key', 'test', to_store.path
    assert_equal 4, nr
    received.sysseek(0)
    assert_equal expected, received.sysread(4096)
    ensure
      TempServer.destroy_all!
  end

  def test_store_content_http
    received = Tempfile.new('received')
    expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"

    t = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      client.sync = true
      seen = ""
      while seen !~ /\r\n\r\ndata/
        buf = client.readpartial(4096)
        seen << buf
        received.syswrite(buf)
      end
      client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'devid' => '1',
      'path' => "http://127.0.0.1:#{t.port}/path",
    }

    nr = @client.store_content 'new_key', 'test', 'data'
    assert nr
    assert_equal 4, nr

    received.sysseek(0)
    assert_equal expected, received.sysread(4096)
    ensure
      TempServer.destroy_all!
  end


  def test_store_content_with_writer_callback
    received = Tempfile.new('received')
    expected = "PUT /path HTTP/1.0\r\nContent-Length: 40\r\n\r\n"
    10.times do
      expected += "data"
    end
    t = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      client.sync = true
      nr = 0
      loop do
        buf = client.readpartial(8192) or break
        break if buf.length == 0
        assert_equal buf.length, received.syswrite(buf)
        nr += buf.length
        break if nr >= expected.size
      end
      client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'devid' => '1',
      'path' => "http://127.0.0.1:#{t.port}/path",
    }

    cbk = MogileFS::Util::StoreContent.new(40) do |write_callback|
      10.times do
        write_callback.call("data")
      end
    end
    assert_equal 40, cbk.length
    nr = @client.store_content('new_key', 'test', cbk)
    assert_equal 40, nr

    received.sysseek(0)
    assert_equal expected, received.sysread(4096)
    ensure
      TempServer.destroy_all!
  end

  def test_store_content_multi_dest_failover_path
    test_store_content_multi_dest_failover(true)
  end

  def test_store_content_multi_dest_failover(big_io = false)
    received1 = Tempfile.new('received')
    received2 = Tempfile.new('received')
    expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"

    t1 = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      seen = ""
      while seen !~ /\r\n\r\ndata/
        buf = client.readpartial(4096)
        seen << buf
        received1.syswrite(buf)
      end
      client.send("HTTP/1.0 500 Internal Server Error\r\n\r\n", 0)
      client.close
    end)

    t2 = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      seen = ""
      while seen !~ /\r\n\r\ndata/
        buf = client.readpartial(4096)
        seen << buf
        received2.syswrite(buf)
      end
      client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'dev_count' => '2',
      'devid_1' => '1',
      'path_1' => "http://127.0.0.1:#{t1.port}/path",
      'devid_2' => '2',
      'path_2' => "http://127.0.0.1:#{t2.port}/path",
    }

    if big_io
      tmp = Tempfile.new('data')
      tmp.sync = true
      tmp.write 'data'
      nr = @client.store_file('new_key', 'test', tmp.path)
      tmp.close!
    else
      nr = @client.store_content 'new_key', 'test', 'data'
    end
    assert_equal 4, nr
    received1.sysseek(0)
    received2.sysseek(0)
    assert_equal expected, received1.sysread(4096)
    assert_equal expected, received2.sysread(4096)
    ensure
      TempServer.destroy_all!
  end

  def test_store_content_http_fail
    t = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      client.sync = true
      client.recv(4096, 0)
      client.send("HTTP/1.0 500 Internal Server Error\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'devid' => '1',
      'path' => "http://127.0.0.1:#{t.port}/path",
    }

    assert_raises MogileFS::HTTPFile::NoStorageNodesError do
      @client.store_content 'new_key', 'test', 'data'
    end
  end

  def test_store_content_http_empty
    received = Tempfile.new('received')
    expected = "PUT /path HTTP/1.0\r\nContent-Length: 0\r\n\r\n"
    t = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      client.sync = true
      received.syswrite(client.recv(4096, 0))
      client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'devid' => '1',
      'path' => "http://127.0.0.1:#{t.port}/path",
    }

    nr = @client.store_content 'new_key', 'test', ''
    assert_equal 0, nr
    received.sysseek(0)
    assert_equal expected, received.sysread(4096)
  end

  def test_store_content_nfs
    @backend.create_open = {
      'dev_count' => '1',
      'devid_1' => '1',
      'path_1' => '/path',
    }
    assert_raises MogileFS::UnsupportedPathError do
      @client.store_content 'new_key', 'test', 'data'
    end
  end

  def test_new_file_http_large
    expect = Tempfile.new('test_mogilefs.expect')
    to_put = Tempfile.new('test_mogilefs.to_put')
    received = Tempfile.new('test_mogilefs.received')

    nr = nr_chunks
    chunk_size = 1024 * 1024
    expect_size = nr * chunk_size

    header = "PUT /path HTTP/1.0\r\n" \
             "Content-Length: #{expect_size}\r\n\r\n"
    assert_equal header.size, expect.syswrite(header)
    nr.times do
      assert_equal chunk_size, expect.syswrite(' ' * chunk_size)
      assert_equal chunk_size, to_put.syswrite(' ' * chunk_size)
    end
    assert_equal expect_size + header.size, expect.stat.size
    assert_equal expect_size, to_put.stat.size

    readed = Tempfile.new('readed')
    t = TempServer.new(Proc.new do |serv, accept|
      client, _ = serv.accept
      client.sync = true
      nr = 0
      loop do
        buf = client.readpartial(8192) or break
        break if buf.length == 0
        assert_equal buf.length, received.syswrite(buf)
        nr += buf.length
        break if nr >= expect.stat.size
      end
      readed.syswrite("#{nr}")
      client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
      client.close
    end)

    @backend.create_open = {
      'devid' => '1',
      'path' => "http://127.0.0.1:#{t.port}/path",
    }

    orig_size = to_put.size
    nr = @client.store_file('new_key', 'test', to_put.path)
    assert nr, nr.inspect
    assert_equal orig_size, nr
    assert_equal orig_size, to_put.size
    readed.sysseek(0)
    assert_equal expect.stat.size, readed.sysread(4096).to_i

    ENV['PATH'].split(/:/).each do |path|
      cmp_bin = "#{path}/cmp"
      File.executable?(cmp_bin) or next
      # puts "running #{cmp_bin} #{expect.path} #{received.path}"
      assert( system(cmp_bin, expect.path, received.path) )
      break
    end

    ensure
      TempServer.destroy_all!
  end

  def test_store_content_readonly
    @client.readonly = true

    assert_raises MogileFS::ReadOnlyError do
      @client.store_content 'new_key', 'test', nil
    end
  end

  def test_store_file_readonly
    @client.readonly = true
    assert_raises MogileFS::ReadOnlyError do
      @client.store_file 'new_key', 'test', nil
    end
  end

  def test_rename_existing
    @backend.rename = {}

    assert_nil @client.rename('from_key', 'to_key')
  end

  def test_rename_nonexisting
    @backend.rename = 'unknown_key', ''

    assert_raises MogileFS::Backend::UnknownKeyError do
      @client.rename('from_key', 'to_key')
    end
  end

  def test_rename_no_key
    @backend.rename = 'no_key', 'no_key'

    e = assert_raises MogileFS::Backend::NoKeyError do
      @client.rename 'new_key', 'test'
    end

    assert_equal 'no_key', e.message
  end

  def test_rename_readonly
    @client.readonly = true

    e = assert_raises MogileFS::ReadOnlyError do
      @client.rename 'new_key', 'test'
    end

    assert_equal 'readonly mogilefs', e.message
  end

  def assert_get_paths_args(expect, *args)
    sock = TCPServer.new("127.0.0.1", 0)
    nargs = { :hosts => [ "127.0.0.1:#{sock.addr[1]}" ], :domain => "foo" }
    c = MogileFS::MogileFS.new(nargs)
    received = []
    th = Thread.new do
      a = sock.accept
      line = a.gets
      received << line
      a.write("OK paths=2&path1=http://0/a&path2=http://0/b\r\n")
      a.close
    end
    paths_expect = %w(http://0/a http://0/b)
    assert_equal paths_expect, c.get_paths("f", *args)
    th.join
    assert_equal 1, received.size
    tmp = c.backend.url_decode(received[0].split(/\s+/)[1])
    assert_equal "f", tmp.delete("key")
    assert_equal "foo", tmp.delete("domain")
    assert_equal expect, tmp
    c.backend.shutdown
    ensure
      sock.close
  end

  def test_get_paths_args
    assert_get_paths_args({"noverify"=>"1", "zone"=>""})
    assert_get_paths_args({"noverify"=>"0", "zone"=>""}, false)
    assert_get_paths_args({"noverify"=>"0", "zone"=>""}, :noverify=>false)
    assert_get_paths_args({"noverify"=>"1", "zone"=>"alt"}, true, "alt")
    assert_get_paths_args({"noverify"=>"1", "zone"=>"alt"},
                          {:noverify => true, :zone => "alt"})
    assert_get_paths_args({"noverify"=>"1", "zone"=>"alt","pathcount"=>"666"},
                          {:noverify => true, :zone => "alt", :pathcount=>666})
  end

  def test_idempotent_command_eof
    ip = "127.0.0.1"
    a, b = TCPServer.new(ip, 0), TCPServer.new(ip, 0)
    hosts = [ "#{ip}:#{a.addr[1]}", "#{ip}:#{b.addr[1]}" ]
    args = { :hosts => hosts, :domain => "foo" }
    c = MogileFS::MogileFS.new(args)
    received = []
    th = Thread.new do
      r = IO.select([a, b])
      x = r[0][0].accept
      received << x.gets
      x.close

      r = IO.select([a, b])
      x = r[0][0].accept
      received << x.gets
      x.write("OK paths=2&path1=http://0/a&path2=http://0/b\r\n")
      x.close
    end
    expect = %w(http://0/a http://0/b)
    assert_equal expect, c.get_paths("f")
    th.join
    assert_equal 2, received.size
    assert_equal received[0], received[1]
  end

  def test_idempotent_command_slow
    ip = "127.0.0.1"
    a = TCPServer.new(ip, 0)
    hosts = [ "#{ip}:#{a.addr[1]}" ]
    q = Queue.new
    timeout = 1
    args = { :hosts => hosts, :domain => "foo", :timeout => timeout }
    c = MogileFS::MogileFS.new(args)
    received = []
    secs = timeout + 1
    th = Thread.new do
      x = a.accept
      line = x.gets
      %r{key=(\w+)} =~ line
      sleep(secs)
      begin
        x.write("OK paths=1&path1=http://0/#{$1}\r\n")
      rescue Errno::EPIPE
        # EPIPE may or may not get raised due to timing issue,
        # we don't care either way
      rescue => e
        flunk("#{e.message} (#{e.class})")
      ensure
        x.close
      end
      q << :continue_test

      x = a.accept
      line = x.gets
      %r{key=(\w+)} =~ line
      begin
        x.write("OK paths=1&path1=http://0/#{$1}\r\n")
      rescue => e
        flunk("#{e.message} (#{e.class})")
      ensure
        x.close
      end
    end
    assert_raises(MogileFS::UnreadableSocketError) do
      c.get_paths("a")
    end
    assert_equal :continue_test, q.pop, "avoid race during test"
    expect2 = %w(http://0/b)
    assert_equal expect2, c.get_paths("b")
    a.close
    th.join
  end

  def test_idempotent_command_response_truncated
    ip = "127.0.0.1"
    a, b = TCPServer.new(ip, 0), TCPServer.new(ip, 0)
    hosts = [ "#{ip}:#{a.addr[1]}", "#{ip}:#{b.addr[1]}" ]
    args = { :hosts => hosts, :domain => "foo" }
    c = MogileFS::MogileFS.new(args)
    received = []
    th = Thread.new do
      r = IO.select([a, b])
      x = r[0][0].accept
      received << x.gets
      x.write("OK paths=2&path1=http://0/a&path2=http://0/")
      x.close

      r = IO.select([a, b])
      x = r[0][0].accept
      received << x.gets
      x.write("OK paths=2&path1=http://0/a&path2=http://0/b\r\n")
      x.close
    end
    expect = %w(http://0/a http://0/b)
    assert_equal expect, c.get_paths("f")
    th.join
    assert_equal 2, received.size
    assert_equal received[0], received[1]
  end

  def test_non_idempotent_command_eof
    ip = "127.0.0.1"
    a, b = TCPServer.new(ip, 0), TCPServer.new(ip, 0)
    hosts = [ "#{ip}:#{a.addr[1]}", "#{ip}:#{b.addr[1]}" ]
    args = { :hosts => hosts, :domain => "foo" }
    c = MogileFS::MogileFS.new(args)
    received = []
    th = Thread.new do
      r = IO.select([a, b])
      x = r[0][0].accept
      received << x.gets
      x.close
    end
    assert_raises(EOFError) { c.rename("a", "b") }
    th.join
    assert_equal 1, received.size
  end

  def test_list_keys_verbose_ordering # implementation detail
    received = []
    sock = TCPServer.new("127.0.0.1", 0)
    nargs = { :hosts => [ "127.0.0.1:#{sock.addr[1]}" ], :domain => "foo" }
    c = MogileFS::MogileFS.new(nargs)
    th = Thread.new do
      a = sock.accept
      %w(a b c d e).each do |key|
        line = a.gets
        cmd, args = line.split(/\s+/, 2)
        args = c.backend.url_decode(args.strip)
        assert_equal "file_info", cmd
        assert_equal key, args["key"]
      end
      out = { "length" => 3, "devcount" => 6 }
      %w(a b c d e).shuffle.each do |key|
        out["key"] = key
        a.write "OK #{c.backend.url_encode(out)}\r\n"
      end
      a.close
    end

    blk = lambda do |key, length, devcount|
      received << [ key, length, devcount ]
    end
    c.list_keys_verbose(%w(a b c d e), blk)
    th.join
    received.map! { |(key,_,_)| key }
    assert_equal %w(a b c d e), received
    ensure
      sock.close
  end

  def test_list_keys_verbose_retry_eof # implementation detail
    received = []
    sock = TCPServer.new("127.0.0.1", 0)
    nargs = { :hosts => [ "127.0.0.1:#{sock.addr[1]}" ], :domain => "foo" }
    c = MogileFS::MogileFS.new(nargs)
    th = Thread.new do
      a = sock.accept
      %w(a b c d e).each do |key|
        line = a.gets
        cmd, args = line.split(/\s+/, 2)
        args = c.backend.url_decode(args.strip)
        assert_equal "file_info", cmd
        assert_equal key, args["key"]
      end
      out = { "length" => 3, "devcount" => 6 }
      %w(d e).each do |key|
        out["key"] = key
        a.write "OK #{c.backend.url_encode(out)}\r\n"
      end
      a.close # trigger EOF
      a = sock.accept # client will retry
      %w(a b c).each do |key|
        line = a.gets
        cmd, args = line.split(/\s+/, 2)
        args = c.backend.url_decode(args.strip)
        assert_equal "file_info", cmd
        assert_equal key, args["key"]
        out["key"] = key
        a.write "OK #{c.backend.url_encode(out)}\r\n"
      end
      a.close
    end

    blk = lambda do |key, length, devcount|
      received << [ key, length, devcount ]
    end
    c.list_keys_verbose(%w(a b c d e), blk)
    th.join
    received.map! { |(key,_,_)| key }
    assert_equal %w(a b c d e), received
    ensure
      sock.close
  end

  def test_list_keys_verbose_retry_truncated # implementation detail
    received = []
    sock = TCPServer.new("127.0.0.1", 0)
    nargs = { :hosts => [ "127.0.0.1:#{sock.addr[1]}" ], :domain => "foo" }
    c = MogileFS::MogileFS.new(nargs)
    th = Thread.new do
      a = sock.accept
      %w(a b c d e).each do |key|
        line = a.gets
        cmd, args = line.split(/\s+/, 2)
        args = c.backend.url_decode(args.strip)
        assert_equal "file_info", cmd
        assert_equal key, args["key"]
      end
      out = { "length" => 3, "devcount" => 6 }
      out["key"] = "a"
      a.write "OK #{c.backend.url_encode(out)}\r\n"
      out["key"] = "b"
      a.write "OK #{c.backend.url_encode(out)}"
      a.close # trigger EOF

      a = sock.accept # client will retry
      %w(b c d e).each do |key|
        line = a.gets
        cmd, args = line.split(/\s+/, 2)
        args = c.backend.url_decode(args.strip)
        assert_equal "file_info", cmd
        assert_equal key, args["key"]
        out["key"] = key
        a.write "OK #{c.backend.url_encode(out)}\r\n"
      end
      a.close
    end

    blk = lambda do |key, length, devcount|
      received << [ key, length, devcount ]
    end
    c.list_keys_verbose(%w(a b c d e), blk)
    th.join
    received.map! { |(key,_,_)| key }
    assert_equal %w(a b c d e), received
    ensure
      sock.close
  end

  def test_sleep
    @backend.sleep = {}
    assert_nothing_raised do
      assert_equal({}, @client.sleep(2))
    end
  end

  private

    # tested with 1000, though it takes a while
    def nr_chunks
      ENV['NR_CHUNKS'] ? ENV['NR_CHUNKS'].to_i : 10
    end

end