Coverage for quibble/backend.py: 61%

4# Licensed under the Apache License, Version 2.0 (the "License");

5# you may not use this file except in compliance with the License.

6# You may obtain a copy of the License at

8# http://www.apache.org/licenses/LICENSE-2.0

10# Unless required by applicable law or agreed to in writing, software

11# distributed under the License is distributed on an "AS IS" BASIS,

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

13# See the License for the specific language governing permissions and

14# limitations under the License.

16import json

17import logging

18import os

19import pwd

20import signal

21import socket

22import subprocess

23import tempfile

24import threading

25import time

26import urllib

28import quibble

30backend_registry = {}

33def _tcp_wait(host, port, timeout=3):

34 step = 0

35 delay = 0.1 # seconds

36 socket_timeout = 1 # seconds

37 connected = False

38 while step < timeout:

39 try:

40 s = socket.socket()

41 s.settimeout(socket_timeout)

42 s.connect((host, int(port)))

43 connected = True

44 break

45 except (ConnectionAbortedError, ConnectionRefusedError):

46 step = step + delay

47 time.sleep(delay)

48 finally:

49 s.close()

51 if not connected:

52 raise TimeoutError(

53 'Could not connect to port %s after %s seconds' % (port, timeout)

54 )

57def backend(interface, key):

58 """Register a backend class by name, for a given interface class."""

60 def _register_backend(backend_class):

61 if not issubclass(backend_class, interface): 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 raise Exception(

63 'Registered backend "%s" does not extend %s'

64 % (backend_class, interface)

65 )

67 interface_name = str(interface)

68 if interface_name not in backend_registry:

69 backend_registry[interface_name] = {}

70 backend_registry[interface_name][key] = backend_class

72 return backend_class

74 return _register_backend

77def get_backend(interface, key):

78 key = key.lower()

79 interface = str(interface)

80 if key in backend_registry[interface]:

81 return backend_registry[interface][key]

83 raise Exception('Backend %s not supported: %s' % (interface, key))

86def web_backend(key):

87 return backend(WebserverEngine, key)

90def db_backend(key):

91 return backend(DatabaseServer, key)

94def getDatabase(engine, db_dir, dump_dir, log_dir):

95 '''Set up a database backend, without starting it.'''

96 dbclass = get_backend(DatabaseServer, engine)

97 db = dbclass(base_dir=db_dir, dump_dir=dump_dir, log_dir=log_dir)

98 db.type = engine

99 return db

100

101

102def getWebserver(engine, mw_install_path, web_url, kwargs={}):

103 webclass = get_backend(WebserverEngine, engine)

104 backend = webclass(mwdir=mw_install_path, url=web_url, **kwargs)

105 return backend

106

107

108def _stream_relay(process, stream, log_function):

109 thread = threading.Thread(

110 target=_stream_to_log, args=(process, stream, log_function)

111 )

112 thread.start()

113 return thread

114

115

116def _stream_to_log(process, stream, log_function):

117 while True:

118 line = stream.readline()

119 if not line:

120 break

121 log_function(line.rstrip())

122

123

124class BackendServer:

125 server = None

126

127 def __init__(self):

128 self.log = logging.getLogger('backend.%s' % self.__class__.__name__)

129

130 def __enter__(self):

131 self.start()

132

133 def __exit__(self, *args):

134 self.stop()

135

136 def __getstate__(self):

137 state = self.__dict__.copy()

138 if "server" in state:

139 del state["server"]

140 return state

141

142 def start(self):

143 pass

144

145 def stop(self):

146 if self.server is not None:

147 self.log.info('Terminating %s', self.__class__.__name__)

148 self.server.terminate()

149 try:

150 self.server.wait(2)

151 except subprocess.TimeoutExpired:

152 self.server.kill() # SIGKILL

153 finally:

154 self.server = None

155

156

157class DatabaseServer(BackendServer):

158 dump_dir = None

159 log_dir = None

160

161 def __init__(self, base_dir=None, dump_dir=None, log_dir=None):

162 super(DatabaseServer, self).__init__()

163 self.base_dir = base_dir

164 self.dump_dir = dump_dir

165 self.log_dir = log_dir

166

167 def _init_rootdir(self, base_dir):

168 # Create a temporary data directory

169 prefix = 'quibble-%s-' % self.__class__.__name__.lower()

170

171 if base_dir is not None: 171 ↛ 176line 171 didn't jump to line 176 because the condition on line 171 was always true

172 base_dir = os.path.abspath(base_dir)

173 os.makedirs(base_dir, exist_ok=True)

174

175 # Create and hold a reference

176 self._tmpdir = tempfile.TemporaryDirectory(dir=base_dir, prefix=prefix)

177 self.rootdir = self._tmpdir.name

178 self.log.debug('Root dir: %s', self.rootdir)

179

180 def start(self):

181 self._init_rootdir(self.base_dir)

182

183 def stop(self):

184 if self.dump_dir:

185 self.dump()

186 super(DatabaseServer, self).stop()

187

188 def dump(self):

189 self.log.warning(

190 '%s does not support dumping database', self.__class__.__name__

191 )

192

193 def __str__(self):

194 return "<%s>" % self.__class__.__name__

195

196

197@db_backend('postgres')

198class Postgres(DatabaseServer):

199 def __init__(self, base_dir=None, dump_dir=None, log_dir=None):

200 super(Postgres, self).__init__(base_dir, dump_dir, log_dir)

201

202 def start(self):

203 super(Postgres, self).start()

204

205 self.conffile = os.path.join(self.rootdir, 'conf')

206 self.socket = os.path.join(self.rootdir, 'socket')

207

208 # Start pg_virtualenv with TMPDIR set to rootdir so its working

209 # directory (and the Postgres data dir inside it) lives under

210 # --db-dir.

211 # https://github.com/credativ/postgresql-common/blob/master/pg_virtualenv

212 self.server = subprocess.Popen(

213 [

214 # fmt: off

215 'pg_virtualenv',

216 # Option for pg_createcluster

217 '-c',

218 '--socketdir=%s' % self.socket,

219 'python3',

220 '-m', 'quibble.pg_virtualenv_hook'

221 # fmt: on

222 ],

223 env={

224 'QUIBBLE_TMPFILE': self.conffile,

225 'TMPDIR': self.rootdir,

226 },

227 )

228

229 while not os.path.exists(self.conffile):

230 if self.server.poll() is not None:

231 raise Exception(

232 'Postgres failed during startup (%s)'

233 % self.server.returncode

234 )

235 time.sleep(1)

236

237 with open(self.conffile) as f:

238 conf = json.load(f)

239

240 self.user = conf['PGUSER']

241 self.password = conf['PGPASSWORD']

242 self.dbname = conf['PGDATABASE']

243 self.dbserver = self.socket

244 self.hook_pid = conf['PID']

245 self.log.info('Postgres is ready')

246

247 def stop(self):

248 # Send a signal to the hook since it's waiting on one

249 os.kill(self.hook_pid, signal.SIGUSR1)

250 super(Postgres, self).stop()

251

252

253@db_backend('mysql')

254class MySQL(DatabaseServer):

255 def __init__(

256 self,

257 base_dir=None,

258 dump_dir=None,

259 log_dir=None,

260 user='wikiuser',

261 password='secret',

262 dbname='wikidb',

263 dbserver='localhost',

264 ):

265 super(MySQL, self).__init__(base_dir, dump_dir, log_dir)

266

267 self.user = user

268 self.password = password

269 self.dbname = dbname

270 self.socket = None

271 self.dbserver = dbserver

272 self.log_dir = log_dir

273

274 def _install_db(self):

275 self.log.info('Initializing MySQL data directory')

276 p = subprocess.Popen(

277 [

278 'mysql_install_db',

279 # Legacy system with a passwordless root user

280 '--auth-root-authentication-method=normal',

281 '--datadir=%s' % self.rootdir,

282 '--user=%s' % pwd.getpwuid(os.getuid())[0],

283 ],

284 text=True,

285 stdout=subprocess.PIPE,

286 stderr=subprocess.STDOUT,

287 )

288 outs, errs = p.communicate()

289 if p.returncode != 0:

290 raise Exception("FAILED (%s): %s" % (p.returncode, outs))

291

292 def _createwikidb(self):

293 """Create a database and necessary grants.

294 Will drop existing database if it already exists."""

295 self.log.info('Creating the wiki database and grant')

296 mysql_cmd = ['mysql', '--user=root']

297 if self.socket: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true

298 mysql_cmd.append('--socket=%s' % self.socket)

299 p = subprocess.Popen(

300 mysql_cmd,

301 text=True,

302 stdin=subprocess.PIPE,

303 stdout=subprocess.PIPE,

304 stderr=subprocess.STDOUT,

305 )

306 grant = (

307 "DROP DATABASE IF EXISTS %s;"

308 "CREATE DATABASE %s;"

309 "GRANT ALL ON %s.* TO '%s'@'localhost'"

310 "IDENTIFIED BY '%s';\n"

311 % (self.dbname, self.dbname, self.dbname, self.user, self.password)

312 )

313 outs, errs = p.communicate(input=grant)

314 if p.returncode != 0: 314 ↛ exitline 314 didn't return from function '_createwikidb' because the condition on line 314 was always true

315 raise Exception("FAILED (%s): %s" % (p.returncode, outs))

316

317 def start(self):

318 self.log.info('Starting MySQL')

319 super(MySQL, self).start()

320

321 if self.log_dir is None:

322 self.errorlog = os.path.join(self.rootdir, 'error.log')

323 else:

324 self.errorlog = os.path.join(self.log_dir, 'mysql-error.log')

325

326 self.pidfile = os.path.join(self.rootdir, 'mysqld.pid')

327 self.socket = os.path.join(self.rootdir, 'socket')

328 self.dbserver = 'localhost:' + self.socket

329

330 self._install_db()

331

332 self.server = subprocess.Popen(

333 [

334 '/usr/sbin/mysqld', # fixme drop path

335 '--skip-networking',

336 '--innodb-print-all-deadlocks',

337 '--datadir=%s' % self.rootdir,

338 '--log-error=%s' % self.errorlog,

339 '--pid-file=%s' % self.pidfile,

340 '--socket=%s' % self.socket,

341 ],

342 stdout=subprocess.DEVNULL,

343 stderr=subprocess.DEVNULL,

344 )

345

346 while not os.path.exists(self.socket):

347 if self.server.poll() is not None:

348 with open(self.errorlog) as errlog:

349 print(errlog.read())

350 raise Exception(

351 "MySQL died during startup (%s)" % self.server.returncode

352 )

353 self.log.info("Waiting for MySQL socket")

354 time.sleep(1)

355

356 self._createwikidb()

357 self.log.info('MySQL is ready')

358

359 def dump(self):

360 dumpfile = os.path.join(self.dump_dir, 'mysqldump.sql')

361 self.log.info('Dumping database to %s', dumpfile)

362

363 mysqldump = open(dumpfile, 'wb')

364 subprocess.Popen(

365 [

366 'mysqldump',

367 '--socket=%s' % self.socket,

368 '--user=root',

369 '--all-databases',

370 ],

371 stdin=subprocess.PIPE,

372 stdout=mysqldump,

373 stderr=subprocess.STDOUT,

374 ).wait()

375

376 def __str__(self):

377 return "<{} {}>".format(

378 self.__class__.__name__,

379 self.socket if self.socket else "(no socket)",

380 )

381

382

383@db_backend('sqlite')

384class SQLite(DatabaseServer):

385 def __init__(

386 self, base_dir=None, dump_dir=None, log_dir=None, dbname='wikidb'

387 ):

388 super(SQLite, self).__init__(base_dir, dump_dir, log_dir)

389

390 self.dbname = dbname

391

392

393class ChromeWebDriver(BackendServer):

394 def __init__(self, display=None, port=4444, url_base='/wd/hub'):

395 super(ChromeWebDriver, self).__init__()

396

397 self.display = display

398 self.port = port

399 self.url_base = url_base

400

401 def start(self):

402 self.log.info('Starting Chromedriver')

403 try:

404 prev_display = os.environ.get('DISPLAY', None)

405 if self.display:

406 # We need DISPLAY in the env for chromium_flags()

407 os.environ.update({'DISPLAY': self.display})

408 env = {

409 'CHROMIUM_FLAGS': quibble.chromium_flags(),

410 'PATH': os.environ.get('PATH'),

411 }

412

413 if self.display is not None:

414 # Pass it to chromedriver

415 env.update({'DISPLAY': self.display})

416

417 self.server = subprocess.Popen(

418 [

419 'chromedriver',

420 '--port=%s' % self.port,

421 '--url-base=%s' % self.url_base,

422 ],

423 env=env,

424 text=True,

425 bufsize=1, # line buffered

426 stdout=subprocess.DEVNULL,

427 stderr=subprocess.PIPE,

428 )

429 _stream_relay(self.server, self.server.stderr, self.log.warning)

430

431 finally:

432 if prev_display:

433 os.environ.update({'DISPLAY': prev_display})

434 elif prev_display is None and self.display:

435 del os.environ['DISPLAY']

436

437 def __str__(self):

438 return "<ChromeWebDriver {}>".format(self.display)

439

440

441class WebserverEngine(BackendServer):

442 default_url = None

443

444 def __init__(self, url=None, mwdir=None):

445 super(WebserverEngine, self).__init__()

446

447 self.url = url or self.default_url

448 self.mwdir = mwdir

449

450 parsed_url = urllib.parse.urlparse(self.url)

451 self.host = parsed_url.hostname

452 self.port = parsed_url.port

453

454 def start(self):

455 if self.server: 455 ↛ 458line 455 didn't jump to line 458 because the condition on line 455 was always true

456 _stream_relay(self.server, self.server.stderr, self.log.info)

457

458 if self.host and self.port: 458 ↛ 459line 458 didn't jump to line 459 because the condition on line 458 was never true

459 _tcp_wait(host=self.host, port=self.port, timeout=5)

460

461

462@web_backend('external')

463class ExternalWebserver(WebserverEngine):

464 def start(self):

465 self.log.info('Not starting a webserver.')

466

467 def __str__(self):

468 return '<ExternalWebserver %s %s>' % (self.url, self.mwdir)

469

470

471@web_backend('php')

472class PhpWebserver(WebserverEngine):

473 default_url = 'http://127.0.0.1:9412'

474

475 def __init__(self, workers=False, **kwargs):

476 self.workers = workers

477

478 super(PhpWebserver, self).__init__(**kwargs)

479

480 def start(self):

481 server_cmd = [

482 # fmt: off

483 'php',

484 '-d', 'output_buffering=Off',

485 '-S', '%s:%s' % (self.host, self.port),

486 # fmt: on

487 ]

488

489 server_env = {}

490 if self.workers:

491 server_env = {

492 'PHP_CLI_SERVER_WORKERS': str(self.workers),

493 }

494 server_env.update(os.environ)

495

496 self.server = subprocess.Popen(

497 server_cmd,

498 cwd=self.mwdir,

499 text=True,

500 bufsize=1, # line buffered

501 stdout=subprocess.DEVNULL,

502 stderr=subprocess.PIPE,

503 env=server_env,

504 )

505 super(PhpWebserver, self).start()

506

507 def __str__(self):

508 if not self.workers: 508 ↛ 511line 508 didn't jump to line 511 because the condition on line 508 was always true

509 return '<PhpWebserver %s %s>' % (self.url, self.mwdir)

510 else:

511 return '<PhpWebserver %s %s with %s workers>' % (

512 self.url,

513 self.mwdir,

514 self.workers,

515 )

516

517

518class Xvfb(BackendServer):

519 def __init__(self, display=':94'):

520 super(Xvfb, self).__init__()

521 self.display = display

522

523 def start(self):

524 self.log.info('Starting Xvfb on display %s', self.display)

525 self.server = subprocess.Popen(

526 [

527 # fmt: off

528 'Xvfb', self.display,

529 '-screen', '0', '1280x1024x24',

530 '-nolisten', 'tcp',

531 '-nolisten', 'unix',

532 # fmt: on

533 ]

534 )

535

536 def __str__(self):

537 return "<Xvfb {}>".format(self.display)

538

539

540class Memcached(BackendServer):

541 def __init__(self, port=11211):

542 super(Memcached, self).__init__()

543 self.port = port

544

545 def start(self):

546 self.log.info('Starting Memcached on port %s', self.port)

547 self.server = subprocess.Popen(

548 [

549 # fmt: off

550 'memcached',

551 '--memory-limit=64', # MBytes

552 '--port=%s' % self.port,

553 '--listen=127.0.0.1',

554 # fmt:on

555 ]

556 )

557 # Note if an externally managed memcached server is already running,

558 # the managed process will exit 71:

559 #

560 # failed to listen on one of interface(s) 127.0.0.1:

561 # Address already in use

562 #

563 # We do not handle that, as long as we have a Memcached to interact

564 # with, that is good enough.

565

566 self.log.info('Waiting for Memcached on port %s', self.port)

567 _tcp_wait(host='127.0.0.1', port=self.port, timeout=2)

568

569 def __str__(self):

570 return "<Memcached on port {}>".format(self.port)

571

572

573class OpenSearch(BackendServer):

574 def __init__(self):

575 super(OpenSearch, self).__init__()

576 self.port = 9200

577

578 def start(self):

579 self.log.info('Waiting for OpenSearch on port %s', self.port)

580 _tcp_wait(host='127.0.0.1', port=self.port, timeout=120)

581

582 def __str__(self):

583 return "<OpenSearch on port {}>".format(self.port)