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