test/test_mogilefs.rb

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