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

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.osx 

19import gws.test.util as u 

20 

21USAGE = """ 

22GWS in-container test runner 

23~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 

24 

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

26 

27Options: 

28 

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

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

31 -k, --keyword <expr> - only run tests matching the pytest expression 

32 -v, --verbose - enable debug logging 

33  

34Pytest options: 

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

36 

37""" 

38 

39 

40def main(args): 

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

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

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

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

45 

46 gws.u.ensure_system_dirs() 

47 

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

49 u.load_options(base) 

50 

51 pytest_args = [ 

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

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

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

55 ] 

56 kw = args.get('keyword') or args.get('k') 

57 if kw: 

58 pytest_args.append(f'-k {kw}') 

59 

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

61 

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

63 gws.log.set_level('DEBUG') 

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

65 pytest_args.append('-vv') 

66 else: 

67 gws.log.set_level('CRITICAL') 

68 

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

70 if not files: 

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

72 return 

73 pytest_args.extend(files) 

74 

75 if not health_check(): 

76 cli.fatal('health check failed') 

77 return 

78 

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

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

81 

82 

83## 

84 

85 

86def enum_files_for_test(only_pattern): 

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

88 

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

90 

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

92 if only_pattern: 

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

94 

95 # sort files semantically 

96 

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

98 

99 def _sort_key(path): 

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

101 if s in path: 

102 return n, path 

103 return 99, path 

104 

105 files.sort(key=_sort_key) 

106 return files 

107 

108 

109## 

110 

111_HEALTH_CHECK_ATTEMPTS = 10 

112_HEALTH_CHECK_PAUSE = 3 

113 

114 

115def health_check(): 

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

117 

118 for _ in range(_HEALTH_CHECK_ATTEMPTS): 

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

120 if not ok: 

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

122 err = fn() if fn else None 

123 if err: 

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

125 else: 

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

127 status[s] = not err 

128 if all(status.values()): 

129 return True 

130 time.sleep(_HEALTH_CHECK_PAUSE) 

131 

132 return False 

133 

134 

135def health_check_service_postgres(): 

136 try: 

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

138 except Exception as exc: 

139 return repr(exc) 

140 

141 

142def health_check_service_qgis(): 

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

144 

145 

146def health_check_service_mockserver(): 

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

148 

149 

150def http_ping(host, port): 

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

152 try: 

153 res = requests.get(url) 

154 if res.status_code == 200: 

155 return 

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

157 except Exception as exc: 

158 return repr(exc) 

159 

160 

161## 

162 

163 

164if __name__ == '__main__': 

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