Coverage for tests / cli / supervisors / test_watchfiles.py: 90%

1import os (empty)

2import signal (empty)

3import time (empty)

4from multiprocessing.context import SpawnProcess (empty)

5from pathlib import Path (empty)

6from unittest.mock import MagicMock, patch (empty)

7 

8import pytest (empty)

9 

10from faststream._internal.cli.dto import RunArgs (empty)

11from faststream._internal.cli.supervisors.utils import get_subprocess (empty)

12from faststream._internal.cli.supervisors.watchfiles import WatchReloader (empty)

13from tests.cli import interfaces (empty)

14from tests.marks import skip_windows (empty)

15 

16DIR = Path(__file__).resolve().parent (empty)

17 

18 

19class PatchedWatchReloader(WatchReloader): (empty)

20 def start_process(self, worker_id: int | None = None) -> SpawnProcess: (empty)

21 process = get_subprocess(target=self._target, args=(self._args,)) (empty)

22 process.start() (empty)

23 return process (empty)

24 

25 

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

27@skip_windows (empty)

28def test_base(generate_template: interfaces.GenerateTemplateFactory) -> None: (empty)

29 with generate_template("") as file_path: (empty)

30 processor = PatchedWatchReloader( (empty)

31 target=exit, 

32 args=RunArgs(app=""), 

33 reload_dirs=[str(file_path.parent)], 

34 ) 

35 

36 processor._args.extra_options = {"parent_id": processor.pid} (empty)

37 processor.run() (empty)

38 

39 code = abs(processor._process.exitcode or 0) (empty)

40 

41 assert code in {signal.SIGTERM.value, 0}, code (empty)

42 

43 

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

45@skip_windows (empty)

46def test_restart( (empty)

47 mock: MagicMock, generate_template: interfaces.GenerateTemplateFactory 

48) -> None: 

49 with generate_template("") as file_path: (empty)

50 processor = PatchedWatchReloader( (empty)

51 target=touch_file, 

52 args=RunArgs(app=str(file_path)), 

53 reload_dirs=[file_path.parent], 

54 ) 

55 

56 mock.side_effect = lambda: exit( (empty)

57 RunArgs(app="", extra_options={"parent_id": processor.pid}) 

58 ) 

59 

60 with patch.object(processor, "restart", mock): (empty)

61 processor.run() (empty)

62 

63 mock.assert_called_once() (empty)

64 

65 

66def touch_file(args: RunArgs) -> None: (empty)

67 while True: 

68 time.sleep(0.1) 

69 Path(args.app).write_text("hello", encoding="utf-8") 

70 

71 

72def exit(args: RunArgs) -> None: (empty)

73 os.kill(int(args.extra_options["parent_id"]), signal.SIGINT) (empty)