diff --git a/test/deployment/test_runtime_config.py b/test/deployment/test_runtime_config.py new file mode 100644 index 00000000..74618f31 --- /dev/null +++ b/test/deployment/test_runtime_config.py @@ -0,0 +1,42 @@ +"""配置历史实机探针必须真实提交变更,不能以空表存在代替验收。""" + +from pathlib import Path +import sys +import unittest +from unittest.mock import MagicMock + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) +from runtime_probe import verify_config_history + + +class RuntimeConfigTest(unittest.TestCase): + def client(self, final_generation=2, final_level='warn'): + client = MagicMock() + client.success.side_effect = [ + {'available': True, 'activeGeneration': 1}, + {'generation': 1, 'logLevel': 'info', 'databasePasswordStatus': {}}, + {}, {'generation': 2, 'candidateId': 'candidate'}, {}, + {'generation': final_generation, 'logLevel': final_level}, + ] + return client + + def test_applies_real_change_and_verifies_generation(self): + client = self.client() + self.assertTrue(verify_config_history(client)['generation_advanced']) + calls = client.success.call_args_list + request = calls[2].args[1] + self.assertEqual(request['expectedGeneration'], 1) + self.assertNotIn('databasePasswordStatus', request['payload']) + self.assertEqual(calls[3].args[2], 'PUT') + self.assertEqual(calls[4].args[1]['candidateId'], 'candidate') + + def test_rejects_unchanged_generation_or_unapplied_payload(self): + for generation, level in ((1, 'warn'), (2, 'info')): + with self.subTest(generation=generation, level=level): + with self.assertRaises(AssertionError): + verify_config_history(self.client(generation, level)) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/deployment/test_runtime_sql.py b/test/deployment/test_runtime_sql.py index 9bb81f0b..5a7acc44 100644 --- a/test/deployment/test_runtime_sql.py +++ b/test/deployment/test_runtime_sql.py @@ -53,6 +53,7 @@ class RuntimeSQLTest(unittest.TestCase): fixture = runtime_sql.SQLFixture(kind) with patch.object(fixture, 'admin', return_value='0') as admin: fixture.create() + admin.return_value = 'Y' if kind == 'mysql' else 'f' fixture.lock() statements = '\n'.join(item.args[0] for item in admin.call_args_list) self.assertNotIn('DROP ', statements) @@ -70,6 +71,15 @@ class RuntimeSQLTest(unittest.TestCase): fixture.lock() admin.assert_not_called() + def test_lock_requires_server_confirmation(self): + for kind in ('mysql', 'postgres'): + with self.subTest(kind=kind): + fixture = runtime_sql.SQLFixture(kind) + fixture.account_created = True + with patch.object(fixture, 'admin', return_value=''): + with self.assertRaises(AssertionError): + fixture.lock() + def test_nonempty_database_fails_first_install_check(self): fixture = runtime_sql.SQLFixture('postgres') with patch.object(fixture, 'admin', return_value='3'): diff --git a/test/tools/runtime_probe.py b/test/tools/runtime_probe.py index 0f083f6b..d349a4ae 100644 --- a/test/tools/runtime_probe.py +++ b/test/tools/runtime_probe.py @@ -106,6 +106,26 @@ def verify_message_content(raw, message_id, body, attachment): assert files[0].get_payload(decode=True) == attachment +def verify_config_history(client): + # 首次安装不一定产生历史,必须通过正式变更接口写入并读回,不能只检查空表存在。 + status = client.success('/api/v1/config/status', method='GET') + assert status['available'] + payload = client.success('/api/v1/config/base', method='GET') + payload.pop('databasePasswordStatus', None) + payload['logLevel'] = 'warn' if payload.get('logLevel') != 'warn' else 'info' + request = {'expectedGeneration': status['activeGeneration'], 'payload': payload, + 'confirmation': {'confirmed': True, 'method': 'recent_auth'}} + client.success('/api/v1/config/base/validate', request) + candidate = client.success('/api/v1/config/base', request, 'PUT') + client.success('/api/v1/config/base/apply', { + 'generation': candidate['generation'], 'candidateId': candidate['candidateId'], + }) + active = client.success('/api/v1/config/base', method='GET') + assert active['logLevel'] == payload['logLevel'] + assert active['generation'] > status['activeGeneration'] + return {'configuration_api': True, 'generation_advanced': True} + + def verify_protocol_read(context, codes, message_id, body, attachment): # 通过协议取回真实正文,不以数据库存在或欢迎响应替代取信验收。 client = LocalIMAP(HOST, 993, ssl_context=context, timeout=30) @@ -332,6 +352,7 @@ def main(): assert len(recovery['codes']) == 10 recovery.clear() result['security_setup'] = {'recovery_codes_generated': 10, 'plaintext_logged': False} + result['config_history'] = verify_config_history(client) if args.proxy: from runtime_proxy import configure_proxy result['proxy'] = configure_proxy(client, context, login, Client, LocalHTTPS) diff --git a/test/tools/runtime_sql.py b/test/tools/runtime_sql.py index 2a87c8aa..b55cc974 100644 --- a/test/tools/runtime_sql.py +++ b/test/tools/runtime_sql.py @@ -124,5 +124,9 @@ class SQLFixture(SQLDatabase): user = self.settings['user'] if self.kind == 'mysql': self.admin("ALTER USER '" + user + "'@'localhost' ACCOUNT LOCK;") + assert self.admin("SELECT account_locked FROM mysql.user WHERE User='" + user + + "' AND Host='localhost';") == 'Y', '测试数据库账号锁定未生效' else: self.admin('ALTER ROLE ' + user + ' NOLOGIN;') + assert self.admin("SELECT rolcanlogin FROM pg_roles WHERE rolname='" + user + + "';") == 'f', '测试数据库账号锁定未生效'