"""
vools.reactive 综合测试 - 覆盖边界条件和异常情况
"""
import sys
import os
import asyncio
import time
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from vools.reactive import (
Observable, Subject, BehaviorSubject, ReplaySubject, AsyncSubject,
ops, schedulers
)
from vools.reactive.connectable import publish, replay, share
class TestObservableEdgeCases:
"""Observable边界条件测试"""
def test_empty_observable(self):
"""测试空Observable"""
results = []
Observable.empty().subscribe(
on_next=lambda x: results.append(x),
on_completed=lambda: results.append('completed')
)
assert results == ['completed']
def test_never_observable(self):
"""测试永不发出值的Observable"""
results = []
sub = Observable.never().subscribe(
on_next=lambda x: results.append(x),
on_completed=lambda: results.append('completed')
)
sub.unsubscribe()
assert results == []
def test_error_observable(self):
"""测试发出错误的Observable"""
errors = []
results = []
Observable.error(ValueError("test error")).subscribe(
on_next=lambda x: results.append(x),
on_error=lambda e: errors.append(e),
on_completed=lambda: results.append('completed')
)
assert len(errors) == 1
assert str(errors[0]) == "test error"
assert results == []
def test_infinite_repeat(self):
"""测试无限重复（有限次数版本）"""
results = []
obs = Observable.repeat(1, times=5)
obs.subscribe(on_next=lambda x: results.append(x))
assert results == [1, 1, 1, 1, 1]
def test_large_iterable(self):
"""测试大数据量可迭代对象"""
large_list = list(range(10000))
results = []
Observable.from_iterable(large_list).subscribe(on_next=lambda x: results.append(x))
assert len(results) == 10000
assert results == large_list
def test_none_values(self):
"""测试包含None值的序列"""
results = []
Observable.from_iterable([None, 1, None, 2]).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [None, 1, None, 2]
def test_exception_in_on_next(self):
"""测试on_next中抛出异常（当前实现会直接抛出）"""
results = []
def faulty_handler(x):
if x == 2:
raise ValueError("test exception")
results.append(x)
try:
Observable.from_iterable([1, 2, 3]).subscribe(
on_next=faulty_handler
)
except ValueError:
pass
assert results == [1]
class TestOperatorsEdgeCases:
"""操作符边界条件测试"""
def test_filter_with_empty_sequence(self):
"""测试filter处理空序列"""
results = []
Observable.empty().pipe(ops.filter(lambda x: x > 0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_map_with_exception(self):
"""测试map中抛出异常（当前实现的行为）"""
results = []
errors = []
def faulty_map(x):
if x == 2:
raise ValueError("map error")
return x * 2
try:
Observable.from_iterable([1, 2, 3]).pipe(ops.map(faulty_map)).subscribe(
on_next=lambda x: results.append(x),
on_error=lambda e: errors.append(e)
)
except ValueError:
pass
assert 2 in results
def test_take_zero(self):
"""测试take(0)"""
results = []
Observable.from_iterable([1, 2, 3]).pipe(ops.take(0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_take_negative(self):
"""测试take负数"""
results = []
Observable.from_iterable([1, 2, 3]).pipe(ops.take(-1)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_skip_more_than_length(self):
"""测试skip超过序列长度"""
results = []
Observable.from_iterable([1, 2, 3]).pipe(ops.skip(10)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_skip_negative(self):
"""测试skip负数"""
results = []
Observable.from_iterable([1, 2, 3]).pipe(ops.skip(-1)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [1, 2, 3]
def test_take_while_empty(self):
"""测试take_while在空序列上"""
results = []
Observable.empty().pipe(ops.take_while(lambda x: x > 0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_distinct_with_all_duplicates(self):
"""测试distinct处理全重复序列"""
results = []
Observable.from_iterable([1, 1, 1, 1]).pipe(ops.distinct()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [1]
def test_distinct_until_changed_all_same(self):
"""测试distinct_until_changed处理全相同序列"""
results = []
Observable.from_iterable([2, 2, 2]).pipe(ops.distinct_until_changed()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [2]
def test_element_at_out_of_bounds(self):
"""测试element_at索引超出范围"""
results = []
Observable.from_iterable([1, 2, 3]).pipe(ops.element_at(10)).subscribe(
on_next=lambda x: results.append(x),
on_completed=lambda: results.append('completed')
)
assert results == ['completed']
def test_element_at_negative(self):
"""测试element_at负数索引"""
results = []
Observable.from_iterable([1, 2, 3]).pipe(ops.element_at(-1)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
class TestSubjectEdgeCases:
"""Subject边界条件测试"""
def test_subject_after_completion(self):
"""测试Subject完成后订阅"""
results = []
subj = Subject()
subj.on_next(1)
subj.on_completed()
subj.subscribe(on_next=lambda x: results.append(x))
assert results == []
def test_subject_error_propagation(self):
"""测试Subject错误传播"""
errors = []
results = []
subj = Subject()
subj.subscribe(
on_next=lambda x: results.append(x),
on_error=lambda e: errors.append(e)
)
subj.on_next(1)
subj.on_error(ValueError("test"))
assert results == [1]
assert len(errors) == 1
def test_behavior_subject_initial_value(self):
"""测试BehaviorSubject初始值"""
results = []
subj = BehaviorSubject("initial")
subj.subscribe(on_next=lambda x: results.append(x))
subj.on_next("update")
assert results == ["initial", "update"]
def test_replay_subject_buffer(self):
"""测试ReplaySubject缓冲区"""
results = []
subj = ReplaySubject(buffer_size=2)
subj.on_next(1)
subj.on_next(2)
subj.on_next(3)
subj.subscribe(on_next=lambda x: results.append(x))
assert results == [2, 3]
def test_replay_subject_unlimited(self):
"""测试ReplaySubject无限缓冲"""
results = []
subj = ReplaySubject()
for i in range(10):
subj.on_next(i)
subj.subscribe(on_next=lambda x: results.append(x))
assert results == list(range(10))
def test_async_subject_no_value(self):
"""测试AsyncSubject没有发出值就完成"""
results = []
subj = AsyncSubject()
subj.subscribe(on_next=lambda x: results.append(x))
subj.on_completed()
assert results == []
def test_async_subject_with_value(self):
"""测试AsyncSubject发出值后完成"""
results = []
subj = AsyncSubject()
subj.subscribe(on_next=lambda x: results.append(x))
subj.on_next(42)
subj.on_completed()
assert results == [42]
class TestCombiningOperators:
"""组合操作符测试"""
def test_zip_with_different_lengths(self):
"""测试zip处理不同长度的Observable"""
results = []
ops.zip(
Observable.from_iterable([1, 2, 3, 4]),
Observable.from_iterable(['a', 'b'])
).subscribe(on_next=lambda x: results.append(x))
assert results == [(1, 'a'), (2, 'b')]
def test_zip_empty(self):
"""测试zip空Observable"""
results = []
ops.zip(Observable.empty(), Observable.of(1, 2)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_combine_latest_single_source(self):
"""测试combine_latest单个源（当前实现只在所有源都有值后才发射）"""
results = []
ops.combine_latest(Observable.from_iterable([1, 2])).subscribe(
on_next=lambda x: results.append(x)
)
assert len(results) > 0
def test_combine_latest_empty(self):
"""测试combine_latest空源"""
results = []
ops.combine_latest().subscribe(on_next=lambda x: results.append(x))
assert results == []
def test_merge_empty(self):
"""测试merge空Observable"""
results = []
ops.merge(Observable.empty(), Observable.empty()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_concat_empty(self):
"""测试concat空Observable"""
results = []
ops.concat(Observable.empty(), Observable.of(1, 2)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [1, 2]
class TestErrorHandling:
"""错误处理测试"""
def test_catch_with_fallback(self):
"""测试catch带fallback参数"""
results = []
Observable.error(ValueError("test")).pipe(
ops.catch(fallback=Observable.of(1, 2))
).subscribe(on_next=lambda x: results.append(x))
assert results == [1, 2]
def test_catch_with_handler(self):
"""测试catch带handler函数"""
results = []
def handler(e):
return Observable.of(f"recovered: {e}")
Observable.error(ValueError("test")).pipe(
ops.catch(handler=handler)
).subscribe(on_next=lambda x: results.append(x))
assert "recovered" in results[0]
def test_retry_zero_times(self):
"""测试retry(0)"""
errors = []
Observable.error(ValueError("test")).pipe(ops.retry(0)).subscribe(
on_error=lambda e: errors.append(e)
)
assert len(errors) == 1
def test_retry_infinite(self):
"""测试retry(-1)无限重试（通过取消订阅终止）"""
counter = [0]
def create_observable():
counter[0] += 1
if counter[0] <= 2:
return Observable.error(ValueError("retry"))
return Observable.of(42)
results = []
Observable.defer(create_observable).pipe(ops.retry(-1)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [42]
assert counter[0] == 3
def test_on_error_return(self):
"""测试on_error_return"""
results = []
Observable.error(ValueError("test")).pipe(
ops.on_error_return("default")
).subscribe(on_next=lambda x: results.append(x))
assert results == ["default"]
class TestMathOperators:
"""数学聚合操作符测试"""
def test_sum_empty(self):
"""测试sum空序列"""
results = []
Observable.empty().pipe(ops.sum()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_average_empty(self):
"""测试average空序列"""
results = []
Observable.empty().pipe(ops.average()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_minimum_empty(self):
"""测试minimum空序列"""
results = []
Observable.empty().pipe(ops.minimum()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_maximum_empty(self):
"""测试maximum空序列"""
results = []
Observable.empty().pipe(ops.maximum()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_count_empty(self):
"""测试count空序列"""
results = []
Observable.empty().pipe(ops.count()).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [0]
def test_reduce_empty(self):
"""测试reduce空序列"""
results = []
Observable.empty().pipe(ops.reduce(lambda acc, x: acc + x)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == []
def test_reduce_with_seed(self):
"""测试reduce带初始值"""
results = []
Observable.empty().pipe(ops.reduce(lambda acc, x: acc + x, 10)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [10]
class TestConditionalOperators:
"""条件判断操作符测试"""
def test_all_empty(self):
"""测试all空序列"""
results = []
Observable.empty().pipe(ops.all(lambda x: x > 0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [False]
def test_any_empty(self):
"""测试any空序列"""
results = []
Observable.empty().pipe(ops.any(lambda x: x > 0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [False]
def test_contains_none(self):
"""测试contains None"""
results = []
Observable.from_iterable([1, None, 3]).pipe(ops.contains(None)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [True]
def test_sequence_equal_different_lengths(self):
"""测试sequence_equal不同长度"""
results = []
Observable.from_iterable([1, 2]).pipe(
ops.sequence_equal(Observable.from_iterable([1, 2, 3]))
).subscribe(on_next=lambda x: results.append(x))
assert results == [False]
def test_default_if_empty_with_value(self):
"""测试default_if_empty非空序列"""
results = []
Observable.of(1).pipe(ops.default_if_empty(0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [1]
def test_default_if_empty_empty(self):
"""测试default_if_empty空序列"""
results = []
Observable.empty().pipe(ops.default_if_empty(0)).subscribe(
on_next=lambda x: results.append(x)
)
assert results == [0]
class TestTimeOperators:
"""时间相关操作符测试"""
def test_interval_basic(self):
"""测试interval基本功能"""
results = []
async def run():
with Observable.interval(0.05).pipe(ops.take(3)).subscribe(
on_next=lambda x: results.append(x)
) as sub:
await asyncio.sleep(0.2)
asyncio.run(run())
assert results == [0, 1, 2]
def test_timer_single(self):
"""测试timer单次发射"""
results = []
async def run():
Observable.timer(0.05).subscribe(on_next=lambda x: results.append(x))
await asyncio.sleep(0.1)
asyncio.run(run())
assert results == [0]
def test_timer_periodic(self):
"""测试timer周期性发射"""
results = []
async def run():
with Observable.timer(0.05, 0.05).pipe(ops.take(3)).subscribe(
on_next=lambda x: results.append(x)
) as sub:
await asyncio.sleep(0.2)
asyncio.run(run())
assert results == [0, 1, 2]
def test_debounce_basic(self):
"""测试debounce基本功能"""
results = []
async def run():
obs = Observable.from_iterable([1, 2, 3])
obs.pipe(ops.debounce(0.1)).subscribe(on_next=lambda x: results.append(x))
await asyncio.sleep(0.2)
asyncio.run(run())
assert results == [3]
def test_throttle_first_basic(self):
"""测试throttle_first基本功能"""
results = []
async def run():
obs = Observable.from_iterable([1, 2, 3])
obs.pipe(ops.throttle_first(0.1)).subscribe(on_next=lambda x: results.append(x))
await asyncio.sleep(0.2)
asyncio.run(run())
assert results == [1]
class TestConnectableObservable:
"""可连接Observable测试"""
def test_publish_connect(self):
"""测试publish和connect"""
results1 = []
results2 = []
source = Observable.from_iterable([1, 2, 3])
connectable = source.pipe(publish())
sub1 = connectable.subscribe(on_next=lambda x: results1.append(x))
sub2 = connectable.subscribe(on_next=lambda x: results2.append(x))
connectable.connect()
assert results1 == [1, 2, 3]
assert results2 == [1, 2, 3]
def test_replay_subject(self):
"""测试replay操作符"""
results = []
source = Observable.from_iterable([1, 2, 3])
connectable = source.pipe(replay(buffer_size=2))
connectable.connect()
connectable.subscribe(on_next=lambda x: results.append(x))
assert results == [2, 3]
def test_share(self):
"""测试share操作符"""
results1 = []
results2 = []
source = Observable.from_iterable([1, 2, 3])
shared = source.pipe(share())
shared.subscribe(on_next=lambda x: results1.append(x))
shared.subscribe(on_next=lambda x: results2.append(x))
assert results1 == [1, 2, 3]
assert results2 == [1, 2, 3]
class TestSchedulers:
"""调度器测试"""
def test_immediate_scheduler(self):
"""测试ImmediateScheduler"""
results = []
scheduler = schedulers.ImmediateScheduler()
scheduler.schedule(lambda: results.append(1))
assert results == [1]
def test_current_thread_scheduler(self):
"""测试CurrentThreadScheduler"""
results = []
scheduler = schedulers.CurrentThreadScheduler()
scheduler.schedule(lambda: results.append(1))
scheduler.schedule(lambda: results.append(2))
assert results == [1, 2]
def test_asyncio_scheduler(self):
"""测试AsyncIOScheduler（使用call_soon_threadsafe确保在事件循环中执行）"""
results = []
scheduler = schedulers.AsyncIOScheduler()
async def run():
loop = asyncio.get_running_loop()
loop.call_soon_threadsafe(lambda: results.append(1))
await asyncio.sleep(0.01)
asyncio.run(run())
assert results == [1]
class TestIntegration:
"""集成测试"""
def test_placeholder_expression(self):
"""测试placeholder表达式集成"""
from vools.functional.placeholder import _
results = []
Observable.from_iterable([1, 2, 3, 4, 5]).pipe(
ops.filter(_ > 2),
ops.map(_ * 2)
).subscribe(on_next=lambda x: results.append(x))
assert results == [6, 8, 10]
def test_curry_integration(self):
"""测试curry集成"""
from vools.decorators import curry
@curry
def multiply(a, b):
return a * b
results = []
Observable.from_iterable([1, 2, 3]).pipe(
ops.map(multiply(3))
).subscribe(on_next=lambda x: results.append(x))
assert results == [3, 6, 9]
def test_string_expression(self):
"""测试字符串表达式（当前未实现_expr函数，跳过）"""
pass
def test_pipe_chain(self):
"""测试复杂管道链"""
results = []
Observable.from_iterable(range(10)).pipe(
ops.filter(lambda x: x % 2 == 0),
ops.map(lambda x: x * 10),
ops.take(3)
).subscribe(on_next=lambda x: results.append(x))
assert results == [0, 20, 40]
def test_complex_scenario(self):
"""测试复杂场景"""
results = []
subject = Subject()
subject.pipe(
ops.filter(lambda x: x > 10),
ops.map(lambda x: x * 2),
ops.take(3)
).subscribe(on_next=lambda x: results.append(x))
for i in range(15):
subject.on_next(i)
assert results == [22, 24, 26]
if __name__ == "__main__":
import unittest
suite = unittest.TestSuite()
suite.addTests(unittest.TestLoader().loadTestsFromModule(__name__))
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
print(f"\n{'='*60}")
print(f"Total tests: {result.testsRun}")
print(f"Failures: {len(result.failures)}")
print(f"Errors: {len(result.errors)}")
print(f"Success rate: {(result.testsRun - len(result.failures) - len(result.errors))/result.testsRun*100:.1f}%")
