Coverage for tests/test_provider.py: 100%

1from contextlib import AsyncExitStack, ExitStack 20 ctx1abcdefghijklmnopqrst

2 

3import pytest 20 ctx1abcdefghijklmnopqrst

4 

5from fast_depends import Depends, Provider 20 ctx1abcdefghijklmnopqrst

6from fast_depends.core import build_call_model 20 ctx1abcdefghijklmnopqrst

7 

8 

9def base_dep() -> int: 20 ctx1abcdefghijklmnopqrst

10 return 1 20 ctx1abcdefghijklmnopqrst

11 

12 

13def override_dep() -> int: 20 ctx1abcdefghijklmnopqrst

14 return 2 20 ctx1abcdefghijklmnopqrst

15 

16 

17def sync_func(d: int = Depends(base_dep)) -> int: 20 ctx1abcdefghijklmnopqrst

18 return d 20 ctx1abcdefghijklmnopqrst

19 

20 

21async def async_func(d: int = Depends(base_dep)) -> int: 20 ctx1abcdefghijklmnopqrst

22 return d 20 ctx1abcdefghijklmnopqrst

23 

24 

25def call_level_provider() -> Provider: 20 ctx1abcdefghijklmnopqrst

26 provider = Provider() 20 ctx1abcdefghijklmnopqrst

27 provider.override(base_dep, override_dep) 20 ctx1abcdefghijklmnopqrst

28 return provider 20 ctx1abcdefghijklmnopqrst

29 

30 

31def test_merge_keeps_both_sides() -> None: 20 ctx1abcdefghijklmnopqrst

32 original, extra = Provider(), call_level_provider() 20 ctx1abcdefghijklmnopqrst

33 original.add_dependant(build_call_model(sync_func, dependency_provider=original)) 20 ctx1abcdefghijklmnopqrst

34 

35 merged = original.merge(extra) 20 ctx1abcdefghijklmnopqrst

36 

37 assert merged is not original 20 ctx1abcdefghijklmnopqrst

38 assert merged is not extra 20 ctx1abcdefghijklmnopqrst

39 assert merged.dependencies == original.dependencies | extra.dependencies 20 ctx1abcdefghijklmnopqrst

40 assert merged.overrides == original.overrides | extra.overrides 20 ctx1abcdefghijklmnopqrst

41 # merging must not mutate either side 

42 assert not original.overrides 20 ctx1abcdefghijklmnopqrst

43 

44 

45def test_sync_call_level_provider() -> None: 20 ctx1abcdefghijklmnopqrst

46 provider = Provider() 20 ctx1abcdefghijklmnopqrst

47 model = build_call_model(sync_func, dependency_provider=provider) 20 ctx1abcdefghijklmnopqrst

48 

49 with ExitStack() as stack: 20 ctx1abcdefghijklmnopqrst

50 assert model.solve(stack=stack, cache_dependencies={}) == 1 20 ctx1abcdefghijklmnopqrst

51 

52 with ExitStack() as stack: 20 ctx1abcdefghijklmnopqrst

53 assert ( 20 ctx1abcdefghijklmnopqrst

54 model.solve( 

55 stack=stack, 

56 cache_dependencies={}, 

57 dependency_provider=call_level_provider(), 

58 ) 

59 == 2 

60 ) 

61 

62 # the call-level provider must not leak into the model's own provider 

63 with ExitStack() as stack: 20 ctx1abcdefghijklmnopqrst

64 assert model.solve(stack=stack, cache_dependencies={}) == 1 20 ctx1abcdefghijklmnopqrst

65 

66 

67@pytest.mark.anyio 20 ctx1abcdefghijklmnopqrst

68async def test_async_call_level_provider() -> None: 20 ctx1abcdefghijklmnopqrst

69 provider = Provider() 20 ctx1abcdefghijklmnopqrst

70 model = build_call_model(async_func, dependency_provider=provider) 20 ctx1abcdefghijklmnopqrst

71 

72 async with AsyncExitStack() as stack: 20 ctx1abcdefghijklmnopqrst

73 assert await model.asolve(stack=stack, cache_dependencies={}) == 1 20 ctx1abcdefghijklmnopqrst

74 

75 async with AsyncExitStack() as stack: 20 ctx1abcdefghijklmnopqrst

76 assert ( 20 ctx1abcdefghijklmnopqrst

77 await model.asolve( 

78 stack=stack, 

79 cache_dependencies={}, 

80 dependency_provider=call_level_provider(), 

81 ) 

82 == 2 

83 ) 

84 

85 async with AsyncExitStack() as stack: 20 ctx1abcdefghijklmnopqrst

86 assert await model.asolve(stack=stack, cache_dependencies={}) == 1 20 ctx1abcdefghijklmnopqrst