fix(C-07):搜索边界收口+结果跳转死链修复

q超长截断100字符、page/page_size健壮解析、types白名单;文章/工具/课程/接口结果URL对齐前端现行路由(/post、/utility、/course-learn、/open-api-detail)。
This commit is contained in:
chunyu
2026-09-15 15:20:19 +08:00
parent 3618323192
commit 1e95db0bda
3 changed files with 133 additions and 10 deletions
View File
+90
View File
@@ -0,0 +1,90 @@
"""C-07 全站统一搜索回归:多模块聚合/类型筛选/分页/异常输入健壮性/结果跳转死链。"""
from django.test import TestCase
from rest_framework.test import APIClient
from article.models import Article
from tool.models import Tool
from learn.models import Course
from apidirectory.models import ApiItem
from user.models import FUser
KW = 'c07searchkey'
class UnifiedSearchTest(TestCase):
def setUp(self):
self.client = APIClient()
self.u = FUser.objects.create_user(username='c07u', password='x12345678')
self.art = Article.objects.create(
title=f'{KW}文章', content='x' * 500, author=self.u, status='published')
self.tool = Tool.objects.create(
name=f'{KW}工具', description='desc', url_path='/utility/c07-tool')
self.course = Course.objects.create(
title=f'{KW}课程', description='desc', author=self.u, status='published')
self.api = ApiItem.objects.create(
name=f'{KW}接口', description='desc', url_path='/api/c07',
method='GET', is_enabled=True)
def _get(self, **params):
r = self.client.get('/search/', params)
self.assertEqual(r.status_code, 200)
body = r.json()
self.assertEqual(body.get('code'), 10000)
return body['data']
def test_multi_module_hit(self):
data = self._get(q=KW)
titles = [x['title'] for x in data['results']]
self.assertIn(f'{KW}文章', titles)
self.assertIn(f'{KW}工具', titles)
self.assertIn(f'{KW}课程', titles)
self.assertIn(f'{KW}接口', titles)
self.assertGreaterEqual(data['total'], 4)
def test_type_filter_article(self):
data = self._get(q=KW, type='article')
self.assertTrue(data['results'])
self.assertTrue(all(x['type'] == '文章' for x in data['results']))
def test_type_filter_tool(self):
data = self._get(q=KW, type='tool')
self.assertTrue(data['results'])
self.assertTrue(all(x['type'] == '工具' for x in data['results']))
def test_result_urls_are_live_routes(self):
data = self._get(q=KW)
by_type = {x['type']: x for x in data['results']}
self.assertTrue(by_type['文章']['url'].startswith('/post/'))
self.assertEqual(by_type['工具']['url'], '/utility/c07-tool')
self.assertTrue(by_type['课程']['url'].startswith('/course-learn?id='))
self.assertTrue(by_type['API']['url'].startswith('/open-api-detail/'))
def test_pagination(self):
data = self._get(q=KW, page=1, page_size=2)
self.assertEqual(len(data['results']), 2)
data2 = self._get(q=KW, page=2, page_size=2)
self.assertTrue(data2['results'])
def test_empty_q_graceful(self):
data = self._get(q='')
self.assertEqual(data['results'], [])
self.assertEqual(data['total'], 0)
def test_long_q_truncated_no_500(self):
data = self._get(q='x' * 5000)
self.assertIn('results', data)
def test_special_chars_no_500(self):
for bad in ['%_%', "a'b\"c", '<script>', '*/--', '你好*?']:
data = self._get(q=bad)
self.assertIn('results', data)
def test_invalid_type_falls_back_all(self):
data = self._get(q=KW, type='not_a_type')
self.assertGreaterEqual(data['total'], 4)
def test_invalid_page_params_no_500(self):
data = self._get(q=KW, page='abc', page_size='xyz')
self.assertIn('results', data)
data = self._get(q=KW, page=-5, page_size=9999)
self.assertIn('results', data)
+43 -10
View File
@@ -22,6 +22,12 @@ SORT_MAPPING = {
'3': 'newest',
}
# C-07 边界收口:q 超长截断、page/page_size 健壮解析、types 白名单。
MAX_Q_LEN = 100
MAX_PAGE_SIZE = 50
DEFAULT_PAGE_SIZE = 12
CONTENT_TYPES = ('all', 'article', 'tool', 'course', 'api')
def normalize_sort(sort_value):
return SORT_MAPPING.get(sort_value, sort_value)
@@ -44,11 +50,30 @@ class GlobalSearchView(APIView):
responses={200: success_response}
)
async def get(self, request):
q = request.query_params.get('q', '').strip()
content_type = request.query_params.get('type', 'all')
page = int(request.query_params.get('page', 1))
page_size = int(request.query_params.get('page_size', 12))
sort = normalize_sort(request.query_params.get('sort', 'relevance'))
raw_q = request.query_params.get('q', '')
if not isinstance(raw_q, str):
raw_q = str(raw_q)
q = raw_q.strip()[:MAX_Q_LEN]
raw_type = request.query_params.get('type', 'all')
if not isinstance(raw_type, str):
raw_type = 'all'
content_type = raw_type.strip().lower() or 'all'
if content_type not in CONTENT_TYPES:
content_type = 'all'
try:
page = int(request.query_params.get('page', 1))
except (TypeError, ValueError):
page = 1
try:
page_size = int(request.query_params.get('page_size', DEFAULT_PAGE_SIZE))
except (TypeError, ValueError):
page_size = DEFAULT_PAGE_SIZE
page = max(page, 1)
page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
raw_sort = request.query_params.get('sort', 'relevance')
if not isinstance(raw_sort, str):
raw_sort = str(raw_sort)
sort = normalize_sort(raw_sort)
if not q:
return create_standardized_response(data={
@@ -82,7 +107,7 @@ class GlobalSearchView(APIView):
'cover': cover_url,
'views': article.views or 0,
'likes': article.likes or 0,
'url': f'/articles/{article.id}',
'url': f'/post/{article.id}',
})
if content_type == 'all' or content_type == 'tool':
@@ -99,7 +124,7 @@ class GlobalSearchView(APIView):
'cover': '',
'views': tool.usage_count or 0,
'likes': 0,
'url': f'/use?tool={tool.id}',
'url': tool.url_path or '/utility',
})
if content_type == 'all' or content_type == 'course':
courses = Course.objects.filter(
@@ -119,6 +144,7 @@ class GlobalSearchView(APIView):
'cover': cover_url,
'views': 0,
'likes': 0,
'url': f'/course-learn?id={course.id}',
})
if content_type == 'all' or content_type == 'api':
@@ -136,7 +162,7 @@ class GlobalSearchView(APIView):
'cover': api.image_url or '',
'views': api.views_count or 0,
'likes': 0,
'url': f'/api-detail/{api.id}',
'url': f'/open-api-detail/{api.id}',
})
if sort == 'views':
@@ -176,8 +202,15 @@ class SearchSuggestionsView(APIView):
responses={200: success_response}
)
async def get(self, request):
q = request.query_params.get('q', '').strip()
limit = int(request.query_params.get('limit', 8))
raw_q = request.query_params.get('q', '')
if not isinstance(raw_q, str):
raw_q = str(raw_q)
q = raw_q.strip()[:MAX_Q_LEN]
try:
limit = int(request.query_params.get('limit', 8))
except (TypeError, ValueError):
limit = 8
limit = min(max(limit, 1), MAX_PAGE_SIZE)
if not q:
return create_standardized_response(data=[], code=ResponseCode.SUCCESS)