1#!/usr/bin/env python3
2
3import atexit
4import signal
5import urllib
6import time
7import os
8import subprocess
9import socket
10from datetime import datetime
11from typing import Optional
12from typing import Union
13from urllib.error import URLError
14
15import pytest
16
17# Methods of dynamically skipping tests i.e. avoid web tests if not configured,
18# avoid database if not specified, avoid PUS tests not non-PUS projects
19
20# collect_ignore = ["setup.py"]
21
22# when starting a webserver automatically (via the --local option) search for a free port
23# in this range
24FIRST_PORT = 39000
25LAST_PORT = 39999
26
27LOCALHOST = '127.0.0.1'
28
29# Number of seconds to wait for the internal web server process to start up
30MAX_SERVER_START_TIME = 15
31
32def pytest_addoption(parser):
33 """Custom options for tests.
34
35 They are only applicable to tests that use the web server, but it looks simplest to add
36 them to all tests."""
37 # if pytest_addoption.dunit:
38 # return
39
40 group = parser.getgroup('project')
41 group.addoption('--project', '--launcher',
42 help='Name of project launcher cmd')
43 group.addoption(
44 '--db',
45 help=(
46 'Name the database connection to use for tests. If omitted then tests that need '
47 'a database connection will be skipped'))
48
49 group = parser.getgroup('server')
50 try:
51 group.addoption('--local',
52 action='store_true',
53 default=True,
54 help='Run local webserver (default)')
55 except ValueError:
56 # If we're run via invoke, the core conftest.py and this file can both try
57 # to add the same options and the thing doesn't allow that
58 pass
59 try:
60 group.addoption('--url',
61 help='Test against remote server')
62 except ValueError:
63 pass
64 try:
65 group.addoption('--env',
66 help='Test against OPE/VAL/TST/EPP/EPPVAL system')
67 except ValueError:
68 pass
69
70@pytest.fixture
71def project(request):
72 """Return project name if the user specified the "pytest --project charteps" etc. option."""
73 result = request.config.getoption('project')
74 # if result is not None:
75 # return result
76
77 # return launcher
78 # if launcher is not None
79 # this saves little bit of typing but is certain to cause weird problems one day
80 # if not result.startswith('chart') and not result.startswith('gsar'):
81 # result = 'chart' + result
82 return result
83
84# Running commands and algorithms
85@pytest.fixture
86def tool(request):
87 """Call a user tool."""
88 # maybe we could use the caplog, capfd, run builtin fixtures to improve functionality
89 # launcher_exe = request.config.getoption('launcher')
90 # if launcher_exe is None:
91 launcher_exe = request.config.getoption('project')
92
93 if launcher_exe is None:
94 pytest.skip('This test requires a project launcher specified (--project)')
95 return None
96
97 def run(cmd, expect_fail=False):
98 command = [launcher_exe] + cmd.split()
99 # print('running', command)
100 child = subprocess.Popen(command)
101 res = child.communicate()
102 retcode = child.returncode
103 print('exit code', retcode)
104 assert retcode == 0
105
106 # print('returning imp')
107 return run
108
109# output directories
110
111# look into testdir, tmp_path, tmp_path_factory, tmpdir, tmpdir factory
112# ideal functionality:
113# by default all tests go into a .pytest(?) directory of project root. Each run of each test
114# deletes and replaces any existing run. Allow a command line option to force a different directory
115
116# products
117
118# https://pypi.org/project/pytest-datadir, https://pypi.org/project/pytest-datafiles/
119
120# tox?
121
122@pytest.fixture
123def database(request):
124 """Specify a test requires a database.
125
126 Args:
127 `writeable`: The test requires write access to the database.
128 (`product`: List of products that should be present in the database)
129
130 `product` is not implemented. In future it will allow a list of products to be
131 specified which are required to be present in the database before continuing with the
132 test - this means they might be ingested by this function if missing, or we just
133 continue if they are present. This is to help write tests for both development and
134 operational databases.
135 """
136 def imp(writeable=False):
137 db_conn = request.config.getoption('db')
138 # print('user option db_conn', db_conn)
139 if db_conn is not None:
140 # print('requested db', db_conn)
141 from chart import settings
142 settings.set_db_name(db_conn)
143
144 import chart.db.connection
145 import chart.db.ts
146 # from chart import db
147 # from chart.db import connection
148 db = chart.db
149 db_conn = db.connection.db_connect()
150 params = db_conn.params
151 # print('db params', params)
152
153 if writeable:
154 # test requires a writeable database suitable for an automated test to mess around with
155 # The case only gets it's database if:
156 # - The user explicitly specified a database connection with "--db <name>" when
157 # invoking pytest
158 # - and that database connection exists and is marked as a test database in project settings
159 # - or the user also passed to "--testdb" option, which enables non-test databases
160 # to be treated as test databases, for example during development or if the user
161 # knows what they are doing and are running a destructive test against an operational
162 # database but it's a test that leaves the database in a good state afterwards
163 if params.get('TEST_DB') is not True:
164 return None
165
166 return db
167
168 return imp
169
170
171@pytest.fixture
172def product():
173 """Read a test data product."""
174 def imp(filename):
175 from chart import settings
176 return settings.PROJ_ROOT_DIR.joinpath('tests/application/products').joinpath(filename)
177
178 return imp
179
180
181@pytest.fixture(scope='session')
182def webserver(request):
183 """Spawn a webserver, finding a free socket to run on. Return the socket number.
184
185 Automatic shutdown. Test fails on any server error code or empty ."""
186 config_local = request.config.getoption('local')
187 config_url = request.config.getoption('url')
188 config_env = request.config.getoption('env')
189 # Inform the client whether we did start a local server
190 local = False
191 process = None
192 if config_url:
193 # this means we use an existing server
194 # print('webserver remote url', config_url)
195 url = config_url
196
197 if not url.endswith('/'):
198 url += '/'
199
200 elif config_env:
201 # pre-loaded URLs. This is probably only useful if we have a way to make them
202 # work nicely across projects with a load of extra configuration work
203 # print('webserver env', config_env)
204 url = {'OPE': 'http://chart/eps/',
205 'VAL': 'http://chart/val/eps/',
206 'TST': 'http://chart/tst/eps/'}[config_env.upper()]
207
208 else:
209 print('webserver spawn')
210 local = True
211 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
212 # sock.settimeout(1)
213 port = None
214 for p in range(FIRST_PORT, LAST_PORT):
215 print('trying port {p}'.format(p=p))
216 # res = sock.connect_ex((LOCALHOST, p))
217 try:
218 sock.bind((LOCALHOST, p))
219 sock.listen(1)
220 sock.close()
221 port = p
222 break
223
224 except socket.error as e:
225 pass
226 # print('sock port',p,'res',res)
227 # if res == 0:
228 # break
229
230 if not port:
231 raise ValueError('Cannot find free port to run local server')
232
233 import os
234 from chart.project import settings
235 # This is needed due to a very obscure error. Some other tests (such as ones that
236 # spawn algorithms) change pwd, potentially to temporary directories that have been
237 # deleted. Then, for very obscure reasons, the matplotlib library calls getcwd() for no
238 # apparent reason during startup, throwing an internal server error as pwd doesn't
239 # exist
240 os.chdir(settings.PROJECT_HOME_DIR)
241 from chart.common.shell import spawn
242 # launcher_exe = request.config.getoption('project')
243 # if lancher_exe is None:
244 # launcher_exe = launcher
245 launcher = request.config.getoption('project')
246
247 cmd = '{launcher} serve -p {port} --prefix test/'.format(port=p, launcher=launcher)
248 print('spawning', cmd)
249
250 # Spawn the subprocess but make sure it gets terminated cleanly after testing
251 process = spawn(cmd.split(' '), linger=True)
252
253 # We have to manually kill the process because Python won't clean this one
254 # up automatically
255 def ending():
256 """Terminate the webserver."""
257 process.kill()
258
259 atexit.register(ending)
260
261 # It takes a while to startup so we poll the server because we can't think
262 # of a better way
263 # print('donespawn')
264 # (we could scan stdout of the child process looking for an end of init message)
265 tries = 0
266 url = 'http://{host}:{port}/test/'.format(host=LOCALHOST, port=p)
267 while True:
268 print('try',url)
269 try:
270 res = urllib.request.urlopen(url)
271 break
272 except URLError as e:
273 # There must be a cleaner way to detect 500 errors
274 # and distinguish them from connection not ready yet errors
275 if 'Internal Server Error' in str(e):
276 print(dir(e))
277 print
278 raise e
279 print('Skip', e)
280 pass
281
282 except IOError as e:
283 print('ioerror', e)
284 pass
285
286 if tries > MAX_SERVER_START_TIME:
287 raise ValueError('Cannot start internal web server')
288
289 tries += 1
290 time.sleep(1)
291 # 1/0
292 # childp.kill()
293
294 return {'base_url': url,
295 'local': local}
296
297
298@pytest.fixture
299def timeseries():
300 """Check if data is present in a timeseries table and ingest if not.
301
302 Ingestion relies on generic ingestion being set up via activity files."""
303 def imp(sid: 'SID',
304 start: datetime,
305 stop: datetime,
306 table: ('TableInfo', str),
307 field: Optional[Union['FieldInfo', str]],
308 product: Optional['Path']=None,
309 force_ingest: bool=False,
310 min_rowcount: Optional[int]=None,
311 required_rowcount: Optional[int]=None,
312 ):
313 from chart.db import ts
314 from chart.common.traits import name_of_thing
315 from chart.cmd.ingest import generic_file_ingest
316
317 assert not (force_ingest and not product), \
318 'If force_ingest is specified then product must be given'
319
320 do_ingest = force_ingest
321 if not force_ingest:
322 rows = ts.count(sid=sid,
323 sensing_start=start,
324 sensing_stop=stop,
325 table=table,
326 fields=field)
327 logger.debug('Found {cc} rows in {tbl} min rows {mn}'.format(
328 cc=rows, tbl=name_of_thing(table), mn=min_rowcount))
329 do_ingest = True
330
331 # Either ingestion was forced by client, or there aren't enough rows in the database
332 if do_ingest:
333 # If the client specified a product, ingest it
334 if product is not None:
335 generic_file_ingest(product, sid=sid)
336
337 # otherwise the test fails
338 else:
339 raise ValueError('Database only has {act} rows ({req} required) and no product '
340 'supplied').format(act=rows, req=min_rowcount)
341
342 if required_rowcount is not None:
343 final_rows = ts.count(sid=sid,
344 sensing_start=start,
345 sensing_stop=stop,
346 table=table,
347 field=field)
348 assert final_rows == required_rowcount
349
350
351 return imp