test/test_fresh.rb

   1 # -*- encoding: binary -*-
   2 require "./test/exec"
   3 require "tmpdir"
   4 require "fileutils"
   5 require "net/http"
   6
   7 class TestMogFresh < Test::Unit::TestCase
   8   include TestExec
   9
  10   def setup
  11     setup_mogilefs
  12   end
  13
  14   def setup_mogstored
  15     @docroot = Dir.mktmpdir(["mogfresh", "docroot"])
  16     @mogstored_mgmt = TCPServer.new(@test_host, 0)
  17     @mogstored_http = TCPServer.new(@test_host, 0)
  18     @mogstored_mgmt_port = @mogstored_mgmt.addr[1]
  19     @mogstored_http_port = @mogstored_http.addr[1]
  20     @mogstored_conf = Tempfile.new(["mogstored", "conf"])
  21     @mogstored_pid = Tempfile.new(["mogstored", "pid"])
  22     @mogstored_conf.write <<EOF
  23 pidfile = #{@mogstored_pid.path}
  24 maxconns = 1000
  25 httplisten = #@test_host:#{@mogstored_http_port}
  26 mgmtlisten = #@test_host:#{@mogstored_mgmt_port}
  27 docroot = #@docroot
  28 EOF
  29     @mogstored_conf.flush
  30     @mogstored_mgmt.close
  31     @mogstored_http.close
  32
  33     x!("mogstored", "--daemon", "--config=#{@mogstored_conf.path}")
  34     wait_for_port @mogstored_mgmt_port
  35     wait_for_port @mogstored_http_port
  36   end
  37
  38   def setup_mogilefs(plugins = nil)
  39     @test_host = "127.0.0.1"
  40     setup_mogstored
  41     @tracker = TCPServer.new(@test_host, 0)
  42     @tracker_port = @tracker.addr[1]
  43
  44     @dbname = Tempfile.new(["mogfresh", ".sqlite3"])
  45     @mogilefsd_conf = Tempfile.new(["mogilefsd", "conf"])
  46     @mogilefsd_pid = Tempfile.new(["mogilefsd", "pid"])
  47
  48     cmd = %w(mogdbsetup --yes --type=SQLite --dbname) << @dbname.path
  49     x!(*cmd)
  50
  51     @mogilefsd_conf.puts "db_dsn DBI:SQLite:#{@dbname.path}"
  52     @mogilefsd_conf.write <<EOF
  53 conf_port #@tracker_port
  54 listen #@test_host
  55 pidfile #{@mogilefsd_pid.path}
  56 replicate_jobs 1
  57 fsck_jobs 1
  58 query_jobs 1
  59 mogstored_stream_port #{@mogstored_mgmt_port}
  60 node_timeout 10
  61 EOF
  62     @mogilefsd_conf.flush
  63
  64     @trackers = @hosts = [ "#@test_host:#@tracker_port" ]
  65     @tracker.close
  66     x!("mogilefsd", "--daemon", "--config=#{@mogilefsd_conf.path}")
  67     wait_for_port @tracker_port
  68     @admin = MogileFS::Admin.new(:hosts => @hosts)
  69     10.times do
  70       break if @mogstored_pid.size > 0
  71       sleep 0.1
  72     end
  73   end
  74
  75   def wait_for_port(port)
  76     tries = 50
  77     begin
  78       TCPSocket.new(@test_host, port).close
  79       return
  80     rescue
  81       sleep 0.1
  82     end while (tries -= 1) > 0
  83     raise "#@test_host:#{port} never became ready"
  84   end
  85
  86   def test_admin_setup_new_host_and_devices
  87     assert_equal [], @admin.get_hosts
  88     args = { :ip => @test_host, :port => @mogstored_http_port }
  89     @admin.create_host("me", args)
  90     yield_for_monitor_update { @admin.get_hosts.empty? or break }
  91     hosts = @admin.get_hosts
  92     assert_equal 1, hosts.size
  93     host = @admin.get_hosts[0]
  94     assert_equal "me", host["hostname"]
  95     assert_equal @mogstored_http_port, host["http_port"]
  96     assert_nil host["http_get_port"]
  97     assert_equal @test_host, host["hostip"]
  98     assert_kind_of Integer, host["hostid"]
  99     assert_equal hosts, @admin.get_hosts(host["hostid"])
 100
 101     assert_equal [], @admin.get_devices
 102   end
 103
 104   def test_replicate_now
 105     assert_equal({"count" => 0}, @admin.replicate_now)
 106   end
 107
 108   def test_clear_cache
 109     assert_nil @admin.clear_cache
 110   end
 111
 112   def test_create_update_delete_class
 113     domain = "rbmogtest#{Time.now.strftime('%Y%m%d%H%M%S')}.#{uuid}"
 114     @admin.create_domain(domain)
 115     yield_for_monitor_update { @admin.get_domains.include?(domain) and break }
 116
 117     assert_nothing_raised do
 118       @admin.create_class(domain, "klassy", 1)
 119     end
 120     assert_raises(MogileFS::Backend::ClassExistsError) do
 121       @admin.create_class(domain, "klassy", 1)
 122     end
 123
 124     assert_nothing_raised do
 125       @admin.update_class(domain, "klassy",
 126                           :mindevcount => 1, :replpolicy => "MultipleHosts(1)")
 127     end
 128
 129     tmp = nil
 130     yield_for_monitor_update do
 131       tmp = @admin.get_domains[domain]["klassy"]
 132       break if tmp && tmp["replpolicy"] == "MultipleHosts(1)"
 133     end
 134     assert tmp, "domain did not show up"
 135     assert_equal 1, tmp["mindevcount"]
 136     assert_equal "MultipleHosts(1)", tmp["replpolicy"]
 137     assert_nothing_raised { @admin.update_class(domain, "klassy", 2) }
 138     ensure
 139       @admin.delete_class(domain, "klassy") rescue nil
 140   end
 141
 142   def test_device_file_add
 143     assert_equal [], @admin.get_hosts
 144     args = { :ip => @test_host, :port => @mogstored_http_port }
 145     args[:status] = "alive"
 146     @admin.create_host("me", args)
 147     Dir.mkdir("#@docroot/dev1")
 148     Dir.mkdir("#@docroot/dev2")
 149
 150     yield_for_monitor_update { @admin.get_hosts.empty? or break }
 151
 152     # TODO: allow adding devices via our MogileFS::Admin class
 153     mogadm!("device", "add", "me", "dev1")
 154     yield_for_monitor_update { @admin.get_devices.empty? or break }
 155     wait_for_usage_file "dev1"
 156     mogadm!("device", "add", "me", "dev2")
 157     wait_for_usage_file "dev2"
 158     out = err = nil
 159     tries = 0
 160     begin
 161       out.close! if out
 162       err.close! if err
 163       status, out, err = mogadm("check")
 164       if (tries += 1) > 100
 165         warn err.read
 166         puts out.read
 167         return
 168       end
 169       sleep 0.1
 170     end until out.read =~ /write?able/
 171
 172     domain = "rbmogtest.#$$"
 173     @admin.create_domain(domain)
 174     yield_for_monitor_update { @admin.get_domains.include?(domain) and break }
 175     client = MogileFS::MogileFS.new :hosts => @hosts, :domain => domain
 176     r, w = IO.pipe
 177     thr = Thread.new do
 178       (0..9).each do |i|
 179         sleep 0.05
 180         w.write("#{i}\n")
 181       end
 182       w.close
 183       :ok
 184     end
 185     assert_equal 20, client.store_file("pipe", nil, r)
 186     assert_equal :ok, thr.value
 187     r.close
 188     assert_equal "0\n1\n2\n3\n4\n5\n6\n7\n8\n9\n", client.get_file_data("pipe")
 189   end
 190
 191   def teardown
 192     if @mogstored_pid && @mogstored_pid.size > 0
 193       Process.kill(:TERM, @mogstored_pid.read.to_i)
 194     end
 195     if @mogilefsd_pid
 196       s = TCPSocket.new(@test_host, @tracker_port)
 197       s.write "!shutdown\r\n"
 198       s.close
 199     end
 200     FileUtils.rmtree(@docroot)
 201   end
 202
 203   def wait_for_usage_file(device)
 204     uri = URI("http://#@test_host:#@mogstored_http_port/#{device}/usage")
 205     res = nil
 206     100.times do
 207       res = Net::HTTP.get_response(uri)
 208       if Net::HTTPOK === res
 209         puts res.body if $DEBUG
 210         return
 211       end
 212       puts res.inspect if $DEBUG
 213       sleep 0.1
 214     end
 215     raise "#{uri} failed to appear: #{res.inspect}"
 216   end
 217 end