"""集中式用户数据变更审计模块测试（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()