补充真实配置历史写入和测试账号锁定复核

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()

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'):

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)

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', '测试数据库账号锁定未生效'