Coverage for gws-app/gws/lib/watcher/_test.py: 100%

1import gws 

2import gws.test.util as u 

3import gws.lib.watcher as watcher 

4 

5 

6def _mkdir(p): 

7 p.mkdir(parents=True) 

8 return p 

9 

10 

11def _touch(p): 

12 p.touch() 

13 gws.u.sleep(0.05) 

14 return p 

15 

16 

17def _change(p): 

18 p.write_text("test") 

19 gws.u.sleep(0.05) 

20 return p 

21 

22 

23@u.fixture 

24def cb(): 

25 class Callback: 

26 def __init__(self): 

27 self.changes = [] 

28 

29 def __call__(self, *args): 

30 self.changes.append(args) 

31 

32 def changed(self): 

33 yes = bool(self.changes) 

34 self.changes = [] 

35 return yes 

36 

37 return Callback() 

38 

39 

40def test_simple(tmp_path, cb): 

41 d1 = _mkdir(tmp_path / '1') 

42 d2 = _mkdir(tmp_path / '2') 

43 

44 w = watcher.new(cb) 

45 w.add_directory(d1) 

46 w.add_directory(d2) 

47 

48 w.start() 

49 

50 _touch(d1 / 'a') 

51 assert cb.changed() 

52 

53 _touch(d2 / 'b') 

54 assert cb.changed() 

55 

56 w.stop() 

57 

58 

59def test_patterns(tmp_path, cb): 

60 d = _mkdir(tmp_path / '1') 

61 

62 w = watcher.new(cb) 

63 w.add_directory(d, file_pattern=r'\.txt$') 

64 

65 w.start() 

66 

67 _touch(d / 'a.txt') 

68 assert cb.changed() 

69 

70 _touch(d / 'a.foo') 

71 assert not cb.changed() 

72 

73 w.stop() 

74 

75 

76def test_exclude(tmp_path, cb): 

77 d = _mkdir(tmp_path / '1') 

78 

79 w = watcher.new(cb) 

80 w.add_directory(d) 

81 w.exclude('ignore_me') 

82 

83 w.start() 

84 

85 a = _touch(d / 'not_ignored') 

86 assert cb.changed() 

87 _change(a) 

88 assert cb.changed() 

89 

90 a = _touch(d / 'ignore_me') 

91 assert not cb.changed() 

92 _change(a) 

93 assert not cb.changed() 

94 

95 

96def test_files(tmp_path, cb): 

97 d = _mkdir(tmp_path / '1') 

98 

99 w = watcher.new(cb) 

100 w.add_file(d / 'watch_me_1') 

101 w.add_file(d / 'watch_me_2') 

102 

103 w.start() 

104 

105 a = _touch(d / 'watch_me_1') 

106 assert cb.changed() 

107 _change(a) 

108 assert cb.changed() 

109 

110 a = _touch(d / 'watch_me_2') 

111 assert cb.changed() 

112 _change(a) 

113 assert cb.changed() 

114 

115 a = _touch(d / 'ignore_me') 

116 assert not cb.changed() 

117 _change(a) 

118 assert not cb.changed() 

119 

120 w.stop() 

121 

122 

123def test_recursive(tmp_path, cb): 

124 a = _mkdir(tmp_path / 'aaa') 

125 deep = _mkdir(a / 'bbb' / 'ccc') 

126 

127 w = watcher.new(cb) 

128 w.add_directory(a, file_pattern=r'\.txt$', recursive=True) 

129 

130 w.start() 

131 

132 _touch(deep / 'a.txt') 

133 assert cb.changed() 

134 

135 _touch(deep / 'a.foo') 

136 assert not cb.changed() 

137 

138 w.stop()