services/api/test/integration/websocket_test.rb

   1 require 'test_helper'
   2 require 'websocket_runner'
   3 require 'oj'
   4 require 'database_cleaner'
   5
   6 DatabaseCleaner.strategy = :deletion
   7
   8 class WebsocketTest < ActionDispatch::IntegrationTest
   9   self.use_transactional_fixtures = false
  10
  11   setup do
  12     DatabaseCleaner.start
  13   end
  14
  15   teardown do
  16     DatabaseCleaner.clean
  17   end
  18
  19   def ws_helper (token = nil, timeout = true)
  20     opened = false
  21     close_status = nil
  22     too_long = false
  23
  24     EM.run {
  25       if token
  26         ws = Faye::WebSocket::Client.new("ws://localhost:3002/websocket?api_token=#{api_client_authorizations(token).api_token}")
  27       else
  28         ws = Faye::WebSocket::Client.new("ws://localhost:3002/websocket")
  29       end
  30
  31       ws.on :open do |event|
  32         opened = true
  33         if timeout
  34           EM::Timer.new 4 do
  35             too_long = true if close_status.nil?
  36             EM.stop_event_loop
  37           end
  38         end
  39       end
  40
  41       ws.on :close do |event|
  42         close_status = [:close, event.code, event.reason]
  43         EM.stop_event_loop
  44       end
  45
  46       yield ws
  47     }
  48
  49     assert opened, "Should have opened web socket"
  50     assert (not too_long), "Test took too long"
  51     assert_equal 1000, close_status[1], "Connection closed unexpectedly (check log for errors)"
  52   end
  53
  54   test "connect with no token" do
  55     status = nil
  56
  57     ws_helper do |ws|
  58       ws.on :message do |event|
  59         d = Oj.load event.data
  60         status = d["status"]
  61         ws.close
  62       end
  63     end
  64
  65     assert_equal 401, status
  66   end
  67
  68
  69   test "connect, subscribe and get response" do
  70     status = nil
  71
  72     ws_helper :admin do |ws|
  73       ws.on :open do |event|
  74         ws.send ({method: 'subscribe'}.to_json)
  75       end
  76
  77       ws.on :message do |event|
  78         d = Oj.load event.data
  79         status = d["status"]
  80         ws.close
  81       end
  82     end
  83
  84     assert_equal 200, status
  85   end
  86
  87   test "connect, subscribe, get event" do
  88     state = 1
  89     spec = nil
  90     ev_uuid = nil
  91
  92     authorize_with :admin
  93
  94     ws_helper :admin do |ws|
  95       ws.on :open do |event|
  96         ws.send ({method: 'subscribe'}.to_json)
  97       end
  98
  99       ws.on :message do |event|
 100         d = Oj.load event.data
 101         case state
 102         when 1
 103           assert_equal 200, d["status"]
 104           spec = Specimen.create
 105           state = 2
 106         when 2
 107           ev_uuid = d["object_uuid"]
 108           ws.close
 109         end
 110       end
 111
 112     end
 113
 114     assert_not_nil spec
 115     assert_equal spec.uuid, ev_uuid
 116   end
 117
 118   test "connect, subscribe, get two events" do
 119     state = 1
 120     spec = nil
 121     human = nil
 122     spec_ev_uuid = nil
 123     human_ev_uuid = nil
 124
 125     authorize_with :admin
 126
 127     ws_helper :admin do |ws|
 128       ws.on :open do |event|
 129         ws.send ({method: 'subscribe'}.to_json)
 130       end
 131
 132       ws.on :message do |event|
 133         d = Oj.load event.data
 134         case state
 135         when 1
 136           assert_equal 200, d["status"]
 137           spec = Specimen.create
 138           human = Human.create
 139           state = 2
 140         when 2
 141           spec_ev_uuid = d["object_uuid"]
 142           state = 3
 143         when 3
 144           human_ev_uuid = d["object_uuid"]
 145           state = 4
 146           ws.close
 147         when 4
 148           assert false, "Should not get any more events"
 149         end
 150       end
 151
 152     end
 153
 154     assert_not_nil spec
 155     assert_not_nil human
 156     assert_equal spec.uuid, spec_ev_uuid
 157     assert_equal human.uuid, human_ev_uuid
 158   end
 159
 160   test "connect, subscribe, filter events" do
 161     state = 1
 162     human = nil
 163     human_ev_uuid = nil
 164
 165     authorize_with :admin
 166
 167     ws_helper :admin do |ws|
 168       ws.on :open do |event|
 169         ws.send ({method: 'subscribe', filters: [['object_uuid', 'is_a', 'arvados#human']]}.to_json)
 170       end
 171
 172       ws.on :message do |event|
 173         d = Oj.load event.data
 174         case state
 175         when 1
 176           assert_equal 200, d["status"]
 177           Specimen.create
 178           human = Human.create
 179           state = 2
 180         when 2
 181           human_ev_uuid = d["object_uuid"]
 182           state = 3
 183           ws.close
 184         when 3
 185           assert false, "Should not get any more events"
 186         end
 187       end
 188
 189     end
 190
 191     assert_not_nil human
 192     assert_equal human.uuid, human_ev_uuid
 193   end
 194
 195
 196   test "connect, subscribe, multiple filters" do
 197     state = 1
 198     spec = nil
 199     human = nil
 200     spec_ev_uuid = nil
 201     human_ev_uuid = nil
 202
 203     authorize_with :admin
 204
 205     ws_helper :admin do |ws|
 206       ws.on :open do |event|
 207         ws.send ({method: 'subscribe', filters: [['object_uuid', 'is_a', 'arvados#human']]}.to_json)
 208         ws.send ({method: 'subscribe', filters: [['object_uuid', 'is_a', 'arvados#specimen']]}.to_json)
 209       end
 210
 211       ws.on :message do |event|
 212         d = Oj.load event.data
 213         case state
 214         when 1
 215           assert_equal 200, d["status"]
 216           state = 2
 217         when 2
 218           assert_equal 200, d["status"]
 219           spec = Specimen.create
 220           Trait.create # not part of filters, should not be received
 221           human = Human.create
 222           state = 3
 223         when 3
 224           spec_ev_uuid = d["object_uuid"]
 225           state = 4
 226         when 4
 227           human_ev_uuid = d["object_uuid"]
 228           state = 5
 229           ws.close
 230         when 5
 231           assert false, "Should not get any more events"
 232         end
 233       end
 234
 235     end
 236
 237     assert_not_nil spec
 238     assert_not_nil human
 239     assert_equal spec.uuid, spec_ev_uuid
 240     assert_equal human.uuid, human_ev_uuid
 241   end
 242
 243
 244   test "connect, subscribe, compound filter" do
 245     state = 1
 246     t1 = nil
 247
 248     authorize_with :admin
 249
 250     ws_helper :admin do |ws|
 251       ws.on :open do |event|
 252         ws.send ({method: 'subscribe', filters: [['object_uuid', 'is_a', 'arvados#trait'], ['event_type', '=', 'update']]}.to_json)
 253       end
 254
 255       ws.on :message do |event|
 256         d = Oj.load event.data
 257         case state
 258         when 1
 259           assert_equal 200, d["status"]
 260           t1 = Trait.create("name" => "foo")
 261           t1.name = "bar"
 262           t1.save!
 263           state = 2
 264          when 2
 265           assert_equal 'update', d['event_type']
 266           state = 3
 267           ws.close
 268         when 3
 269           assert false, "Should not get any more events"
 270         end
 271       end
 272
 273     end
 274
 275     assert_equal 3, state
 276     assert_not_nil t1
 277   end
 278
 279   test "connect, subscribe, ask events starting at seq num" do
 280     state = 1
 281     human = nil
 282     human_ev_uuid = nil
 283
 284     authorize_with :admin
 285
 286     lastid = logs(:log3).id
 287     l1 = nil
 288     l2 = nil
 289
 290     ws_helper :admin do |ws|
 291       ws.on :open do |event|
 292         ws.send ({method: 'subscribe', last_log_id: lastid}.to_json)
 293       end
 294
 295       ws.on :message do |event|
 296         d = Oj.load event.data
 297         case state
 298         when 1
 299           assert_equal 200, d["status"]
 300           state = 2
 301         when 2
 302           l1 = d["object_uuid"]
 303           assert_not_nil l1, "Unexpected message: #{d}"
 304           state = 3
 305         when 3
 306           l2 = d["object_uuid"]
 307           assert_not_nil l2, "Unexpected message: #{d}"
 308           state = 4
 309           ws.close
 310         when 4
 311           assert false, "Should not get any more events"
 312         end
 313       end
 314
 315     end
 316
 317     assert_equal logs(:log4).object_uuid, l1
 318     assert_equal logs(:log5).object_uuid, l2
 319   end
 320
 321   test "connect, subscribe, get event, unsubscribe" do
 322     state = 1
 323     spec = nil
 324     spec_ev_uuid = nil
 325     filter_id = nil
 326
 327     authorize_with :admin
 328
 329     ws_helper :admin, false do |ws|
 330       ws.on :open do |event|
 331         ws.send ({method: 'subscribe'}.to_json)
 332         EM::Timer.new 3 do
 333           # Set a time limit on the test because after unsubscribing the server
 334           # still has to process the next event (and then hopefully correctly
 335           # decides not to send it because we unsubscribed.)
 336           ws.close
 337         end
 338       end
 339
 340       ws.on :message do |event|
 341         d = Oj.load event.data
 342         case state
 343         when 1
 344           assert_equal 200, d["status"]
 345           spec = Specimen.create
 346           state = 2
 347         when 2
 348           spec_ev_uuid = d["object_uuid"]
 349           ws.send ({method: 'unsubscribe'}.to_json)
 350
 351           EM::Timer.new 1 do
 352             Specimen.create
 353           end
 354
 355           state = 3
 356         when 3
 357           assert_equal 200, d["status"]
 358           state = 4
 359         when 4
 360           assert false, "Should not get any more events"
 361         end
 362       end
 363
 364     end
 365
 366     assert_not_nil spec
 367     assert_equal spec.uuid, spec_ev_uuid
 368   end
 369
 370   test "connect, subscribe, get event, unsubscribe with filter" do
 371     state = 1
 372     spec = nil
 373     spec_ev_uuid = nil
 374
 375     authorize_with :admin
 376
 377     ws_helper :admin, false do |ws|
 378       ws.on :open do |event|
 379         ws.send ({method: 'subscribe', filters: [['object_uuid', 'is_a', 'arvados#human']]}.to_json)
 380         EM::Timer.new 3 do
 381           # Set a time limit on the test because after unsubscribing the server
 382           # still has to process the next event (and then hopefully correctly
 383           # decides not to send it because we unsubscribed.)
 384           ws.close
 385         end
 386       end
 387
 388       ws.on :message do |event|
 389         d = Oj.load event.data
 390         case state
 391         when 1
 392           assert_equal 200, d["status"]
 393           spec = Human.create
 394           state = 2
 395         when 2
 396           spec_ev_uuid = d["object_uuid"]
 397           ws.send ({method: 'unsubscribe', filters: [['object_uuid', 'is_a', 'arvados#human']]}.to_json)
 398
 399           EM::Timer.new 1 do
 400             Human.create
 401           end
 402
 403           state = 3
 404         when 3
 405           assert_equal 200, d["status"]
 406           state = 4
 407         when 4
 408           assert false, "Should not get any more events"
 409         end
 410       end
 411
 412     end
 413
 414     assert_not_nil spec
 415     assert_equal spec.uuid, spec_ev_uuid
 416   end
 417
 418
 419   test "connect, subscribe, get event, try to unsubscribe with bogus filter" do
 420     state = 1
 421     spec = nil
 422     spec_ev_uuid = nil
 423     human = nil
 424     human_ev_uuid = nil
 425
 426     authorize_with :admin
 427
 428     ws_helper :admin do |ws|
 429       ws.on :open do |event|
 430         ws.send ({method: 'subscribe'}.to_json)
 431       end
 432
 433       ws.on :message do |event|
 434         d = Oj.load event.data
 435         case state
 436         when 1
 437           assert_equal 200, d["status"]
 438           spec = Specimen.create
 439           state = 2
 440         when 2
 441           spec_ev_uuid = d["object_uuid"]
 442           ws.send ({method: 'unsubscribe', filters: [['foo', 'bar', 'baz']]}.to_json)
 443
 444           EM::Timer.new 1 do
 445             human = Human.create
 446           end
 447
 448           state = 3
 449         when 3
 450           assert_equal 404, d["status"]
 451           state = 4
 452         when 4
 453           human_ev_uuid = d["object_uuid"]
 454           state = 5
 455           ws.close
 456         when 5
 457           assert false, "Should not get any more events"
 458         end
 459       end
 460
 461     end
 462
 463     assert_not_nil spec
 464     assert_not_nil human
 465     assert_equal spec.uuid, spec_ev_uuid
 466     assert_equal human.uuid, human_ev_uuid
 467   end
 468
 469
 470
 471   test "connected, not subscribed, no event" do
 472     authorize_with :admin
 473
 474     ws_helper :admin, false do |ws|
 475       ws.on :open do |event|
 476         EM::Timer.new 1 do
 477           Specimen.create
 478         end
 479
 480         EM::Timer.new 3 do
 481           ws.close
 482         end
 483       end
 484
 485       ws.on :message do |event|
 486         assert false, "Should not get any messages, message was #{event.data}"
 487       end
 488     end
 489   end
 490
 491   test "connected, not authorized to see event" do
 492     state = 1
 493
 494     authorize_with :admin
 495
 496     ws_helper :active, false do |ws|
 497       ws.on :open do |event|
 498         ws.send ({method: 'subscribe'}.to_json)
 499
 500         EM::Timer.new 3 do
 501           ws.close
 502         end
 503       end
 504
 505       ws.on :message do |event|
 506         d = Oj.load event.data
 507         case state
 508         when 1
 509           assert_equal 200, d["status"]
 510           Specimen.create
 511           state = 2
 512         when 2
 513           assert false, "Should not get any messages, message was #{event.data}"
 514         end
 515       end
 516
 517     end
 518
 519   end
 520
 521   test "connect, try bogus method" do
 522     status = nil
 523
 524     ws_helper :admin do |ws|
 525       ws.on :open do |event|
 526         ws.send ({method: 'frobnabble'}.to_json)
 527       end
 528
 529       ws.on :message do |event|
 530         d = Oj.load event.data
 531         status = d["status"]
 532         ws.close
 533       end
 534     end
 535
 536     assert_equal 400, status
 537   end
 538
 539   test "connect, missing method" do
 540     status = nil
 541
 542     ws_helper :admin do |ws|
 543       ws.on :open do |event|
 544         ws.send ({fizzbuzz: 'frobnabble'}.to_json)
 545       end
 546
 547       ws.on :message do |event|
 548         d = Oj.load event.data
 549         status = d["status"]
 550         ws.close
 551       end
 552     end
 553
 554     assert_equal 400, status
 555   end
 556
 557   test "connect, send malformed request" do
 558     status = nil
 559
 560     ws_helper :admin do |ws|
 561       ws.on :open do |event|
 562         ws.send '<XML4EVER></XML4EVER>'
 563       end
 564
 565       ws.on :message do |event|
 566         d = Oj.load event.data
 567         status = d["status"]
 568         ws.close
 569       end
 570     end
 571
 572     assert_equal 400, status
 573   end
 574
 575
 576   test "connect, try subscribe too many filters" do
 577     state = 1
 578
 579     authorize_with :admin
 580
 581     ws_helper :admin do |ws|
 582       ws.on :open do |event|
 583         (1..17).each do |i|
 584           ws.send ({method: 'subscribe', filters: [['object_uuid', '=', i]]}.to_json)
 585         end
 586       end
 587
 588       ws.on :message do |event|
 589         d = Oj.load event.data
 590         case state
 591         when (1..EventBus::MAX_FILTERS)
 592           assert_equal 200, d["status"]
 593           state += 1
 594         when (EventBus::MAX_FILTERS+1)
 595           assert_equal 403, d["status"]
 596           ws.close
 597         end
 598       end
 599
 600     end
 601
 602     assert_equal 17, state
 603
 604   end
 605
 606   test "connect, subscribe, lots of events" do
 607     state = 1
 608     event_count = 0
 609     log_start = Log.order(:id).last.id
 610
 611     authorize_with :admin
 612
 613     ws_helper :admin, false do |ws|
 614       EM::Timer.new 45 do
 615         # Needs a longer timeout than the default
 616         ws.close
 617       end
 618
 619       ws.on :open do |event|
 620         ws.send ({method: 'subscribe'}.to_json)
 621       end
 622
 623       ws.on :message do |event|
 624         d = Oj.load event.data
 625         case state
 626         when 1
 627           assert_equal 200, d["status"]
 628           ActiveRecord::Base.transaction do
 629             (1..202).each do
 630               spec = Specimen.create
 631             end
 632           end
 633           state = 2
 634         when 2
 635           event_count += 1
 636           assert_equal d['id'], event_count+log_start
 637           if event_count == 202
 638             ws.close
 639           end
 640         end
 641       end
 642
 643     end
 644
 645     assert_equal 202, event_count
 646   end
 647
 648
 649 end