Coverage for gws-app/gws/test/team_engine/main.py: 0%
132 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-16 22:59 +0200
1"""
3https://opengeospatial.github.io/teamengine/users.html
4https://github.com/opengeospatial/teamengine
6"""
8import os
9import sys
10import re
11import base64
12import html
13import shutil
15LOCAL_APP_DIR = os.path.abspath(os.path.dirname(__file__) + '/../../..')
16sys.path.insert(0, LOCAL_APP_DIR)
18import gws
19import gws.lib.cli as cli
20import gws.lib.net
21import gws.lib.xmlx as xmlx
23USERNAME = "gwstest"
24PASSWORD = "gws"
26USER_DIR = os.path.dirname(__file__) + f'/te_base/users/{USERNAME}'
28USAGE = """
29GWS TeamEngine runner
30~~~~~~~~~~~~~~~~~~~~~
32Commands:
34 main.py ls
35 - show available test suites
37 main.py args <suite>
38 - show arguments for the given suite
40 main.py run <suite> <url>
41 - run a suite against the given URL
43Options:
44 --host - TeamEngine host (default: 127.0.0.1)
45 --port - TeamEngine port (default: 8090)
46 --level - list of error levels to report (default: 4,6) or 'all'
48 --save-xml <path> - save raw XML output
50The TE docker container must be started before running this script (see `./docker-compose.yml`).
52"""
54OPTIONS = {}
56# https://www.w3.org/TR/EARL10-Schema/#OutcomeValue + http://cite.opengeospatial.org/earl#inheritedFailure"
57# using CTL XML error levels from https://opengeospatial.github.io/teamengine/users.html
59ERROR_LEVELS = {
60 'passed': ('pass', 1),
61 'inapplicable': ('pass', 2),
62 'untested': ('warn', 3),
63 'cantTell': ('warn', 4),
64 'inheritedFailure': ('FAIL', 5),
65 'failed': ('FAIL', 6),
66}
68STATUS_MARKS = {
69 'FAIL': '\u274C',
70 'warn': '\u2754',
71 'pass': '\u2705',
72}
75def main(args):
76 OPTIONS['host'] = args.get('host', 'localhost')
77 OPTIONS['port'] = args.get('port', '8080')
78 OPTIONS['save-xml'] = args.get('save-xml', '')
80 s = args.get('level', '4,6')
81 if s == 'all':
82 OPTIONS['level'] = list(range(7))
83 else:
84 OPTIONS['level'] = [int(x) for x in gws.u.to_list(s)]
86 cmd = args.get(1)
88 if cmd == 'ls':
89 print(cli.text_table(_get_suites(), header='auto'))
90 return 0
92 if cmd == 'args':
93 suite = args.get(2)
94 print(cli.text_table(_get_args(suite)))
95 return 0
97 if cmd == 'run':
98 shutil.rmtree(f'{USER_DIR}/rest', ignore_errors=True)
99 OPTIONS['suite'] = args.get(2)
100 OPTIONS['url'] = args.get(3)
101 xml = _invoke_test()
102 results = _parse_test_results(xml)
103 _report_test(results)
104 return 0
107def _get_suites():
108 xml = xmlx.from_string(_invoke('suites'))
109 return sorted(
110 [el.textdict() for el in xml.findall('.//testSuite')],
111 key=str
112 )
115def _get_args(suite):
116 xml = xmlx.from_string(_invoke(f'suites/{suite}'))
117 return [el.textdict() for el in xml.findall('.//testrunargument')]
120URL_ARG = {
121 'wfs20': 'wfs',
122 'wms13': 'capabilities-url'
123}
126def _invoke_test():
127 if not OPTIONS['url'].startswith('http'):
128 OPTIONS['url'] = 'http://' + OPTIONS['url']
129 url = OPTIONS['url']
130 if '?' not in url:
131 url += '?request=GetCapabilities'
132 url = url.replace('&', '&')
134 suite = OPTIONS['suite']
135 text = _invoke(f'suites/{suite}/run?{URL_ARG[suite]}={url}', accept='application/rdf+xml')
136 if OPTIONS['save-xml']:
137 gws.u.write_file(OPTIONS['save-xml'], text)
139 return xmlx.from_string(text, remove_namespaces=True)
142def _parse_test_results(xml):
143 # parse test results in the EARL format (https://www.w3.org/TR/EARL10-Guide)
145 tc_map = {}
146 results = []
148 for case_el in xml.findall('.//TestCase'):
149 uid = case_el.get('about')
150 tc = dict(
151 testCase=uid,
152 title=_nows(case_el.textof('title')),
153 description=_nows(html.unescape(case_el.textof('description'))),
154 details='',
155 )
156 if re.match(r'^s\d+', uid):
157 # uids starting with sNNNN reference log files in the rest dir
158 path = USER_DIR + '/rest/' + uid.split('#')[0] + '/log.xml'
159 tc['details'] = _parse_ctl_log(path)
161 tc_map[uid] = tc
163 for ass_el in xml.findall('.//Assertion'):
164 test_el = ass_el.find('test')
166 # it's either an inline <test><TestCase about=ID> or a reference <test resource=ID>
167 case_el = test_el.find('TestCase')
168 tc = tc_map[case_el.get('about')] if case_el else tc_map[test_el.get('resource')]
170 res_el = ass_el.find('result/TestResult')
172 r = dict(tc)
173 # eg <earl:outcome rdf:resource="http://www.w3.org/ns/earl#passed"/>
174 r['status2'] = res_el.find('outcome').get('resource').split('#')[-1]
175 r['status'], r['level'] = ERROR_LEVELS[r['status2']]
177 # eg <earl:Assertion rdf:about="assert-7">
178 r['uid'] = int(re.sub(r'\D+', '', ass_el.get('about')))
180 s = res_el.textof('description')
181 if s and s != 'No details available.':
182 r['details'] = s + '\n' + r['details']
184 results.append(r)
186 return sorted(results, key=lambda r: r['uid'])
188def _parse_ctl_log(path):
189 if not os.path.exists(path):
190 return ''
192 xml = xmlx.from_path(path, remove_namespaces=True)
194 desc = [el.text for el in xml.findall('.//message')]
196 params = []
197 for el in xml.findall('.//param'):
198 if el.get('name'):
199 params.append(el.get('name', '') + '=' + (el.text or ''))
200 if params:
201 desc.append(OPTIONS['url'].split('?')[0] + '?' + '&'.join(params))
203 return '\n'.join(desc)
206def _report_test(results):
207 stats = f'TOTAL={len(results)}'
209 by_status = {}
210 for r in results:
211 by_status[r['status2']] = by_status.get(r['status2'], 0) + 1
212 for k, v in sorted(by_status.items()):
213 stats += f' {k}={v}'
215 cli.info(f'')
216 cli.info(stats)
217 cli.info(f'')
219 results = [r for r in results if r['level'] in OPTIONS['level']]
221 for r in results:
222 print(_nows(f"""
223 {STATUS_MARKS[r['status']]}
224 {r['status']}
225 ({r['status2']}, {r['level']}):
226 {r['title']}. {r['description']}
227 [{r['testCase']}]
228 """))
230 if r['details']:
231 for ln in r['details'].splitlines():
232 print(f' | {ln}')
234 print()
236 return
239def _invoke(path, **kwargs):
240 url = f'http://{OPTIONS["host"]}:{OPTIONS["port"]}/teamengine/rest/{path}'
241 headers = {
242 'Authorization': f'Basic ' + base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode(),
243 'Accept': kwargs.pop('accept', 'application/xml'),
244 }
245 cli.info(f'>> {url}')
246 try:
247 res = gws.lib.net.http_request(url, headers=headers, **kwargs)
248 if res.status_code != 200:
249 cli.error(res.text)
250 cli.fatal(f'HTTP ERROR {res.status_code}')
251 return res.text
252 except gws.lib.net.Error as exc:
253 cli.fatal(f'HTTP ERROR: {exc!r}')
256def _nows(s):
257 return re.sub(r'\s+', ' ', s.strip())
260if __name__ == '__main__':
261 cli.main('test', main, USAGE)