railwaysystem/scripts/qa/test_qa_tools.py

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.qa_isolation, 'require', return_value={'isolation_id':'synthetic'}), \
                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.qa_isolation, 'require', return_value={'isolation_id':'synthetic'}), \
                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()