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 -v, --verbose - enable debug logging 

32  

33Pytest options: 

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

35 

36""" 

37 

38 

39def main(args): 

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

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

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

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

44 

45 gws.u.ensure_system_dirs() 

46 

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

48 u.load_options(base) 

49 

50 pytest_args = [ 

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

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

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

54 ] 

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

56 

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

58 gws.log.set_level('DEBUG') 

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

60 pytest_args.append('-vv') 

61 else: 

62 gws.log.set_level('CRITICAL') 

63 

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

65 if not files: 

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

67 return 

68 pytest_args.extend(files) 

69 

70 if not health_check(): 

71 cli.fatal('health check failed') 

72 return 

73 

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

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

76 

77 

78## 

79 

80 

81def enum_files_for_test(only_pattern): 

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

83 

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

85 

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

87 if only_pattern: 

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

89 

90 # sort files semantically 

91 

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

93 

94 def _sort_key(path): 

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

96 if s in path: 

97 return n, path 

98 return 99, path 

99 

100 files.sort(key=_sort_key) 

101 return files 

102 

103 

104## 

105 

106_HEALTH_CHECK_ATTEMPTS = 10 

107_HEALTH_CHECK_PAUSE = 3 

108 

109 

110def health_check(): 

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

112 

113 for _ in range(_HEALTH_CHECK_ATTEMPTS): 

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

115 if not ok: 

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

117 err = fn() if fn else None 

118 if err: 

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

120 else: 

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

122 status[s] = not err 

123 if all(status.values()): 

124 return True 

125 time.sleep(_HEALTH_CHECK_PAUSE) 

126 

127 return False 

128 

129 

130def health_check_service_postgres(): 

131 try: 

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

133 except Exception as exc: 

134 return repr(exc) 

135 

136 

137def health_check_service_qgis(): 

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

139 

140 

141def health_check_service_mockserver(): 

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

143 

144 

145def http_ping(host, port): 

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

147 try: 

148 res = requests.get(url) 

149 if res.status_code == 200: 

150 return 

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

152 except Exception as exc: 

153 return repr(exc) 

154 

155 

156## 

157 

158 

159if __name__ == '__main__': 

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