111 lines
6.1 KiB
Python
111 lines
6.1 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('qa', Path(__file__).with_name('qa.py'))
|
|
qa = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(qa)
|
|
|
|
|
|
class QaToolsTest(unittest.TestCase):
|
|
def test_defaults_never_use_old_database_url(self):
|
|
env = qa.environment({'QA_DB_URL': 'jdbc:kingbase8://localhost/training', 'FUSION_QA_APP_PASSWORD': 'synthetic'})
|
|
self.assertIn('/' + qa.DATABASE, env['QA_DB_URL'])
|
|
self.assertEqual('fusion_qa_app', env['QA_DB_USER'])
|
|
|
|
def test_wrong_database_rejected(self):
|
|
for database in ['training', 'iotlocks', 'railwaysystem_integrated_qa_20260916']:
|
|
with self.assertRaises(ValueError):
|
|
qa.environment({'FUSION_QA_DB_URL': 'jdbc:kingbase8://localhost/' + database, 'FUSION_QA_APP_PASSWORD': 'synthetic'})
|
|
|
|
def test_url_credentials_rejected(self):
|
|
for url in ['jdbc:kingbase8://user:secret@localhost/' + qa.DATABASE,
|
|
'jdbc:kingbase8://localhost/' + qa.DATABASE + '?password=synthetic']:
|
|
with self.assertRaises(ValueError):
|
|
qa.environment({'FUSION_QA_DB_URL': url, 'FUSION_QA_APP_PASSWORD': 'synthetic'})
|
|
|
|
def test_missing_password_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
qa.environment({})
|
|
|
|
def test_druid_timeout_units_cannot_be_overridden_by_jdbc_url_seconds(self):
|
|
url = 'jdbc:kingbase8://192.168.139.218:4321/' + qa.DATABASE + '?connectTimeout=5&socketTimeout=15&characterEncoding=utf8'
|
|
self.assertEqual('jdbc:kingbase8://192.168.139.218:4321/' + qa.DATABASE + '?characterEncoding=utf8', qa.druid_url(url))
|
|
|
|
def test_unsafe_inherited_overrides_removed(self):
|
|
with patch.dict(os.environ, {'SPRING_APPLICATION_JSON': '{"danger":true}', 'IOT_MQTT_ENABLED': 'true', 'JAVA_TOOL_OPTIONS': '-Dunsafe=true'}):
|
|
env = qa.environment({'FUSION_QA_APP_PASSWORD': 'synthetic'})
|
|
self.assertNotIn('SPRING_APPLICATION_JSON', env)
|
|
self.assertNotIn('IOT_MQTT_ENABLED', env)
|
|
self.assertNotIn('JAVA_TOOL_OPTIONS', env)
|
|
|
|
def test_private_config_required(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'config.env'
|
|
path.write_text('FUSION_QA_APP_PASSWORD=synthetic\n')
|
|
path.chmod(0o644)
|
|
with self.assertRaises(ValueError):
|
|
qa.load_config(path)
|
|
path.chmod(0o600)
|
|
self.assertEqual('synthetic', qa.load_config(path)['FUSION_QA_APP_PASSWORD'])
|
|
|
|
def test_config_does_not_execute_shell(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'config.env'
|
|
path.write_text("KEY='$(touch should-not-exist)'\n")
|
|
path.chmod(0o600)
|
|
self.assertEqual('$(touch should-not-exist)', qa.load_config(path)['KEY'])
|
|
|
|
def test_migration_preflight_prevents_all_processes(self):
|
|
with patch.object(qa, 'inspect', return_value=2), patch.object(qa, 'launch') as launch:
|
|
with self.assertRaises(ValueError):
|
|
qa.start({}, {}, 18096, 16389)
|
|
launch.assert_not_called()
|
|
|
|
def test_start_flags_and_secret_handling(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(qa, 'RUNTIME', Path(directory)), \
|
|
patch.object(qa, 'inspect', return_value=0), patch.object(qa, 'available'), patch.object(qa, 'launch') as launch:
|
|
config = {'FUSION_QA_APP_PASSWORD':'synthetic', 'TRAIN_STUDENT_IDENTITY_ENCRYPTION_KEY':'synthetic', 'TRAIN_STUDENT_IDENTITY_HASH_KEY':'synthetic'}
|
|
qa.start(config, qa.environment(config), 18096, 16389)
|
|
self.assertEqual(2,launch.call_count)
|
|
name, command, env = launch.call_args.args
|
|
self.assertEqual('backend',name)
|
|
for flag in ['--iot.mqtt.enabled=false','--iot.mqtt.mock=true', '--iot.mqtt.real-device-write-enabled=false',
|
|
'--iot.mqtt.device-whitelist=', '--iot.mqtt.retry-task-enabled=false', '--iot.qa-api-enabled=false',
|
|
'--iot.openapi.enabled=false','--iot.face.real-http-enabled=false','--dms.enabled=false',
|
|
'--spring.quartz.auto-startup=false','--spring.datasource.druid.master.username=fusion_qa_app']:
|
|
self.assertIn(flag,command)
|
|
self.assertNotIn(env['TOKEN_SECRET'],' '.join(command))
|
|
self.assertEqual(0o600,stat.S_IMODE((Path(directory)/'runtime-secrets.env').stat().st_mode))
|
|
|
|
def test_stop_rejects_reused_pid(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(qa, 'RUNTIME', Path(directory)), \
|
|
patch.object(qa.subprocess, 'run') as run, patch.object(qa.os, 'kill') as kill:
|
|
(Path(directory)/'backend.json').write_text(json.dumps({'pid':123,'command':['task-owned-java']}))
|
|
run.return_value.stdout='unrelated-user-process'
|
|
with self.assertRaises(ValueError):
|
|
qa.stop()
|
|
kill.assert_not_called()
|
|
|
|
def test_stop_accepts_python_wrapper_only_with_task_identity(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(qa, 'RUNTIME', Path(directory)), \
|
|
patch.object(qa.subprocess, 'run') as run, patch.object(qa.os, 'kill') as kill, patch.object(qa, 'wait_for_exit'):
|
|
script = str(qa.ROOT / 'scripts/qa/serve_frontend.py')
|
|
command = ['/usr/local/bin/python3', script, '--port', '18088']
|
|
(Path(directory)/'frontend.json').write_text(json.dumps({'pid':123,'command':command,'started':'same-birth-time'}))
|
|
run.side_effect = [SimpleNamespace(stdout='/Frameworks/Python.app/Python ' + ' '.join(command[1:])),
|
|
SimpleNamespace(stdout='p123\nfcwd\nn' + str(qa.ROOT) + '\n'),
|
|
SimpleNamespace(stdout='same-birth-time'), SimpleNamespace(stdout='/Frameworks/Python.app/Python')]
|
|
qa.stop()
|
|
kill.assert_called_once_with(123,qa.signal.SIGTERM)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|