geronimo-iia /
async-rx
| Total Complexity | 1 |
| Total Lines | 17 |
| Duplicated Lines | 0 % |
| Changes | 0 | ||
| 1 | from async_rx import rx_count |
||
| 2 | |||
| 3 | from ..model import ObserverCounterCollector |
||
| 4 | from .model import get_observable_to_21 |
||
| 5 | |||
| 6 | |||
| 7 | def test_rx_count(kernel): |
||
| 8 | |||
| 9 | seeker = ObserverCounterCollector() |
||
| 10 | sub = kernel.run(rx_count(observable=get_observable_to_21()).subscribe(seeker)) |
||
| 11 | kernel.run(sub()) |
||
| 12 | |||
| 13 | assert seeker.on_completed_count == 1 |
||
| 14 | assert seeker.on_next_count == 1 |
||
| 15 | assert seeker.on_error_count == 0 |
||
| 16 | assert seeker.items == [21] |
||
| 17 |
