Тестирование объекта с помощью TestScheduler в RxJs

Question

Я использую RxJs для подсчета количества пакетов в определенное временное окно. Мой код в основном выглядит так:

var packetSubject = new Rx.Subject(); 
var packetsInWindow = []; 

function startMonitoring() { 
    var subscription = packetSubject 
     .windowWithTime(1000) 
     .select(function(window) { 
      window.toArray().subscribe(function(elements) { 
       packetsInWindow.push(elements.length); 
      }); 
     }) 
     .subscribe(); 
} 

function newPacket(packet) { 
    packetSubject.onNext(packet); 
} 

Как проверить этот код с помощью Rx TestScheduler? Я не нашел подходящего примера для тестирования объектов.

Answer 1

Взгляните этот пример:

var x = 0, 
     scheduler = new Rx.TestScheduler(); 

    var subject = new Rx.Subject(); 
    subject.throttle(100, scheduler).subscribe(function (value) { 
     x = value; 
    }); 

    scheduler.scheduleWithAbsolute(0, function() { 
     subject.onNext(1);//trigger first event with value 1 
    }); 
    scheduler.scheduleWithAbsolute(50, function() { 
     expect(x).toEqual(0);//value hasn't been updated 
    }); 
    scheduler.scheduleWithAbsolute(200, function() { 
     expect(x).toEqual(1);//value update after throttle's windowDuration 
    }); 

    scheduler.start(); 

https://emmkong.wordpress.com/2015/03/18/how-to-unit-test-rxjs-throttle-with-rx-testscheduler/

Answer 2

С RxJs 4.x вы можете позвонить advanceBy на TestScheduler. TestScheduler можно вводить во время настройки. Я написал об этом blog post.

Вот пример.

var throttleWindowDuration = 2 * 1000; /* 2 seconds */ 

    function throttleTest() { 
     var unthrottledStream = new Rx.Subject(); 
     var source = unthrottledStream.throttle(throttleWindowDuration); 
     var result = { 
     emitCounter: 0, 
     unthrottledStream 
     }; 

     var subscription = source.subscribe(
     function() { 
      result.emitCounter++; 
     }); 

     return result; 
    } 

    describe('run with test scheduler', function() { 
     var testScheduler; 
     var throttleSpy; 

     beforeAll(function() { 
     testScheduler = new Rx.TestScheduler(); 
     var originalThrottle = Rx.Observable.prototype.throttle; 
     throttleSpy = spyOn(Rx.Observable.prototype, 'throttle') 
      .and.callFake(function(dueTime) { 
      return originalThrottle.call(this, dueTime, testScheduler); 
      }); 
     }); 

     afterAll(function() { 
     throttleSpy.and.callThrough(); 
     }); 

     it('advancing testScheduler allows to test throttling synchronously', function() { 
     var throttleTestResult = throttleTest(); 

     //throttled stream will not emit if scheduler clock is at zero 
     testScheduler.advanceBy(1); 
     throttleTestResult.unthrottledStream.onNext(); 
     throttleTestResult.unthrottledStream.onNext(); 
     throttleTestResult.unthrottledStream.onNext(); 

     testScheduler.advanceBy(throttleWindowDuration); 
     throttleTestResult.unthrottledStream.onNext(); 
     throttleTestResult.unthrottledStream.onNext(); 

     testScheduler.advanceBy(throttleWindowDuration); 
     throttleTestResult.unthrottledStream.onNext(); 

     testScheduler.advanceBy(throttleWindowDuration); 
     throttleTestResult.unthrottledStream.onNext(); 

     expect(throttleTestResult.emitCounter).toBe(4); 
     }); 
    }); 

    describe('run without test scheduler', function() { 
     it('without test scheduler the emit counter will stay at 1 ' 
     + 'as throttle duration is not elapsed', function() { 
     var throttleTestResult = throttleTest(); 

     throttleTestResult.unthrottledStream.onNext(); 
     throttleTestResult.unthrottledStream.onNext(); 
     throttleTestResult.unthrottledStream.onNext(); 
     throttleTestResult.unthrottledStream.onNext(); 
     expect(throttleTestResult.emitCounter).toBe(1); 
     }); 
    }); 

С RxJs 5 все намного проще. Вы можете просто использовать jasmine.clock, как и для тестирования других JavaScript-операций. Я продемонстрировал решение в своем blog post