Add an assertion like 'expect(observable).to.produce.error()' to test 'onError'.

Fixed wrong test codes with the assertion.

Changed the behaviors of the following functions caused by argument types, to raise an error in the creation phase.

- Observable.defer
- Observable:buffer
- Observable:elementAt
- Observable:skipLast
- Observable:takeLast
- Observable:window

doc/CONTRIBUTING.md

@@ -25,6 +25,7 @@ Tests
 - To run a specific test file, run `lua tests/runner.lua average` to just run the tests in `tests/average.lua`.
 - In addition to lust's default operators, there are also a few additional utilities available via `runner.lua`:
   - `expect(Observable).to.produce(...)` will assert that the Observable produces the specified values, in order.  If you need to assert against multiple values emitted by a single `onNext`, you can pass in a table (i.e. `{{1, 2, 3}, {4, 5, 6}}` to check that an Observable calls `onNext` twice with 3 values each).
+  - `expect(Observable).to.produce.error()` will assert that the Observable produces an error.
   - There is also `expect(Observable).to.produce.nothing()`.
   - `local onNext, onError, onCompleted = observableSpy(observable)` will create three spies for each of the three events.  You can read more about spies on lust's README.
 

rx.lua

@@ -249,6 +249,10 @@ end
 -- @arg {function} factory - A function that returns an Observable.
 -- @returns {Observable}
 function Observable.defer(fn)
+  if not fn or type(fn) ~= 'function' then
+    error('Expected a function')
+  end
+
   return setmetatable({
     subscribe = function(_, ...)
       local observable = fn()
@@ -396,6 +400,10 @@ end
 -- values.
 -- @arg {number} size - The size of the buffer.
 function Observable:buffer(size)
+  if not size or type(size) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local buffer = {}
 
@@ -802,6 +810,10 @@ end
 -- @arg {number} index - The index of the item, with an index of 1 representing the first.
 -- @returns {Observable}
 function Observable:elementAt(index)
+  if not index or type(index) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local subscription
     local i = 1
@@ -1354,6 +1366,10 @@ end
 -- @arg {number} count - The number of items to omit from the end.
 -- @returns {Observable}
 function Observable:skipLast(count)
+  if not count or type(count) ~= 'number' then
+    error('Expected a number')
+  end
+
   local buffer = {}
   return Observable.create(function(observer)
     local function emit()
@@ -1550,6 +1566,10 @@ end
 -- @arg {number} count - The number of elements to produce.
 -- @returns {Observable}
 function Observable:takeLast(count)
+  if not count or type(count) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local buffer = {}
 
@@ -1708,6 +1728,10 @@ end
 --                      of the most recent values as multiple arguments to onNext.
 -- @returns {Observable}
 function Observable:window(size)
+  if not size or type(size) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local window = {}
 

src/observable.lua

@@ -154,6 +154,10 @@ end
 -- @arg {function} factory - A function that returns an Observable.
 -- @returns {Observable}
 function Observable.defer(fn)
+  if not fn or type(fn) ~= 'function' then
+    error('Expected a function')
+  end
+
   return setmetatable({
     subscribe = function(_, ...)
       local observable = fn()

src/operators/buffer.lua

@@ -5,6 +5,10 @@ local util = require 'util'
 -- values.
 -- @arg {number} size - The size of the buffer.
 function Observable:buffer(size)
+  if not size or type(size) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local buffer = {}
 

src/operators/elementAt.lua

@@ -4,6 +4,10 @@ local Observable = require 'observable'
 -- @arg {number} index - The index of the item, with an index of 1 representing the first.
 -- @returns {Observable}
 function Observable:elementAt(index)
+  if not index or type(index) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local subscription
     local i = 1

src/operators/skipLast.lua

@@ -6,6 +6,10 @@ local util = require 'util'
 -- @arg {number} count - The number of items to omit from the end.
 -- @returns {Observable}
 function Observable:skipLast(count)
+  if not count or type(count) ~= 'number' then
+    error('Expected a number')
+  end
+
   local buffer = {}
   return Observable.create(function(observer)
     local function emit()

src/operators/takeLast.lua

@@ -6,6 +6,10 @@ local util = require 'util'
 -- @arg {number} count - The number of elements to produce.
 -- @returns {Observable}
 function Observable:takeLast(count)
+  if not count or type(count) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local buffer = {}
 

src/operators/window.lua

@@ -6,6 +6,10 @@ local util = require 'util'
 --                      of the most recent values as multiple arguments to onNext.
 -- @returns {Observable}
 function Observable:window(size)
+  if not size or type(size) ~= 'number' then
+    error('Expected a number')
+  end
+
   return Observable.create(function(observer)
     local window = {}
 

tests/all.lua

@@ -1,18 +1,14 @@
 describe('all', function()
   it('passes through errors', function()
-    expect(Rx.Observable.throw():all().subscribe).to.fail()
+    expect(Rx.Observable.throw():all()).to.produce.error()
   end)
 
   it('calls onError if the predicate errors', function()
-    local observable = Rx.Observable.fromRange(3):all(error)
-    local onError = spy()
-    observable:subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):all(error)).to.produce.error()
   end)
 
   it('produces an error if the parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():all(function(x) return x end))
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():all(function(x) return x end)).to.produce.error()
   end)
 
   it('produces true if all elements satisfy the predicate', function()

tests/average.lua

@@ -1,7 +1,6 @@
 describe('average', function()
   it('errors when its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():average())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():average()).to.produce.error()
   end)
 
   it('produces a single value representing the average of the values produced by the source', function()

tests/buffer.lua

@@ -1,11 +1,10 @@
 describe('buffer', function()
   it('produces an error if its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():buffer())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():buffer(1)).to.produce.error()
   end)
 
   it('fails if size is not specified', function()
-    expect(Rx.Observable.fromRange(5):buffer().subscribe).to.fail()
+    expect(function () Rx.Observable.fromRange(5):buffer() end).to.fail()
   end)
 
   it('produces values wrapped to the specified width', function()

tests/catch.lua

@@ -31,9 +31,7 @@ describe('catch', function()
 
   it('calls onError if the supplied function errors', function()
     local handler = error
-    local onError = spy()
-    Rx.Observable.throw():catch(handler):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():catch(handler)).to.produce.error()
   end)
 
   it('calls onComplete when the parent completes', function()

tests/combineLatest.lua

@@ -69,13 +69,7 @@ describe('combineLatest', function()
     expect(#errored).to.equal(1)
   end)
 
-  it('calls onError if the combinator is absent', function()
-    expect(Rx.Observable.combineLatest(Rx.Observable.fromRange(3)).subscribe).to.fail()
-  end)
-
   it('calls onError if the combinator errors', function()
-    local onError = spy()
-    Rx.Observable.combineLatest(Rx.Observable.fromRange(3), error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.combineLatest(Rx.Observable.fromRange(3), error)).to.produce.error()
   end)
 end)

tests/compact.lua

@@ -1,8 +1,8 @@
 describe('compact', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:compact().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:compact()).to.produce.error()
   end)
 
   it('does not produce values that are false or nil', function()

tests/concat.lua

@@ -1,7 +1,6 @@
 describe('concat', function()
   it('produces an error if its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():concat())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():concat()).to.produce.error()
   end)
 
   it('returns the first argument if it is the only argument', function()
@@ -28,7 +27,7 @@ describe('concat', function()
   it('should error if any of the sources error', function()
     local badObservable = Rx.Observable.create(function(observer) observer:onError('oh no') end)
     local observable = Rx.Observable.of(1):concat(Rx.Observable.of(2), badObservable)
-    expect(observable.subscribe).to.fail()
+    expect(observable).to.produce.error()
   end)
 
   it('should complete once the rightmost observable completes', function()

tests/contains.lua

@@ -1,7 +1,6 @@
 describe('contains', function()
   it('errors when its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():contains(1))
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():contains(1)).to.produce.error()
   end)
 
   it('returns false if the source Observable produces no values', function()

tests/count.lua

@@ -1,7 +1,6 @@
 describe('count', function()
   it('passes through errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():count())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():count()).to.produce.error()
   end)
 
   it('produces a single value representing the number of elements produced by the source', function()
@@ -15,8 +14,6 @@ describe('count', function()
   end)
 
   it('calls onError if the predicate errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):count(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):count(error)).to.produce.error()
   end)
 end)

tests/defaultIfEmpty.lua

@@ -1,7 +1,6 @@
 describe('defaultIfEmpty', function()
   it('errors if the source errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():defaultIfEmpty(1))
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():defaultIfEmpty(1)).to.produce.error()
   end)
 
   it('produces the values from the source unchanged if at least one value is produced', function()

tests/distinct.lua

@@ -5,8 +5,7 @@ describe('distinct', function()
   end)
 
   it('produces an error if its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():distinct())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():distinct()).to.produce.error()
   end)
 
   it('completes when its parent completes', function()

tests/distinctUntilChanged.lua

@@ -1,7 +1,6 @@
 describe('distinctUntilChanged', function()
   it('produces an error if its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():distinctUntilChanged())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():distinctUntilChanged()).to.produce.error()
   end)
 
   describe('with the default comparator', function()
@@ -33,9 +32,7 @@ describe('distinctUntilChanged', function()
     end)
 
     it('calls onError if the comparator errors', function()
-      local onError = spy()
-      Rx.Observable.fromRange(2):distinctUntilChanged(error):subscribe(nil, onError, nil)
-      expect(#onError).to.equal(1)
+      expect(Rx.Observable.fromRange(2):distinctUntilChanged(error)).to.produce.error()
     end)
   end)
 end)

tests/elementAt.lua

@@ -1,7 +1,6 @@
 describe('elementAt', function()
   it('errors when its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable:throw():elementAt(0))
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable:throw():elementAt(0)).to.produce.error()
   end)
 
   it('chains subscriptions', function()
@@ -22,7 +21,7 @@ describe('elementAt', function()
   end)
 
   it('errors if no index is specified', function()
-    expect(Rx.Observable.of(1):elementAt().subscribe).to.fail()
+    expect(function () Rx.Observable.of(1):elementAt() end).to.fail()
   end)
 
   it('produces no values if the specified index is less than one', function()

tests/filter.lua

@@ -17,13 +17,10 @@ describe('filter', function()
   end)
 
   it('errors when its parent errors', function()
-    local _, onError = observableSpy(Rx.Observable.throw():filter())
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():filter()).to.produce.error()
   end)
 
   it('calls onError if the predicate errors', function()
-    local onError = spy()
-    Rx.Observable.of(5):filter(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.of(5):filter(error)).to.produce.error()
   end)
 end)

tests/find.lua

@@ -1,14 +1,12 @@
 describe('find', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:find().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:find()).to.produce.error()
   end)
 
   it('calls onError if the predicate errors', function()
-    local onError = spy()
-    Rx.Observable.of(3):find(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.of(3):find(error)).to.produce.error()
   end)
 
   it('uses the identity function as a predicate if none is specified', function()

tests/first.lua

@@ -1,8 +1,8 @@
 describe('first', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:first().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:first()).to.produce.error()
   end)
 
   it('produces no elements if its parent produces no elements', function()

tests/flatMap.lua

@@ -1,7 +1,7 @@
 describe('flatMap', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):flatMap(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
+    expect(observable).to.produce.error()
   end)
 
   it('uses the identity function as the callback if none is specified', function()

tests/flatMapLatest.lua

@@ -4,9 +4,7 @@ describe('flatMapLatest', function()
   end)
 
   it('produces an error if the callback errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):flatMapLatest(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):flatMapLatest(error)).to.produce.error()
   end)
 
   it('unsubscribes from the source and the projected observable', function()

tests/flatten.lua

@@ -1,8 +1,8 @@
 describe('flatten', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:flatten().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:flatten()).to.produce.error()
   end)
 
   it('produces all values produced by the observables produced by its parent', function()

tests/ignoreElements.lua

@@ -1,6 +1,6 @@
 describe('ignoreErrors', function()
   it('passes through errors from the source', function()
-    expect(Rx.Observable.throw():ignoreElements().subscribe).to.fail()
+    expect(Rx.Observable.throw():ignoreElements()).to.produce.error()
   end)
 
   it('does not produce any values produced by the source', function()

tests/last.lua

@@ -1,8 +1,8 @@
 describe('last', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:last().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:last()).to.produce.error()
   end)
 
   it('produces no elements if its parent produces no elements', function()

tests/map.lua

@@ -1,8 +1,8 @@
 describe('map', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.throw():map(function(x) return x end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:map().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:map()).to.produce.error()
   end)
 
   it('uses the identity function as the callback if none is specified', function()
@@ -18,8 +18,6 @@ describe('map', function()
   end)
 
   it('calls onError if the callback errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):map(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):map(error)).to.produce.error()
   end)
 end)

tests/max.lua

@@ -1,13 +1,13 @@
 describe('max', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:max().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:max()).to.produce.error()
   end)
 
   it('produces an error if one of the values produced is a string', function()
-    local observable = Rx.Observable.of('string'):max()
-    expect(observable.subscribe).to.fail()
+    local observable = Rx.Observable.of(1, 'string'):max()
+    expect(observable).to.produce.error()
   end)
 
   it('produces the maximum of all values produced', function()

tests/min.lua

@@ -1,13 +1,13 @@
 describe('min', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:min().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:min()).to.produce.error()
   end)
 
   it('produces an error if one of the values produced is a string', function()
-    local observable = Rx.Observable.of('string'):min()
-    expect(observable.subscribe).to.fail()
+    local observable = Rx.Observable.of(1, 'string'):min()
+    expect(observable).to.produce.error()
   end)
 
   it('produces the minimum of all values produced', function()

tests/observable.lua

@@ -249,15 +249,15 @@ describe('Observable', function()
 
   describe('defer', function()
     it('returns an Observable', function()
-      expect(Rx.Observable.defer()).to.be.an(Rx.Observable)
+      expect(Rx.Observable.defer(function() end)).to.be.an(Rx.Observable)
     end)
 
     it('fails if no factory is specified', function()
-      expect(Rx.Observable.defer().subscribe).to.fail()
+      expect(function () Rx.Observable.defer() end).to.fail()
     end)
 
     it('fails if the factory does not return an Observable', function()
-      expect(Rx.Observable.defer(function() return nil end).subscribe).to.fail()
+      expect(function () Rx.Observable.defer(function() end):subscribe() end).to.fail()
     end)
 
     it('uses the factory function to create a new Observable for each subscriber', function()

tests/pack.lua

@@ -1,8 +1,8 @@
 describe('pack', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:pack().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:pack()).to.produce.error()
   end)
 
   it('wraps elements of the source in tables', function()

tests/partition.lua

@@ -1,8 +1,8 @@
 describe('partition', function()
   it('errors when its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:partition().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:partition()).to.produce.error()
   end)
 
   it('uses the identity function as the predicate if none is specified', function()

tests/reduce.lua

@@ -35,8 +35,6 @@ describe('reduce', function()
   end)
 
   it('calls onError if the accumulator errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):reduce(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):reduce(error)).to.produce.error()
   end)
 end)

tests/reject.lua

@@ -18,13 +18,11 @@ describe('reject', function()
 
   it('errors when its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:reject().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:reject()).to.produce.error()
   end)
 
   it('calls onError when the predicate errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):reject(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):reject(error)).to.produce.error()
   end)
 end)

tests/runner.lua

@@ -16,6 +16,7 @@ end
 
 lust.paths['produce'] = {
   'nothing',
+  'error',
   f = function(observable, ...)
     local args = {...}
     local values
@@ -48,6 +49,15 @@ lust.paths['nothing'] = {
   end
 }
 
+lust.paths['error'] = {
+  f = function(observable)
+    local _, onError = observableSpy(observable)
+    expect(observable).to.be.an(Rx.Observable)
+    expect(#onError).to.equal(1)
+    return true
+  end
+}
+
 table.insert(lust.paths['to'], 'produce')
 
 if arg[1] then

tests/sample.lua

@@ -42,14 +42,12 @@ describe('sample', function()
   it('errors when the source errors', function()
     local a = Rx.Observable.throw()
     local b = Rx.Observable.fromRange(3)
-    local onNext, onError, onCompleted = observableSpy(a:sample(b))
-    expect(#onError).to.equal(1)
+    expect(a:sample(b)).to.produce.error()
   end)
 
   it('errors when the sampler errors', function()
     local a = Rx.Observable.fromRange(3)
     local b = Rx.Observable.throw()
-    local onNext, onError, onCompleted = observableSpy(a:sample(b))
-    expect(#onError).to.equal(1)
+    expect(a:sample(b)).to.produce.error()
   end)
 end)

tests/scan.lua

@@ -35,8 +35,6 @@ describe('scan', function()
   end)
 
   it('calls onError if the accumulator errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):scan(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):scan(error)).to.produce.error()
   end)
 end)

tests/skip.lua

@@ -1,8 +1,8 @@
 describe('skip', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:skip(1).subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:skip(1)).to.produce.error()
   end)
 
   it('produces all values if the count is zero', function()

tests/skipLast.lua

@@ -1,10 +1,10 @@
 describe('skipLast', function()
   it('produces an error if its parent errors', function()
-    expect(Rx.Observable.throw():skipLast(1).subscribe).to.fail()
+    expect(Rx.Observable.throw():skipLast(1)).to.produce.error()
   end)
 
   it('fails if the count is not specified', function()
-    expect(Rx.Observable.fromRange(3):skipLast().subscribe).to.fail()
+    expect(function () Rx.Observable.fromRange(3):skipLast() end).to.fail()
   end)
 
   it('skips the specified number of values from the end of the source Observable', function()

tests/skipUntil.lua

@@ -1,8 +1,9 @@
 describe('skipUntil', function()
   it('produces an error if its parent errors', function()
+    local trigger = Rx.Observable.of()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:skipUntil(1).subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:skipUntil(trigger)).to.produce.error()
   end)
 
   it('fails if the first argument is not an Observable', function()

tests/skipWhile.lua

@@ -1,8 +1,8 @@
 describe('skipWhile', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:skipWhile(function() end).subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:skipWhile(function() end)).to.produce.error()
   end)
 
   it('uses the identity function if no predicate is specified', function()
@@ -23,8 +23,6 @@ describe('skipWhile', function()
   end)
 
   it('calls onError if the predicate errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):skipWhile(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):skipWhile(error)).to.produce.error()
   end)
 end)

tests/startWith.lua

@@ -1,6 +1,6 @@
 describe('startWith', function()
   it('produces errors emitted by the source', function()
-    expect(Rx.Observable.throw():startWith(1).subscribe).to.fail()
+    expect(Rx.Observable.throw():startWith(1)).to.produce.error()
   end)
 
   it('produces all specified elements in a single onNext before producing values normally', function()

tests/sum.lua

@@ -1,6 +1,6 @@
 describe('sum', function()
   it('passes through errors from the source', function()
-    expect(Rx.Observable.throw():sum().subscribe).to.fail()
+    expect(Rx.Observable.throw():sum()).to.produce.error()
   end)
 
   it('produces the sum of the numeric values from the source', function()

tests/switch.lua

@@ -1,6 +1,6 @@
 describe('switch', function()
   it('errors when the source errors', function()
-    expect(Rx.Observable.throw():switch().subscribe).to.fail()
+    expect(Rx.Observable.throw():switch()).to.produce.error()
   end)
 
   it('errors when an Observable produced by the source errors', function()
@@ -9,7 +9,7 @@ describe('switch', function()
       observer:onCompleted()
     end)
 
-    expect(observable:switch().subscribe).to.fail()
+    expect(observable:switch()).to.produce.error()
   end)
 
   it('produces the values produced by the latest Observable produced by the source', function()

tests/take.lua

@@ -1,8 +1,8 @@
 describe('take', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:take(1).subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:take(1)).to.produce.error()
   end)
 
   it('produces nothing if the count is zero', function()

tests/takeLast.lua

@@ -1,10 +1,10 @@
 describe('takeLast', function()
   it('produces an error if its parent errors', function()
-    expect(Rx.Observable.throw():takeLast(1).subscribe).to.fail()
+    expect(Rx.Observable.throw():takeLast(1)).to.produce.error()
   end)
 
   it('produces an error if the count is not specified', function()
-    expect(Rx.Observable.fromRange(3):takeLast().subscribe).to.fail()
+    expect(function () Rx.Observable.fromRange(3):takeLast() end).to.fail()
   end)
 
   it('produces nothing if the count is zero', function()

tests/takeUntil.lua

@@ -1,8 +1,9 @@
 describe('takeUntil', function()
   it('produces an error if its parent errors', function()
+    local trigger = Rx.Observable.create(function() end)
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:takeUntil().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:takeUntil(trigger)).to.produce.error()
   end)
 
   it('fails if the first argument is not an Observable', function()

tests/takeWhile.lua

@@ -1,8 +1,8 @@
 describe('takeWhile', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:takeWhile(function() end).subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:takeWhile(function() end)).to.produce.error()
   end)
 
   it('uses the identity function if no predicate is specified', function()
@@ -23,8 +23,6 @@ describe('takeWhile', function()
   end)
 
   it('calls onError if the predicate errors', function()
-    local onError = spy()
-    Rx.Observable.fromRange(3):takeWhile(error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.fromRange(3):takeWhile(error)).to.produce.error()
   end)
 end)

tests/tap.lua

@@ -14,12 +14,7 @@ describe('tap', function()
   end)
 
   it('calls onError if the onNext callback errors', function()
-    local onNext = spy()
-    local onError = spy()
-    local observer = Rx.Observer.create(onNext, onError)
-    Rx.Observable.of(1):tap(error):subscribe(observer)
-    expect(#onNext).to.equal(0)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.of(1):tap(error)).to.produce.error()
   end)
 
   it('runs the specified onError function', function()
@@ -31,9 +26,7 @@ describe('tap', function()
   end)
 
   it('calls onError if the onError callback errors', function()
-    local onError = spy()
-    Rx.Observable.throw():tap(nil, error):subscribe(nil, onError, nil)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.throw():tap(nil, error)).to.produce.error()
   end)
 
   it('runs the specified onCompleted function', function()
@@ -45,10 +38,6 @@ describe('tap', function()
   end)
 
   it('calls onError if the onCompleted callback errors', function()
-    local onError = spy()
-    local onCompleted = spy()
-    Rx.Observable.of(1):tap(nil, nil, error):subscribe(nil, onError, onCompleted)
-    expect(#onCompleted).to.equal(0)
-    expect(#onError).to.equal(1)
+    expect(Rx.Observable.of(1):tap(nil, nil, error)).to.produce.error()
   end)
 end)

tests/unpack.lua

@@ -1,12 +1,12 @@
 describe('unpack', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:unpack().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:unpack()).to.produce.error()
   end)
 
   it('fails if the observable produces an element that is not a table', function()
-    expect(Rx.Observable.of(3):unpack().subscribe).to.fail()
+    expect(Rx.Observable.of(3):unpack()).to.produce.error()
   end)
 
   it('produces all elements in the tables produced as multiple values', function()

tests/unwrap.lua

@@ -1,8 +1,8 @@
 describe('unwrap', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:unwrap().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:unwrap()).to.produce.error()
   end)
 
   it('produces any multiple values as individual values', function()

tests/window.lua

@@ -1,12 +1,12 @@
 describe('window', function()
   it('produces an error if its parent errors', function()
     local observable = Rx.Observable.of(''):map(function(x) return x() end)
-    expect(observable.subscribe).to.fail()
-    expect(observable:window().subscribe).to.fail()
+    expect(observable).to.produce.error()
+    expect(observable:window(2)).to.produce.error()
   end)
 
   it('fails if size is not specified', function()
-    expect(Rx.Observable.fromRange(5):window().subscribe).to.fail()
+    expect(function () Rx.Observable.fromRange(5):window() end).to.fail()
   end)
 
   it('produces a specified number of the most recent values', function()