298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""集中式用户数据变更审计模块测试（tests/test_audit.py）
|
||
|
||
不依赖真实 MySQL：使用 Flask test app + unittest.mock 替换 services.db.query。
|
||
运行方式：python3 -m unittest tests.test_audit -v（pytest 亦可运行本文件）。
|
||
"""
|
||
import io
|
||
import json
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
from flask import Flask, jsonify
|
||
|
||
import services.db as db
|
||
from services.audit import (
|
||
init_audit, mask_sensitive, _client_ip, _is_sensitive_key,
|
||
EXCLUDED_PATHS, ACTION_DATA_CHANGE,
|
||
)
|
||
|
||
|
||
def _make_app():
|
||
"""构建最小 Flask 测试应用，注册集中式审计钩子。"""
|
||
app = Flask(__name__)
|
||
app.secret_key = 'test-secret'
|
||
|
||
@app.route('/api/simple', methods=['POST'])
|
||
def simple():
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
@app.route('/api/no_code', methods=['POST'])
|
||
def no_code():
|
||
# 无 code 字段但 HTTP 成功 → 视为成功并记录
|
||
return jsonify({"success": True})
|
||
|
||
@app.route('/api/biz_fail', methods=['POST'])
|
||
def biz_fail():
|
||
return jsonify({"code": 500, "msg": "业务失败"})
|
||
|
||
@app.route('/api/http_fail', methods=['POST'])
|
||
def http_fail():
|
||
return jsonify({"code": 0, "msg": "ok"}), 400
|
||
|
||
@app.route('/api/with_log', methods=['POST'])
|
||
def with_log():
|
||
db.add_operation_log('业务日志', '用户修改了产品')
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
@app.route('/api/multi_log', methods=['POST'])
|
||
def multi_log():
|
||
db.add_operation_log('日志A', '第一条业务日志')
|
||
db.add_operation_log('日志B', '第二条业务日志')
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
@app.route('/api/with_log_fail', methods=['POST'])
|
||
def with_log_fail():
|
||
db.add_operation_log('业务日志', '执行了但响应失败')
|
||
return jsonify({"code": 1, "msg": "失败"}), 200
|
||
|
||
@app.route('/api/with_password', methods=['POST'])
|
||
def with_password():
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
@app.route('/api/upload', methods=['POST'])
|
||
def upload():
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
@app.route('/api/put_style', methods=['PUT'])
|
||
def put_style():
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
@app.route('/api/get_only', methods=['GET'])
|
||
def get_only():
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
|
||
# 排除接口（本应返回成功，验证不记录）
|
||
for path in EXCLUDED_PATHS:
|
||
def _excluded_route():
|
||
return jsonify({"code": 0, "msg": "ok"})
|
||
app.add_url_rule(path, view_func=_excluded_route,
|
||
methods=['POST'], endpoint='excluded_' + path.replace('/', '_'))
|
||
|
||
init_audit(app)
|
||
return app
|
||
|
||
|
||
class AuditTestCase(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.app = _make_app()
|
||
self.app.config['TESTING'] = True
|
||
self.calls = [] # 记录所有 (sql, params)
|
||
|
||
patcher = mock.patch.object(db, 'query', side_effect=self._fake_query)
|
||
patcher.start()
|
||
self.addCleanup(patcher.stop)
|
||
|
||
def _fake_query(self, sql, params=None):
|
||
self.calls.append((sql, params))
|
||
return 123 # 模拟 INSERT 返回的 lastrowid
|
||
|
||
# ── 工具方法 ──
|
||
def _client(self):
|
||
client = self.app.test_client()
|
||
with client.session_transaction() as sess:
|
||
sess['user_id'] = 7
|
||
sess['username'] = 'tester'
|
||
return client
|
||
|
||
def _inserts(self):
|
||
"""返回所有 INSERT dashboard_logs 调用。"""
|
||
return [c for c in self.calls if 'INSERT INTO dashboard_logs' in c[0]]
|
||
|
||
# ── 用例 ──
|
||
def test_success_without_existing_log_auto_records(self):
|
||
client = self._client()
|
||
resp = client.post('/api/simple', json={"name": "产品A"})
|
||
self.assertEqual(resp.status_code, 200)
|
||
inserts = self._inserts()
|
||
self.assertEqual(len(inserts), 1, f"应有 1 条审计日志，实际 {len(inserts)}")
|
||
sql, params = inserts[0]
|
||
self.assertEqual(params[0], 7) # user_id
|
||
self.assertEqual(params[1], ACTION_DATA_CHANGE) # action
|
||
detail = params[2]
|
||
self.assertIn('POST /api/simple', detail)
|
||
self.assertIn('name：产品A', detail)
|
||
self.assertNotIn('{"', detail)
|
||
self.assertFalse(detail.endswith('}'))
|
||
self.assertEqual(len(detail) <= 12000, True)
|
||
|
||
def test_nested_json_is_summarized_not_dumped(self):
|
||
client = self._client()
|
||
client.post('/api/simple', json={
|
||
"product_code": "ABC-001",
|
||
"materials": [{"material_code": "M1"}, {"material_code": "M2"}],
|
||
})
|
||
detail = self._inserts()[0][1][2]
|
||
self.assertIn('product_code：ABC-001', detail)
|
||
self.assertIn('materials：2项', detail)
|
||
self.assertNotIn('material_code', detail)
|
||
|
||
def test_response_without_code_field_still_recorded(self):
|
||
client = self._client()
|
||
client.post('/api/no_code', json={})
|
||
self.assertEqual(len(self._inserts()), 1)
|
||
|
||
def test_business_failure_not_recorded(self):
|
||
client = self._client()
|
||
client.post('/api/biz_fail', json={})
|
||
self.assertEqual(self._inserts(), [], '业务 code 非 0/200 不应记录')
|
||
|
||
def test_http_failure_not_recorded(self):
|
||
client = self._client()
|
||
client.post('/api/http_fail', json={})
|
||
self.assertEqual(self._inserts(), [], 'HTTP >= 400 不应记录')
|
||
|
||
def test_existing_business_log_is_kept_readable_without_technical_suffix(self):
|
||
client = self._client()
|
||
client.post('/api/with_log', json={"qty": 10})
|
||
inserts = self._inserts()
|
||
# 只有 add_operation_log 产生的业务日志，不再插入第二条
|
||
self.assertEqual(len(inserts), 1)
|
||
self.assertEqual(inserts[0][1][1], '业务日志')
|
||
self.assertFalse(
|
||
any('UPDATE dashboard_logs' in sql for sql, _ in self.calls),
|
||
'已有可读业务日志时不应追加技术路径或请求体'
|
||
)
|
||
|
||
def test_multiple_business_logs_all_appended_no_new_log(self):
|
||
client = self._client()
|
||
client.post('/api/multi_log', json={})
|
||
inserts = self._inserts()
|
||
self.assertEqual(len(inserts), 2, '两条业务日志，不应再插入审计日志')
|
||
self.assertFalse(
|
||
any('UPDATE dashboard_logs' in sql for sql, _ in self.calls),
|
||
'多条业务日志也应保持原样'
|
||
)
|
||
|
||
def test_failure_with_existing_log_does_not_append_context(self):
|
||
client = self._client()
|
||
client.post('/api/with_log_fail', json={})
|
||
self.assertEqual(len(self._inserts()), 1, '业务日志本身存在')
|
||
self.assertFalse(any('UPDATE dashboard_logs' in sql for sql, _ in self.calls))
|
||
|
||
def test_password_masked_in_detail(self):
|
||
client = self._client()
|
||
client.post('/api/with_password', json={
|
||
"username": "admin",
|
||
"password": "TopSecret123",
|
||
"new_password": "NewSecret456",
|
||
"confirm_password": "NewSecret456",
|
||
"access_token": "tok-abc",
|
||
"api_key": "key-xyz",
|
||
"remark": "普通内容",
|
||
})
|
||
inserts = self._inserts()
|
||
self.assertEqual(len(inserts), 1)
|
||
detail = inserts[0][1][2]
|
||
self.assertIn('[已隐藏]', detail)
|
||
self.assertNotIn('TopSecret123', detail)
|
||
self.assertNotIn('NewSecret456', detail)
|
||
self.assertNotIn('tok-abc', detail)
|
||
self.assertNotIn('key-xyz', detail)
|
||
self.assertIn('普通内容', detail)
|
||
|
||
def test_upload_filename_recorded(self):
|
||
client = self._client()
|
||
data = {
|
||
'file': (io.BytesIO(b'%PDF-1.4 fake'), '报价单.pdf'),
|
||
'product_code': 'ABC-001',
|
||
}
|
||
client.post('/api/upload', data=data, content_type='multipart/form-data')
|
||
inserts = self._inserts()
|
||
self.assertEqual(len(inserts), 1)
|
||
detail = inserts[0][1][2]
|
||
self.assertIn('文件名: 报价单.pdf', detail)
|
||
self.assertIn('ABC-001', detail)
|
||
|
||
def test_put_method_recorded_get_not(self):
|
||
client = self._client()
|
||
client.put('/api/put_style', json={"a": 1})
|
||
self.assertEqual(len(self._inserts()), 1)
|
||
self.calls.clear()
|
||
client.get('/api/get_only')
|
||
self.assertEqual(self._inserts(), [], 'GET 不记录')
|
||
|
||
def test_excluded_paths_not_recorded(self):
|
||
client = self._client()
|
||
for path in EXCLUDED_PATHS:
|
||
self.calls.clear()
|
||
resp = client.post(path, json={"x": 1})
|
||
self.assertEqual(resp.status_code, 200, f'{path} 应返回成功')
|
||
self.assertEqual(self._inserts(), [], f'排除接口不应记录: {path}')
|
||
|
||
def test_anonymous_user_not_recorded(self):
|
||
client = self.app.test_client() # 未登录
|
||
client.post('/api/simple', json={})
|
||
self.assertEqual(self._inserts(), [], '未登录用户不记录')
|
||
|
||
# ── 单元级：脱敏 ──
|
||
def test_mask_sensitive_nested(self):
|
||
data = {
|
||
"login": {"password": "p1", "token": "t1", "note": "ok"},
|
||
"items": [{"secret": "s1"}, {"cookie": "c1"}],
|
||
"old_password": "op",
|
||
"user_password_reset": "x", # 非 *_password 结尾、非精确匹配 → 不脱敏
|
||
}
|
||
out = mask_sensitive(data)
|
||
self.assertEqual(out['login']['password'], '[已隐藏]')
|
||
self.assertEqual(out['login']['token'], '[已隐藏]')
|
||
self.assertEqual(out['login']['note'], 'ok')
|
||
self.assertEqual(out['items'][0]['secret'], '[已隐藏]')
|
||
self.assertEqual(out['items'][1]['cookie'], '[已隐藏]')
|
||
self.assertEqual(out['old_password'], '[已隐藏]')
|
||
self.assertEqual(out['user_password_reset'], 'x')
|
||
|
||
def test_is_sensitive_key(self):
|
||
for k in ['password', 'new_password', 'confirm_password', 'token',
|
||
'access_token', 'refresh_token', 'secret', 'api_key',
|
||
'authorization', 'cookie', 'my_password']:
|
||
self.assertTrue(_is_sensitive_key(k), k)
|
||
for k in ['username', 'remark', 'password_reset', 'width', 'page']:
|
||
self.assertFalse(_is_sensitive_key(k), k)
|
||
|
||
# ── 单元级：IP 获取 ──
|
||
def test_client_ip_prefers_xff_first_address(self):
|
||
with self.app.test_request_context(
|
||
'/api/simple', headers={'X-Forwarded-For': '1.2.3.4, 5.6.7.8'}):
|
||
self.assertEqual(_client_ip(), '1.2.3.4')
|
||
|
||
def test_client_ip_falls_back_to_remote_addr(self):
|
||
with self.app.test_request_context('/api/simple', environ_base={'REMOTE_ADDR': '9.9.9.9'}):
|
||
self.assertEqual(_client_ip(), '9.9.9.9')
|
||
|
||
def test_client_ip_empty_xff_falls_back(self):
|
||
with self.app.test_request_context(
|
||
'/api/simple', headers={'X-Forwarded-For': ' '},
|
||
environ_base={'REMOTE_ADDR': '8.8.8.8'}):
|
||
self.assertEqual(_client_ip(), '8.8.8.8')
|
||
|
||
# ── 集成：add_operation_log 返回 id 并注册到 g ──
|
||
def test_add_operation_log_returns_id_and_registers_in_g(self):
|
||
with self.app.test_request_context('/api/simple', method='POST'):
|
||
from flask import g
|
||
log_id = db.add_operation_log('测试动作', '测试详情', user_id=1)
|
||
self.assertEqual(log_id, 123)
|
||
self.assertEqual(g.audit_log_ids, [123])
|
||
|
||
def test_audit_insensitive_to_db_failure(self):
|
||
"""DB 异常时审计静默失败，不影响业务响应。"""
|
||
with mock.patch.object(db, 'query', side_effect=RuntimeError('db down')):
|
||
client = self._client()
|
||
resp = client.post('/api/simple', json={"a": 1})
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.get_json()['code'], 0)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|