test/test_mogilefs.rb

   1 # -*- encoding: binary -*-
   2 require './test/setup'
   3 require 'stringio'
   4 require 'tempfile'
   5 require 'fileutils'
   6
   7 class TestMogileFS__MogileFS < TestMogileFS
   8   def setup
   9     @klass = MogileFS::MogileFS
  10     super
  11   end
  12
  13   def test_initialize
  14     assert_equal 'test', @client.domain
  15
  16     assert_raises ArgumentError do
  17       MogileFS::MogileFS.new :hosts => ['kaa:6001']
  18     end
  19   end
  20
  21   def test_get_file_data_http
  22     tmp = Tempfile.new('accept')
  23     accept = File.open(tmp.path, "ab")
  24     svr = Proc.new do |serv, port|
  25       client, _ = serv.accept
  26       client.sync = true
  27       readed = client.recv(4096, 0)
  28       assert(readed =~ \
  29             %r{\AGET /dev[12]/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
  30       accept.syswrite('.')
  31       client.send("HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\ndata!", 0)
  32       client.close
  33     end
  34     t1 = TempServer.new(svr)
  35     t2 = TempServer.new(svr)
  36     path1 = "http://127.0.0.1:#{t1.port}/dev1/0/000/000/0000000062.fid"
  37     path2 = "http://127.0.0.1:#{t2.port}/dev2/0/000/000/0000000062.fid"
  38
  39     @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }
  40
  41     assert_equal 'data!', @client.get_file_data('key')
  42     assert_equal 1, accept.stat.size
  43     ensure
  44       TempServer.destroy_all!
  45   end
  46
  47   def test_get_file_data_http_not_found_failover
  48     tmp = Tempfile.new('accept')
  49     accept = File.open(tmp.path, 'ab')
  50     svr1 = Proc.new do |serv, port|
  51       client, _ = serv.accept
  52       client.sync = true
  53       readed = client.recv(4096, 0)
  54       assert(readed =~ \
  55             %r{\AGET /dev1/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
  56       accept.syswrite('.')
  57       client.send("HTTP/1.0 404 Not Found\r\n\r\ndata!", 0)
  58       client.close
  59     end
  60
  61     svr2 = Proc.new do |serv, port|
  62       client, _ = serv.accept
  63       client.sync = true
  64       readed = client.recv(4096, 0)
  65       assert(readed =~ \
  66             %r{\AGET /dev2/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
  67       accept.syswrite('.')
  68       client.send("HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\ndata!", 0)
  69       client.close
  70     end
  71
  72     t1 = TempServer.new(svr1)
  73     t2 = TempServer.new(svr2)
  74     path1 = "http://127.0.0.1:#{t1.port}/dev1/0/000/000/0000000062.fid"
  75     path2 = "http://127.0.0.1:#{t2.port}/dev2/0/000/000/0000000062.fid"
  76     @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }
  77
  78     assert_equal 'data!', @client.get_file_data('key')
  79     assert_equal 2, accept.stat.size
  80     ensure
  81       TempServer.destroy_all!
  82   end
  83
  84   def test_get_file_data_http_block
  85     tmpfp = Tempfile.new('test_mogilefs.open_data')
  86     nr = nr_chunks
  87     chunk_size = 1024 * 1024
  88     expect_size = nr * chunk_size
  89     header = "HTTP/1.0 200 OK\r\n" \
  90              "Content-Length: #{expect_size}\r\n\r\n"
  91     assert_equal header.size, tmpfp.syswrite(header)
  92     nr.times { assert_equal chunk_size, tmpfp.syswrite(' ' * chunk_size) }
  93     assert_equal expect_size + header.size, File.size(tmpfp.path)
  94     tmpfp.sysseek(0)
  95
  96     accept = Tempfile.new('accept')
  97     svr = Proc.new do |serv, port|
  98       client, _ = serv.accept
  99       client.sync = true
 100       accept.syswrite('.')
 101       readed = client.recv(4096, 0)
 102       assert(readed =~ \
 103             %r{\AGET /dev[12]/0/000/000/0000000062\.fid HTTP/1.[01]\r\n\r\n\Z})
 104       MogileFS::X.copy_stream(tmpfp, client)
 105       client.close
 106       exit 0
 107     end
 108     t1 = TempServer.new(svr)
 109     t2 = TempServer.new(svr)
 110     path1 = "http://127.0.0.1:#{t1.port}/dev1/0/000/000/0000000062.fid"
 111     path2 = "http://127.0.0.1:#{t2.port}/dev2/0/000/000/0000000062.fid"
 112
 113     @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }
 114
 115     data = Tempfile.new('test_mogilefs.dest_data')
 116     read_nr = nr = 0
 117     @client.get_file_data('key') do |fp|
 118       buf = ''
 119       loop do
 120         begin
 121           fp.sysread(16384, buf)
 122           read_nr = buf.size
 123           nr += read_nr
 124           assert_equal read_nr, data.syswrite(buf), "partial write"
 125         rescue Errno::EAGAIN
 126           retry
 127         rescue EOFError
 128           break
 129         end
 130       end
 131     end
 132     assert_equal expect_size, nr, "size mismatch"
 133     assert_equal 1, accept.stat.size
 134   end
 135
 136   def test_get_paths
 137     path1 = 'http://rur-1/dev1/0/000/000/0000000062.fid'
 138     path2 = 'http://rur-2/dev2/0/000/000/0000000062.fid'
 139
 140     @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }
 141
 142     expected = [ path1, path2 ]
 143
 144     assert_equal expected, @client.get_paths('key').sort
 145   end
 146
 147   def test_get_uris
 148     path1 = 'http://rur-1/dev1/0/000/000/0000000062.fid'
 149     path2 = 'http://rur-2/dev2/0/000/000/0000000062.fid'
 150
 151     @backend.get_paths = { 'paths' => 2, 'path1' => path1, 'path2' => path2 }
 152
 153     expected = [ URI.parse(path1), URI.parse(path2) ]
 154
 155     assert_equal expected, @client.get_uris('key')
 156   end
 157
 158
 159   def test_get_paths_unknown_key
 160     @backend.get_paths = ['unknown_key', '']
 161
 162     assert_raises MogileFS::Backend::UnknownKeyError do
 163       assert_equal nil, @client.get_paths('key')
 164     end
 165   end
 166
 167   def test_delete_existing
 168     @backend.delete = { }
 169     assert_nothing_raised do
 170       @client.delete 'no_such_key'
 171     end
 172   end
 173
 174   def test_delete_nonexisting
 175     @backend.delete = 'unknown_key', ''
 176     assert_raises MogileFS::Backend::UnknownKeyError do
 177       @client.delete('no_such_key')
 178     end
 179   end
 180
 181   def test_delete_readonly
 182     @client.readonly = true
 183     assert_raises MogileFS::ReadOnlyError do
 184       @client.delete 'no_such_key'
 185     end
 186   end
 187
 188   def test_each_key
 189     @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_2',
 190                            'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }
 191     @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_4',
 192                            'key_1' => 'new_key_3', 'key_2' => 'new_key_4' }
 193     @backend.list_keys = { 'key_count' => 0, 'next_after' => 'new_key_4' }
 194     keys = []
 195     @client.each_key 'new' do |key|
 196       keys << key
 197     end
 198
 199     assert_equal %w[new_key_1 new_key_2 new_key_3 new_key_4], keys
 200   end
 201
 202   def test_list_keys
 203     @backend.list_keys = { 'key_count' => '2', 'next_after' => 'new_key_2',
 204                            'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }
 205
 206     keys, next_after = @client.list_keys 'new'
 207     assert_equal ['new_key_1', 'new_key_2'], keys.sort
 208     assert_equal 'new_key_2', next_after
 209   end
 210
 211   def test_new_file_http
 212     @client.readonly = true
 213     assert_raises MogileFS::ReadOnlyError do
 214       @client.new_file 'new_key', 'test'
 215     end
 216   end
 217
 218   def test_new_file_readonly
 219     @client.readonly = true
 220     assert_raises MogileFS::ReadOnlyError do
 221       @client.new_file 'new_key', 'test'
 222     end
 223   end
 224
 225   def test_store_file_small_http
 226     received = Tempfile.new('received')
 227     to_store = Tempfile.new('small')
 228     to_store.syswrite('data')
 229
 230     expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"
 231     t = TempServer.new(Proc.new do |serv, accept|
 232       client, _ = serv.accept
 233       client.sync = true
 234       received.syswrite(client.read(expected.bytesize))
 235       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 236       client.close
 237     end)
 238
 239     @backend.create_open = {
 240       'devid' => '1',
 241       'path' => "http://127.0.0.1:#{t.port}/path",
 242     }
 243     nr = @client.store_file 'new_key', 'test', to_store.path
 244     assert_equal 4, nr
 245     received.sysseek(0)
 246     assert_equal expected, received.sysread(4096)
 247     ensure
 248       TempServer.destroy_all!
 249   end
 250
 251   def test_store_content_http
 252     received = Tempfile.new('recieved')
 253     expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"
 254
 255     t = TempServer.new(Proc.new do |serv, accept|
 256       client, _ = serv.accept
 257       client.sync = true
 258       received.syswrite(client.recv(4096, 0))
 259       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 260       client.close
 261     end)
 262
 263     @backend.create_open = {
 264       'devid' => '1',
 265       'path' => "http://127.0.0.1:#{t.port}/path",
 266     }
 267
 268     nr = @client.store_content 'new_key', 'test', 'data'
 269     assert nr
 270     assert_equal 4, nr
 271
 272     received.sysseek(0)
 273     assert_equal expected, received.sysread(4096)
 274     ensure
 275       TempServer.destroy_all!
 276   end
 277
 278
 279   def test_store_content_with_writer_callback
 280     received = Tempfile.new('recieved')
 281     expected = "PUT /path HTTP/1.0\r\nContent-Length: 40\r\n\r\n"
 282     10.times do
 283       expected += "data"
 284     end
 285     t = TempServer.new(Proc.new do |serv, accept|
 286       client, _ = serv.accept
 287       client.sync = true
 288       nr = 0
 289       loop do
 290         buf = client.readpartial(8192) or break
 291         break if buf.length == 0
 292         assert_equal buf.length, received.syswrite(buf)
 293         nr += buf.length
 294         break if nr >= expected.size
 295       end
 296       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 297       client.close
 298     end)
 299
 300     @backend.create_open = {
 301       'devid' => '1',
 302       'path' => "http://127.0.0.1:#{t.port}/path",
 303     }
 304
 305     cbk = MogileFS::Util::StoreContent.new(40) do |write_callback|
 306       10.times do
 307         write_callback.call("data")
 308       end
 309     end
 310     assert_equal 40, cbk.length
 311     nr = @client.store_content('new_key', 'test', cbk)
 312     assert_equal 40, nr
 313
 314     received.sysseek(0)
 315     assert_equal expected, received.sysread(4096)
 316     ensure
 317       TempServer.destroy_all!
 318   end
 319
 320   def test_store_content_multi_dest_failover
 321     received1 = Tempfile.new('received')
 322     received2 = Tempfile.new('received')
 323     expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"
 324
 325     t1 = TempServer.new(Proc.new do |serv, accept|
 326       client, _ = serv.accept
 327       client.sync = true
 328       received1.syswrite(client.recv(4096, 0))
 329       client.send("HTTP/1.0 500 Internal Server Error\r\n\r\n", 0)
 330       client.close
 331     end)
 332
 333     t2 = TempServer.new(Proc.new do |serv, accept|
 334       client, _ = serv.accept
 335       client.sync = true
 336       received2.syswrite(client.recv(4096, 0))
 337       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 338       client.close
 339     end)
 340
 341     @backend.create_open = {
 342       'dev_count' => '2',
 343       'devid_1' => '1',
 344       'path_1' => "http://127.0.0.1:#{t1.port}/path",
 345       'devid_2' => '2',
 346       'path_2' => "http://127.0.0.1:#{t2.port}/path",
 347     }
 348
 349     nr = @client.store_content 'new_key', 'test', 'data'
 350     assert_equal 4, nr
 351     received1.sysseek(0)
 352     received2.sysseek(0)
 353     assert_equal expected, received1.sysread(4096)
 354     assert_equal expected, received2.sysread(4096)
 355     ensure
 356       TempServer.destroy_all!
 357   end
 358
 359   def test_store_content_http_fail
 360     t = TempServer.new(Proc.new do |serv, accept|
 361       client, _ = serv.accept
 362       client.sync = true
 363       client.recv(4096, 0)
 364       client.send("HTTP/1.0 500 Internal Server Error\r\n\r\n", 0)
 365       client.close
 366     end)
 367
 368     @backend.create_open = {
 369       'devid' => '1',
 370       'path' => "http://127.0.0.1:#{t.port}/path",
 371     }
 372
 373     assert_raises MogileFS::HTTPFile::NoStorageNodesError do
 374       @client.store_content 'new_key', 'test', 'data'
 375     end
 376   end
 377
 378   def test_store_content_http_empty
 379     received = Tempfile.new('received')
 380     expected = "PUT /path HTTP/1.0\r\nContent-Length: 0\r\n\r\n"
 381     t = TempServer.new(Proc.new do |serv, accept|
 382       client, _ = serv.accept
 383       client.sync = true
 384       received.syswrite(client.recv(4096, 0))
 385       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 386       client.close
 387     end)
 388
 389     @backend.create_open = {
 390       'devid' => '1',
 391       'path' => "http://127.0.0.1:#{t.port}/path",
 392     }
 393
 394     nr = @client.store_content 'new_key', 'test', ''
 395     assert_equal 0, nr
 396     received.sysseek(0)
 397     assert_equal expected, received.sysread(4096)
 398   end
 399
 400   def test_store_content_nfs
 401     @backend.create_open = {
 402       'dev_count' => '1',
 403       'devid_1' => '1',
 404       'path_1' => '/path',
 405     }
 406     assert_raises MogileFS::UnsupportedPathError do
 407       @client.store_content 'new_key', 'test', 'data'
 408     end
 409   end
 410
 411   def test_new_file_http_large
 412     expect = Tempfile.new('test_mogilefs.expect')
 413     to_put = Tempfile.new('test_mogilefs.to_put')
 414     received = Tempfile.new('test_mogilefs.received')
 415
 416     nr = nr_chunks
 417     chunk_size = 1024 * 1024
 418     expect_size = nr * chunk_size
 419
 420     header = "PUT /path HTTP/1.0\r\n" \
 421              "Content-Length: #{expect_size}\r\n\r\n"
 422     assert_equal header.size, expect.syswrite(header)
 423     nr.times do
 424       assert_equal chunk_size, expect.syswrite(' ' * chunk_size)
 425       assert_equal chunk_size, to_put.syswrite(' ' * chunk_size)
 426     end
 427     assert_equal expect_size + header.size, expect.stat.size
 428     assert_equal expect_size, to_put.stat.size
 429
 430     readed = Tempfile.new('readed')
 431     t = TempServer.new(Proc.new do |serv, accept|
 432       client, _ = serv.accept
 433       client.sync = true
 434       nr = 0
 435       loop do
 436         buf = client.readpartial(8192) or break
 437         break if buf.length == 0
 438         assert_equal buf.length, received.syswrite(buf)
 439         nr += buf.length
 440         break if nr >= expect.stat.size
 441       end
 442       readed.syswrite("#{nr}")
 443       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 444       client.close
 445     end)
 446
 447     @backend.create_open = {
 448       'devid' => '1',
 449       'path' => "http://127.0.0.1:#{t.port}/path",
 450     }
 451
 452     orig_size = to_put.size
 453     nr = @client.store_file('new_key', 'test', to_put.path)
 454     assert nr, nr.inspect
 455     assert_equal orig_size, nr
 456     assert_equal orig_size, to_put.size
 457     readed.sysseek(0)
 458     assert_equal expect.stat.size, readed.sysread(4096).to_i
 459
 460     ENV['PATH'].split(/:/).each do |path|
 461       cmp_bin = "#{path}/cmp"
 462       File.executable?(cmp_bin) or next
 463       # puts "running #{cmp_bin} #{expect.path} #{received.path}"
 464       assert( system(cmp_bin, expect.path, received.path) )
 465       break
 466     end
 467
 468     ensure
 469       TempServer.destroy_all!
 470   end
 471
 472   def test_store_content_readonly
 473     @client.readonly = true
 474
 475     assert_raises MogileFS::ReadOnlyError do
 476       @client.store_content 'new_key', 'test', nil
 477     end
 478   end
 479
 480   def test_store_file_readonly
 481     @client.readonly = true
 482     assert_raises MogileFS::ReadOnlyError do
 483       @client.store_file 'new_key', 'test', nil
 484     end
 485   end
 486
 487   def test_rename_existing
 488     @backend.rename = {}
 489
 490     assert_nil @client.rename('from_key', 'to_key')
 491   end
 492
 493   def test_rename_nonexisting
 494     @backend.rename = 'unknown_key', ''
 495
 496     assert_raises MogileFS::Backend::UnknownKeyError do
 497       @client.rename('from_key', 'to_key')
 498     end
 499   end
 500
 501   def test_rename_no_key
 502     @backend.rename = 'no_key', 'no_key'
 503
 504     e = assert_raises MogileFS::Backend::NoKeyError do
 505       @client.rename 'new_key', 'test'
 506     end
 507
 508     assert_equal 'no_key', e.message
 509   end
 510
 511   def test_rename_readonly
 512     @client.readonly = true
 513
 514     e = assert_raises MogileFS::ReadOnlyError do
 515       @client.rename 'new_key', 'test'
 516     end
 517
 518     assert_equal 'readonly mogilefs', e.message
 519   end
 520
 521   def test_get_paths_args
 522     sock = TCPServer.new("127.0.0.1", 0)
 523     args = { :hosts => [ "127.0.0.1:#{sock.addr[1]}" ], :domain => "foo" }
 524     c = MogileFS::MogileFS.new(args)
 525     received = []
 526     th = Thread.new do
 527       a = sock.accept
 528       6.times do
 529         line = a.gets
 530         received << line
 531         a.write("OK paths=2&path1=http://0/a&path2=http://0/b\r\n")
 532       end
 533       a.close
 534     end
 535     expect = %w(http://0/a http://0/b)
 536     assert_equal expect, c.get_paths("f")
 537     assert_equal expect, c.get_paths("f",true,"alt")
 538     assert_equal expect, c.get_paths("f",:noverify=>true,:zone=>"alt")
 539     assert_equal expect,
 540                 c.get_paths("f",:noverify=>true,:zone=>"alt",:pathcount=>666)
 541     c.instance_variable_set(:@zone, "alt")
 542     assert_equal expect, c.get_paths("f")
 543     assert_equal expect, c.get_paths("f", :zone=>"zzz")
 544     th.join
 545
 546     backend = c.instance_variable_get(:@backend)
 547     expect = { "domain"=>"foo", "key"=>"f", "noverify"=>"0", "zone"=>"" }
 548     tmp = backend.url_decode(received[0].split(/\s+/)[1])
 549     assert_equal expect, tmp
 550     expect["noverify"] = "1"
 551     expect["zone"] = "alt"
 552     tmp = backend.url_decode(received[1].split(/\s+/)[1])
 553     assert_equal expect, tmp
 554
 555     tmp = backend.url_decode(received[2].split(/\s+/)[1])
 556     assert_equal expect, tmp
 557
 558     expect["pathcount"] = "666"
 559     tmp = backend.url_decode(received[3].split(/\s+/)[1])
 560     assert_equal expect, tmp
 561
 562     expect = { "domain"=>"foo", "key"=>"f", "noverify"=>"0", "zone"=>"alt" }
 563     tmp = backend.url_decode(received[4].split(/\s+/)[1])
 564
 565     expect = { "domain"=>"foo", "key"=>"f", "noverify"=>"0", "zone"=>"zzz" }
 566     tmp = backend.url_decode(received[5].split(/\s+/)[1])
 567     ensure
 568       sock.close
 569   end
 570
 571   def test_sleep
 572     @backend.sleep = {}
 573     assert_nothing_raised do
 574       assert_equal({}, @client.sleep(2))
 575     end
 576   end
 577
 578   private
 579
 580     # tested with 1000, though it takes a while
 581     def nr_chunks
 582       ENV['NR_CHUNKS'] ? ENV['NR_CHUNKS'].to_i : 10
 583     end
 584
 585 end
 586