Coverage for tests / brokers / supervisor / test_supervisor.py: 95%

1import asyncio (empty)

2import logging (empty)

3from contextlib import suppress (empty)

4from unittest.mock import patch (empty)

5 

6import pytest (empty)

7 

8from faststream._internal.endpoint.subscriber.supervisor import ( (empty)

9 TaskCallbackSupervisor, 

10 _SupervisorCache, 

11) 

12 

13 

14@pytest.mark.asyncio() (empty)

15async def test_task_failing(subscriber_with_task_mixin): (empty)

16 async def failing_task(): (empty)

17 raise ValueError (empty)

18 

19 logging.disable(logging.CRITICAL + 1) (empty)

20 

21 task = subscriber_with_task_mixin.add_task(failing_task) (empty)

22 with suppress(ValueError): (empty)

23 await task (empty)

24 

25 assert len(subscriber_with_task_mixin.tasks) > 1 (empty)

26 assert len(TaskCallbackSupervisor._TaskCallbackSupervisor__cache) == 1 (empty)

27 

28 

29@pytest.mark.asyncio() (empty)

30async def test_task_successful(subscriber_with_task_mixin): (empty)

31 async def successful_task(): (empty)

32 return True (empty)

33 

34 task = subscriber_with_task_mixin.add_task(successful_task) (empty)

35 await task (empty)

36 assert len(subscriber_with_task_mixin.tasks) == 1 (empty)

37 assert task.result() (empty)

38 

39 

40@pytest.mark.asyncio() (empty)

41@pytest.mark.slow() (empty)

42async def test_ignore_cancellation_error(subscriber_with_task_mixin): (empty)

43 async def cancelled_task(): (empty)

44 await asyncio.sleep(10) 

45 return True 

46 

47 task = subscriber_with_task_mixin.add_task(cancelled_task) (empty)

48 task.cancel() (empty)

49 with pytest.raises(asyncio.CancelledError): (empty)

50 await task (empty)

51 

52 await asyncio.sleep(3) (empty)

53 assert len(subscriber_with_task_mixin.tasks) == 1 (empty)

54 

55 

56def test_supervisor_cache(monkeypatch): (empty)

57 with patch("time.time") as mocked_time: (empty)

58 mocked_time.return_value = 0 (empty)

59 cache = _SupervisorCache() (empty)

60 cache.add(1) (empty)

61 cache.add(2) (empty)

62 mocked_time.return_value = cache.ttl - 1 (empty)

63 assert 1 in cache (empty)

64 assert 2 in cache (empty)

65 assert 1 in cache (empty)

66 mocked_time.return_value = cache.ttl + 1 (empty)

67 assert 1 not in cache (empty)

68 assert 2 not in cache (empty)