Coverage for gws-app/gws/test/container_runner.py: 77%

1"""Test runner (container). 

2 

3Container test runner. Assumes tests are configured on the host with ``test/test.py configure``. 

4 

5""" 

6import os 

7 

8os.environ['GWS_IN_TEST'] = '1' 

9 

10import time 

11import re 

12import requests 

13 

14import pytest 

15 

16import gws 

17import gws.lib.cli as cli 

18import gws.lib.jsonx 

19import gws.lib.osx 

20import gws.spec.runtime 

21import gws.test.util as u 

22 

23USAGE = """ 

24GWS in-container test runner 

25~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 

26 

27 python3 runner.py <options> - <pytest options> 

28 

29Options: 

30 

31 -b, --base <path> - path to the base dir (see `runner.base_dir` in `test.ini`) 

32 -o, --only <regex> - only run filenames matching the pattern  

33 -v, --verbose - enable debug logging 

34  

35Pytest options: 

36 see https://docs.pytest.org/latest/reference.html#command-line-flags 

37 

38""" 

39 

40 

41def main(args): 

42 version = cli.read_file(gws.c.APP_DIR + '/VERSION') 

43 cli.info(f'GWS version {version}') 

44 if not os.path.isfile('/gws-app/.dockerenv'): 

45 cli.warning(f'GWS local application mounted') 

46 

47 gws.u.ensure_system_dirs() 

48 

49 base = args.get('base') or args.get('b') or gws.env.GWS_TEST_DIR 

50 u.OPTIONS = gws.lib.jsonx.from_path(f'{base}/config/OPTIONS.json') 

51 u.OPTIONS['BASE_DIR'] = base 

52 

53 u.ensure_dir(u.OPTIONS['BASE_DIR'] + '/tmp', clear=True) 

54 

55 if not gws.env.GWS_IN_CONTAINER: 

56 # if we are not in a container, use 'localhost:exposed_port' for all services 

57 for k, v in u.OPTIONS.items(): 

58 if k.endswith('.host'): 

59 u.OPTIONS[k] = 'localhost' 

60 if k.endswith('.port'): 

61 u.OPTIONS[k] = u.OPTIONS[k.replace('.port', '.expose_port')] 

62 

63 pytest_args = [ 

64 f'--config-file={base}/config/pytest.ini', 

65 f'--rootdir=/gws-app', 

66 f'--ignore-glob=__build', 

67 ] 

68 pytest_args.extend(args.get('_rest', [])) 

69 

70 if args.get('verbose') or args.get('v'): 

71 gws.log.set_level('DEBUG') 

72 pytest_args.append('--tb=native') 

73 pytest_args.append('-vv') 

74 else: 

75 gws.log.set_level('CRITICAL') 

76 

77 files = enum_files_for_test(args.get('only') or args.get('o')) 

78 if not files: 

79 cli.fatal(f'no files to test') 

80 return 

81 pytest_args.extend(files) 

82 

83 if not health_check(): 

84 cli.fatal('health check failed') 

85 return 

86 

87 cli.info('pytest ' + ' '.join(pytest_args)) 

88 return pytest.main(pytest_args, plugins=['gws.test.util']) 

89 

90 

91## 

92 

93 

94def enum_files_for_test(only_pattern): 

95 """Enumerate files to test, wrt --only option.""" 

96 

97 regex = u.OPTIONS.get('pytest.python_files').replace('*', '.*') 

98 

99 files = list(gws.lib.osx.find_files(f'{gws.c.APP_DIR}/gws', regex)) 

100 if only_pattern: 

101 files = [f for f in files if re.search(only_pattern, f)] 

102 

103 # sort files semantically 

104 

105 _sort_order = '/core/ /lib/ /gis/ /test/ /base/ /plugin/' 

106 

107 def _sort_key(path): 

108 for n, s in enumerate(_sort_order.split()): 

109 if s in path: 

110 return n, path 

111 return 99, path 

112 

113 files.sort(key=_sort_key) 

114 return files 

115 

116 

117## 

118 

119_HEALTH_CHECK_ATTEMPTS = 10 

120_HEALTH_CHECK_PAUSE = 3 

121 

122 

123def health_check(): 

124 status = {s: False for s in u.OPTIONS['runner.services']} 

125 

126 for _ in range(_HEALTH_CHECK_ATTEMPTS): 

127 for s, ok in status.items(): 

128 if not ok: 

129 fn = globals().get(f'health_check_service_{s}') 

130 err = fn() if fn else None 

131 if err: 

132 cli.warning(f'health_check: service {s!r}: waiting: {err}') 

133 else: 

134 cli.info(f'health_check: service {s!r}: ok') 

135 status[s] = not err 

136 if all(status.values()): 

137 return True 

138 time.sleep(_HEALTH_CHECK_PAUSE) 

139 

140 return False 

141 

142 

143def health_check_service_postgres(): 

144 try: 

145 r = u.pg.rows('select 1') 

146 u.pg.close() 

147 except Exception as exc: 

148 return repr(exc) 

149 

150 

151def health_check_service_qgis(): 

152 return http_ping(u.OPTIONS['service.qgis.host'], u.OPTIONS['service.qgis.port']) 

153 

154 

155def health_check_service_mockserver(): 

156 return http_ping(u.OPTIONS['service.mockserver.host'], u.OPTIONS['service.mockserver.port']) 

157 

158 

159def http_ping(host, port): 

160 url = f'http://{host}:{port}' 

161 try: 

162 res = requests.get(url) 

163 if res.status_code == 200: 

164 return 

165 return f'http status={res.status_code}' 

166 except Exception as exc: 

167 return repr(exc) 

168 

169 

170## 

171 

172 

173if __name__ == '__main__': 

174 cli.main('test', main, USAGE)