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