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       client.send("HTTP/1.0 404 Not Found\r\n\r\ndata!", 0)
  58       accept.syswrite('.')
  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       client.send("HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\ndata!", 0)
  69       accept.syswrite('.')
  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_paths_unknown_key
 149     @backend.get_paths = ['unknown_key', '']
 150
 151     assert_raises MogileFS::Backend::UnknownKeyError do
 152       assert_equal nil, @client.get_paths('key')
 153     end
 154   end
 155
 156   def test_delete_existing
 157     @backend.delete = { }
 158     assert_nothing_raised do
 159       @client.delete 'no_such_key'
 160     end
 161   end
 162
 163   def test_delete_nonexisting
 164     @backend.delete = 'unknown_key', ''
 165     assert_raises MogileFS::Backend::UnknownKeyError do
 166       @client.delete('no_such_key')
 167     end
 168   end
 169
 170   def test_delete_readonly
 171     @client.readonly = true
 172     assert_raises MogileFS::ReadOnlyError do
 173       @client.delete 'no_such_key'
 174     end
 175   end
 176
 177   def test_each_key
 178     @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_2',
 179                            'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }
 180     @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_4',
 181                            'key_1' => 'new_key_3', 'key_2' => 'new_key_4' }
 182     @backend.list_keys = { 'key_count' => 0, 'next_after' => 'new_key_4' }
 183     keys = []
 184     @client.each_key 'new' do |key|
 185       keys << key
 186     end
 187
 188     assert_equal %w[new_key_1 new_key_2 new_key_3 new_key_4], keys
 189   end
 190
 191   def test_list_keys
 192     @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_2',
 193                            'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }
 194
 195     keys, next_after = @client.list_keys 'new'
 196     assert_equal ['new_key_1', 'new_key_2'], keys.sort
 197     assert_equal 'new_key_2', next_after
 198   end
 199
 200   def test_list_keys_block
 201     @backend.list_keys = { 'key_count' => 2, 'next_after' => 'new_key_2',
 202                            'key_1' => 'new_key_1', 'key_2' => 'new_key_2' }
 203     http_resp = "HTTP/1.0 200 OK\r\nContent-Length: %u\r\n"
 204     srv = Proc.new do |serv, port, size|
 205       client, client_addr = serv.accept
 206       client.sync = true
 207       readed = client.readpartial(4096)
 208       assert %r{\AHEAD } =~ readed
 209       client.send(http_resp % size, 0)
 210       client.close
 211     end
 212     t1 = TempServer.new(Proc.new { |serv, port| srv.call(serv, port, 5) })
 213     t2 = TempServer.new(Proc.new { |serv, port| srv.call(serv, port, 5) })
 214     t3 = TempServer.new(Proc.new { |serv, port| srv.call(serv, port, 10) })
 215     @backend.get_paths = { 'paths' => 2,
 216                            'path1' => "http://127.0.0.1:#{t1.port}/",
 217                            'path2' => "http://127.0.0.1:#{t2.port}/" }
 218     @backend.get_paths = { 'paths' => 1,
 219                            'path1' => "http://127.0.0.1:#{t3.port}/" }
 220
 221     res = []
 222     keys, next_after = @client.list_keys('new') do |key,length,devcount|
 223       res << [ key, length, devcount ]
 224     end
 225
 226     expect_res = [ [ 'new_key_1', 5, 2 ], [ 'new_key_2', 10, 1 ] ]
 227     assert_equal expect_res, res
 228     assert_equal ['new_key_1', 'new_key_2'], keys.sort
 229     assert_equal 'new_key_2', next_after
 230     ensure
 231       TempServer.destroy_all!
 232   end
 233
 234   def test_new_file_http
 235     @client.readonly = true
 236     assert_raises MogileFS::ReadOnlyError do
 237       @client.new_file 'new_key', 'test'
 238     end
 239   end
 240
 241   def test_new_file_readonly
 242     @client.readonly = true
 243     assert_raises MogileFS::ReadOnlyError do
 244       @client.new_file 'new_key', 'test'
 245     end
 246   end
 247
 248   def test_size_http
 249     accept = Tempfile.new('accept')
 250     t = TempServer.new(Proc.new do |serv,port|
 251       client, client_addr = serv.accept
 252       client.sync = true
 253       readed = client.recv(4096, 0) rescue nil
 254       accept.syswrite('.')
 255       assert_equal "HEAD /path HTTP/1.0\r\n\r\n", readed
 256       client.send("HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\n", 0)
 257       client.close
 258     end)
 259
 260     path = "http://127.0.0.1:#{t.port}/path"
 261     @backend.get_paths = { 'paths' => 1, 'path1' => path }
 262
 263     assert_equal 5, @client.size('key')
 264     assert_equal 1, accept.stat.size
 265   end
 266
 267   def test_bad_size_http
 268     tmp = Tempfile.new('accept')
 269     t = TempServer.new(Proc.new do |serv,port|
 270       client, client_addr = serv.accept
 271       client.sync = true
 272       readed = client.recv(4096, 0) rescue nil
 273       assert_equal "HEAD /path HTTP/1.0\r\n\r\n", readed
 274       client.send("HTTP/1.0 404 Not Found\r\nContent-Length: 5\r\n\r\n", 0)
 275       tmp.syswrite('.')
 276       client.close
 277     end)
 278
 279     path = "http://127.0.0.1:#{t.port}/path"
 280     @backend.get_paths = { 'paths' => 1, 'path1' => path }
 281
 282     assert_nil @client.size('key')
 283     assert_equal 1, tmp.stat.size
 284   end
 285
 286   def test_store_content_http
 287     received = Tempfile.new('recieved')
 288     expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"
 289
 290     t = TempServer.new(Proc.new do |serv, accept|
 291       client, client_addr = serv.accept
 292       client.sync = true
 293       received.syswrite(client.recv(4096, 0))
 294       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 295       client.close
 296     end)
 297
 298     @backend.create_open = {
 299       'devid' => '1',
 300       'path' => "http://127.0.0.1:#{t.port}/path",
 301     }
 302
 303     @client.store_content 'new_key', 'test', 'data'
 304
 305     received.sysseek(0)
 306     assert_equal expected, received.sysread(4096)
 307     ensure
 308       TempServer.destroy_all!
 309   end
 310
 311
 312   def test_store_content_with_writer_callback
 313     received = Tempfile.new('recieved')
 314     expected = "PUT /path HTTP/1.0\r\nContent-Length: 40\r\n\r\n"
 315     10.times do
 316       expected += "data"
 317     end
 318     t = TempServer.new(Proc.new do |serv, accept|
 319       client, client_addr = serv.accept
 320       client.sync = true
 321       nr = 0
 322       loop do
 323         buf = client.readpartial(8192) or break
 324         break if buf.length == 0
 325         assert_equal buf.length, received.syswrite(buf)
 326         nr += buf.length
 327         break if nr >= expected.size
 328       end
 329       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 330       client.close
 331     end)
 332
 333     @backend.create_open = {
 334       'devid' => '1',
 335       'path' => "http://127.0.0.1:#{t.port}/path",
 336     }
 337
 338     cbk = MogileFS::Util::StoreContent.new(40) do |write_callback|
 339       10.times do
 340         write_callback.call("data")
 341       end
 342     end
 343     @client.store_content('new_key', 'test', cbk)
 344
 345     received.sysseek(0)
 346     assert_equal expected, received.sysread(4096)
 347     ensure
 348       TempServer.destroy_all!
 349   end
 350
 351   def test_store_content_multi_dest_failover
 352     received1 = Tempfile.new('received')
 353     received2 = Tempfile.new('received')
 354     expected = "PUT /path HTTP/1.0\r\nContent-Length: 4\r\n\r\ndata"
 355
 356     t1 = TempServer.new(Proc.new do |serv, accept|
 357       client, client_addr = serv.accept
 358       client.sync = true
 359       received1.syswrite(client.recv(4096, 0))
 360       client.send("HTTP/1.0 500 Internal Server Error\r\n\r\n", 0)
 361       client.close
 362     end)
 363
 364     t2 = TempServer.new(Proc.new do |serv, accept|
 365       client, client_addr = serv.accept
 366       client.sync = true
 367       received2.syswrite(client.recv(4096, 0))
 368       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 369       client.close
 370     end)
 371
 372     @backend.create_open = {
 373       'dev_count' => '2',
 374       'devid_1' => '1',
 375       'path_1' => "http://127.0.0.1:#{t1.port}/path",
 376       'devid_2' => '2',
 377       'path_2' => "http://127.0.0.1:#{t2.port}/path",
 378     }
 379
 380     @client.store_content 'new_key', 'test', 'data'
 381     received1.sysseek(0)
 382     received2.sysseek(0)
 383     assert_equal expected, received1.sysread(4096)
 384     assert_equal expected, received2.sysread(4096)
 385     ensure
 386       TempServer.destroy_all!
 387   end
 388
 389   def test_store_content_http_fail
 390     t = TempServer.new(Proc.new do |serv, accept|
 391       client, client_addr = serv.accept
 392       client.sync = true
 393       client.recv(4096, 0)
 394       client.send("HTTP/1.0 500 Internal Server Error\r\n\r\n", 0)
 395       client.close
 396     end)
 397
 398     @backend.create_open = {
 399       'devid' => '1',
 400       'path' => "http://127.0.0.1:#{t.port}/path",
 401     }
 402
 403     assert_raises MogileFS::HTTPFile::BadResponseError do
 404       @client.store_content 'new_key', 'test', 'data'
 405     end
 406   end
 407
 408   def test_store_content_http_empty
 409     received = Tempfile.new('received')
 410     expected = "PUT /path HTTP/1.0\r\nContent-Length: 0\r\n\r\n"
 411     t = TempServer.new(Proc.new do |serv, accept|
 412       client, client_addr = serv.accept
 413       client.sync = true
 414       received.syswrite(client.recv(4096, 0))
 415       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 416       client.close
 417     end)
 418
 419     @backend.create_open = {
 420       'devid' => '1',
 421       'path' => "http://127.0.0.1:#{t.port}/path",
 422     }
 423
 424     @client.store_content 'new_key', 'test', ''
 425     received.sysseek(0)
 426     assert_equal expected, received.sysread(4096)
 427   end
 428
 429   def test_store_content_nfs
 430     @backend.create_open = {
 431       'dev_count' => '1',
 432       'devid_1' => '1',
 433       'path_1' => '/path',
 434     }
 435     assert_raises MogileFS::UnsupportedPathError do
 436       @client.store_content 'new_key', 'test', 'data'
 437     end
 438   end
 439
 440   def test_new_file_http_large
 441     expect = Tempfile.new('test_mogilefs.expect')
 442     to_put = Tempfile.new('test_mogilefs.to_put')
 443     received = Tempfile.new('test_mogilefs.received')
 444
 445     nr = nr_chunks
 446     chunk_size = 1024 * 1024
 447     expect_size = nr * chunk_size
 448
 449     header = "PUT /path HTTP/1.0\r\n" \
 450              "Content-Length: #{expect_size}\r\n\r\n"
 451     assert_equal header.size, expect.syswrite(header)
 452     nr.times do
 453       assert_equal chunk_size, expect.syswrite(' ' * chunk_size)
 454       assert_equal chunk_size, to_put.syswrite(' ' * chunk_size)
 455     end
 456     assert_equal expect_size + header.size, expect.stat.size
 457     assert_equal expect_size, to_put.stat.size
 458
 459     readed = Tempfile.new('readed')
 460     t = TempServer.new(Proc.new do |serv, accept|
 461       client, client_addr = serv.accept
 462       client.sync = true
 463       nr = 0
 464       loop do
 465         buf = client.readpartial(8192) or break
 466         break if buf.length == 0
 467         assert_equal buf.length, received.syswrite(buf)
 468         nr += buf.length
 469         break if nr >= expect.stat.size
 470       end
 471       readed.syswrite("#{nr}")
 472       client.send("HTTP/1.0 200 OK\r\n\r\n", 0)
 473       client.close
 474     end)
 475
 476     @backend.create_open = {
 477       'devid' => '1',
 478       'path' => "http://127.0.0.1:#{t.port}/path",
 479     }
 480
 481     @client.store_file('new_key', 'test', to_put.path)
 482     readed.sysseek(0)
 483     assert_equal expect.stat.size, readed.sysread(4096).to_i
 484
 485     ENV['PATH'].split(/:/).each do |path|
 486       cmp_bin = "#{path}/cmp"
 487       File.executable?(cmp_bin) or next
 488       # puts "running #{cmp_bin} #{expect.path} #{received.path}"
 489       assert( system(cmp_bin, expect.path, received.path) )
 490       break
 491     end
 492
 493     ensure
 494       TempServer.destroy_all!
 495   end
 496
 497   def test_store_content_readonly
 498     @client.readonly = true
 499
 500     assert_raises MogileFS::ReadOnlyError do
 501       @client.store_content 'new_key', 'test', nil
 502     end
 503   end
 504
 505   def test_store_file_readonly
 506     @client.readonly = true
 507     assert_raises MogileFS::ReadOnlyError do
 508       @client.store_file 'new_key', 'test', nil
 509     end
 510   end
 511
 512   def test_rename_existing
 513     @backend.rename = {}
 514
 515     assert_nil @client.rename('from_key', 'to_key')
 516   end
 517
 518   def test_rename_nonexisting
 519     @backend.rename = 'unknown_key', ''
 520
 521     assert_raises MogileFS::Backend::UnknownKeyError do
 522       @client.rename('from_key', 'to_key')
 523     end
 524   end
 525
 526   def test_rename_no_key
 527     @backend.rename = 'no_key', 'no_key'
 528
 529     e = assert_raises MogileFS::Backend::NoKeyError do
 530       @client.rename 'new_key', 'test'
 531     end
 532
 533     assert_equal 'no_key', e.message
 534   end
 535
 536   def test_rename_readonly
 537     @client.readonly = true
 538
 539     e = assert_raises MogileFS::ReadOnlyError do
 540       @client.rename 'new_key', 'test'
 541     end
 542
 543     assert_equal 'readonly mogilefs', e.message
 544   end
 545
 546   def test_sleep
 547     @backend.sleep = {}
 548     assert_nothing_raised do
 549       assert_equal({}, @client.sleep(2))
 550     end
 551   end
 552
 553   private
 554
 555     # tested with 1000, though it takes a while
 556     def nr_chunks
 557       ENV['NR_CHUNKS'] ? ENV['NR_CHUNKS'].to_i : 10
 558     end
 559
 560 end
 561