You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
4.7 KiB
100 lines
4.7 KiB
import io
|
|
import runpy
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from tests import test_auxiliary_tools
|
|
|
|
|
|
class ReportSQLConfigTests(unittest.TestCase):
|
|
def setUp(self):
|
|
test_auxiliary_tools.AuxiliaryToolTests()._install_mysql_stub()
|
|
import report_sql
|
|
self.parse = report_sql.parse_args
|
|
self.env = {
|
|
'REPORT_SQL_DB_HOST': 'database',
|
|
'REPORT_SQL_DB_USER': 'exporter',
|
|
'REPORT_SQL_DB_PASSWORD': 'space % $ quote" ; secret',
|
|
'REPORT_SQL_DB_NAME': 'logs',
|
|
}
|
|
|
|
def test_environment_defaults_and_literal_password(self):
|
|
args = self.parse(['--from-env'], self.env)
|
|
self.assertEqual((args.host, args.port), ('127.0.0.1', 4321))
|
|
self.assertEqual(args.db_host, 'database')
|
|
self.assertEqual(args.db_port, 3306)
|
|
self.assertEqual(args.db_user, 'exporter')
|
|
self.assertEqual(args.db_password, self.env['REPORT_SQL_DB_PASSWORD'])
|
|
self.assertEqual(args.db_name, 'logs')
|
|
self.assertEqual((args.relay_interface, args.relay_port), ('127.0.0.1', 0))
|
|
|
|
def test_environment_ports_and_relay(self):
|
|
self.env.update(REPORT_SQL_HOST='reports', REPORT_SQL_PORT='4400',
|
|
REPORT_SQL_DB_PORT='3307', REPORT_SQL_RELAY_PORT='4330',
|
|
REPORT_SQL_RELAY_INTERFACE='0.0.0.0')
|
|
args = self.parse(['--from-env'], self.env)
|
|
self.assertEqual((args.host, args.port), ('reports', 4400))
|
|
self.assertEqual(args.db_port, 3307)
|
|
self.assertEqual((args.relay_interface, args.relay_port), ('0.0.0.0', 4330))
|
|
|
|
def test_legacy_cli_does_not_read_environment(self):
|
|
args = self.parse(['reports', '4321', 'db', 'user', 'password', 'logs',
|
|
'--relay-port', '4330'], self.env)
|
|
self.assertEqual(args.db_password, 'password')
|
|
self.assertEqual(args.db_host, 'db')
|
|
self.assertEqual(args.relay_port, 4330)
|
|
self.assertEqual(args.db_port, 3306)
|
|
|
|
def test_missing_or_invalid_environment_fails_without_exposing_password(self):
|
|
for name, value in (
|
|
('REPORT_SQL_DB_HOST', ''), ('REPORT_SQL_DB_USER', ''),
|
|
('REPORT_SQL_DB_NAME', ''), ('REPORT_SQL_DB_PASSWORD', None),
|
|
('REPORT_SQL_PORT', 'bad'), ('REPORT_SQL_PORT', '0'),
|
|
('REPORT_SQL_DB_PORT', '65536'), ('REPORT_SQL_RELAY_PORT', '-1'),
|
|
):
|
|
with self.subTest(name=name, value=value):
|
|
env = dict(self.env)
|
|
if value is None:
|
|
env.pop(name)
|
|
else:
|
|
env[name] = value
|
|
error = io.StringIO()
|
|
with redirect_stderr(error), self.assertRaises(SystemExit) as raised:
|
|
self.parse(['--from-env'], env)
|
|
self.assertEqual(raised.exception.code, 2)
|
|
self.assertNotIn(self.env['REPORT_SQL_DB_PASSWORD'], error.getvalue())
|
|
|
|
def test_no_arguments_still_requires_configuration(self):
|
|
with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
self.parse([], self.env)
|
|
|
|
def test_environment_mode_cannot_mix_credentials_with_positionals(self):
|
|
with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
|
|
self.parse(['--from-env', 'host', '4321', 'db', 'user', 'pass', 'logs'], self.env)
|
|
|
|
def test_environment_startup_wires_database_report_and_relay(self):
|
|
script = Path(__file__).resolve().parents[1] / 'report_sql.py'
|
|
self.env.update(REPORT_SQL_DB_PORT='3307', REPORT_SQL_RELAY_PORT='4330',
|
|
REPORT_SQL_RELAY_INTERFACE='0.0.0.0')
|
|
with (
|
|
patch.dict('os.environ', self.env, clear=True),
|
|
patch('sys.argv', [str(script), '--from-env']),
|
|
patch('os.chdir'), patch('signal.signal'),
|
|
patch('setproctitle.setproctitle'),
|
|
patch('mysql.connector.connect', create=True) as connect,
|
|
patch('twisted.internet.reactor.connectTCP') as report,
|
|
patch('twisted.internet.reactor.listenTCP') as relay,
|
|
patch('twisted.internet.reactor.run') as run,
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
runpy.run_path(str(script), run_name='__main__')
|
|
connect.assert_called_once_with(
|
|
host='database', port=3307, user='exporter',
|
|
password=self.env['REPORT_SQL_DB_PASSWORD'], database='logs', charset='utf8mb4')
|
|
self.assertEqual(report.call_args.args[:2], ('127.0.0.1', 4321))
|
|
self.assertEqual(relay.call_args.args[0], 4330)
|
|
self.assertEqual(relay.call_args.kwargs, {'interface': '0.0.0.0'})
|
|
run.assert_called_once_with()
|