huangzd1997
5ee99f9441
Build Backend JAR / build (push) Has been cancelled
A1 收敛后 Java 管理接口普遍要求 require* 身份，而浏览器打到 15124 的 请求不带 Java JWT/aiimage_token Cookie，list_shop_manages 等裸代理调用 只转发浏览器 Cookie，被 Java 判未登录，生产后台店铺列表报「加载失败: 未登录」。_proxy_backend_java 默认补 X-Internal-Token（凭 Flask 侧会话 校验通过的管理员身份）+ 自动注入当前操作用户 operatorId，修复同批全部 裸代理调用的 401；显式传身份参数或 use_internal_token=False 可绕过， 供转发浏览器原始 JWT 的场景。新增回归测试锁定默认行为。
112 lines
5.3 KiB
Python
112 lines
5.3 KiB
Python
"""回归测试：Flask 代理转发 Java 时默认携带 X-Internal-Token。
|
||
|
||
背景：A1 管理后台收敛后，Java 管理接口普遍要求身份（requireAdminOrInternal）。
|
||
浏览器打到 15124 的请求不带 Java JWT/aiimage_token Cookie，list_shop_manages 等
|
||
裸代理调用若只转发浏览器 Cookie 会被 Java 判「未登录」，生产后台店铺列表报
|
||
「加载失败:未登录」。本文件锁定 _proxy_backend_java 默认补内部令牌的行为。
|
||
"""
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import Mock, patch
|
||
|
||
from flask import Flask
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
from blueprints import admin_api
|
||
|
||
|
||
class _FakeSession:
|
||
"""记录请求的伪 requests.Session，返回 Java ApiResponse 成功体。"""
|
||
|
||
def __init__(self):
|
||
self.calls = []
|
||
|
||
def request(self, method, url, **kwargs):
|
||
self.calls.append((method, url, kwargs))
|
||
response = Mock()
|
||
response.status_code = 200
|
||
response.json.return_value = {'success': True, 'data': {'items': []}}
|
||
return response
|
||
|
||
|
||
class AdminProxyInternalTokenTest(unittest.TestCase):
|
||
|
||
def setUp(self):
|
||
self.app = Flask(__name__)
|
||
self.session = _FakeSession()
|
||
self.patches = [
|
||
patch.object(admin_api, '_get_backend_java_session', return_value=self.session),
|
||
patch.object(admin_api, '_resolve_internal_token', return_value='tk-internal-test'),
|
||
]
|
||
for p in self.patches:
|
||
p.start()
|
||
self.addCleanup(p.stop)
|
||
|
||
def _call(self, **kwargs):
|
||
with self.app.test_request_context('/api/admin/shop-manages?page=1&page_size=15'):
|
||
return admin_api._proxy_backend_java('GET', '/api/admin/shop-manages', **kwargs)
|
||
|
||
def test_default_carries_internal_token(self):
|
||
"""默认转发必须带 X-Internal-Token（否则 Java requireAdminOrInternal 判未登录）。"""
|
||
result, error_response, status = self._call(params={'page': 1, 'pageSize': 15})
|
||
self.assertEqual(status, 200)
|
||
self.assertIsNone(error_response)
|
||
self.assertTrue(result.get('success'))
|
||
method, url, kwargs = self.session.calls[0]
|
||
self.assertEqual(method, 'GET')
|
||
self.assertEqual(url, f"{admin_api.backend_java_base_url}/api/admin/shop-manages")
|
||
self.assertEqual(kwargs['headers'].get('X-Internal-Token'), 'tk-internal-test')
|
||
|
||
def test_auto_injects_operator_id_from_session(self):
|
||
"""默认转发自动补当前操作用户 operatorId（Java 内部通道要求令牌+operatorId 双条件）。"""
|
||
with self.app.test_request_context('/api/admin/shop-manages?page=1&page_size=15'):
|
||
with patch.object(admin_api, '_resolve_current_operator_id', return_value=7):
|
||
result, error_response, status = admin_api._proxy_backend_java(
|
||
'GET', '/api/admin/shop-manages', params={'page': 1, 'pageSize': 15})
|
||
self.assertEqual(status, 200)
|
||
self.assertIsNone(error_response)
|
||
method, url, kwargs = self.session.calls[0]
|
||
self.assertEqual(kwargs['params'].get('operatorId'), 7)
|
||
|
||
def test_explicit_operator_id_wins(self):
|
||
"""调用方显式传的 operatorId 优先于自动注入（如超管 role 参数场景）。"""
|
||
with self.app.test_request_context('/api/admin/shop-manages?page=1&page_size=15'):
|
||
with patch.object(admin_api, '_resolve_current_operator_id', return_value=7):
|
||
result, error_response, status = admin_api._proxy_backend_java(
|
||
'GET', '/api/admin/shop-manages',
|
||
params={'page': 1, 'pageSize': 15, 'operatorId': 99, 'superAdmin': 'true'})
|
||
self.assertEqual(status, 200)
|
||
method, url, kwargs = self.session.calls[0]
|
||
self.assertEqual(kwargs['params'].get('operatorId'), 99)
|
||
|
||
def test_no_session_no_operator_id(self):
|
||
"""无请求上下文（定时任务线程）时不注入 operatorId，由调用方自行决定。"""
|
||
result, error_response, status = admin_api._proxy_backend_java(
|
||
'GET', '/api/admin/shop-manages', params={'page': 1})
|
||
self.assertEqual(status, 200)
|
||
method, url, kwargs = self.session.calls[0]
|
||
self.assertNotIn('operatorId', kwargs.get('params') or {})
|
||
|
||
def test_opt_out_omits_internal_token(self):
|
||
"""显式 use_internal_token=False 时不带内部令牌（供转发浏览器原始 JWT 的场景）。"""
|
||
result, error_response, status = self._call(
|
||
params={'page': 1}, use_internal_token=False)
|
||
self.assertEqual(status, 200)
|
||
self.assertIsNone(error_response)
|
||
method, url, kwargs = self.session.calls[0]
|
||
self.assertNotIn('X-Internal-Token', kwargs.get('headers') or {})
|
||
self.assertNotIn('operatorId', kwargs.get('params') or {})
|
||
|
||
def test_explicit_headers_override_merge(self):
|
||
"""显式传 headers 时与内部令牌合并，不互相覆盖。"""
|
||
result, error_response, status = self._call(headers={'X-Forwarded-For': '1.2.3.4'})
|
||
self.assertEqual(status, 200)
|
||
method, url, kwargs = self.session.calls[0]
|
||
self.assertEqual(kwargs['headers'].get('X-Internal-Token'), 'tk-internal-test')
|
||
self.assertEqual(kwargs['headers'].get('X-Forwarded-For'), '1.2.3.4')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|