import asyncio from datetime import datetime, timedelta, timezone import pytest from aiohttp.test_utils import TestClient, TestServer from api.app import ALIYUN_LOGGER_KEY, API_PATH, LOGGER_KEY, create_app from api.chui_zhi.videos import ( BusinessValidationError, MYSQL_KEY, QUERY_SEMAPHORE_KEY, FilterCondition, VideoQueryRequest, build_query, compile_filter_condition, ) from config import settings def test_build_query_contains_parameterized_filters(): request = VideoQueryRequest( platforms=['xiaoniangao', 'xiaoniangaotuijianliu'], start_time=datetime(2026, 8, 4), end_time=datetime(2026, 8, 5), keywords=['养生'], filter_match_mode=2, filters=[ FilterCondition(field='like_cnt', operator='>=', value=100), FilterCondition(field='duration', operator='between', value=[60, 300]), ], limit=20, ) sql, params = build_query(request) assert '`platform` IN (%s, %s)' in sql assert '`video_title` LIKE %s' in sql assert '`like_cnt` >= %s' in sql assert '`duration` BETWEEN %s AND %s' in sql assert 'SELECT MAX(`source`.`id`) AS `selected_id`' in sql assert "CASE WHEN `source`.`out_video_id` = '' THEN `source`.`id` ELSE 0 END" in sql assert '`dedup`.`selected_id` = `cv`.`id`' in sql assert 'ORDER BY `cv`.`id` DESC' in sql assert 'OFFSET' not in sql assert params[-1] == 21 assert '%养生%' in params def test_deduplication_groups_ids_before_cursor_pagination(): request = VideoQueryRequest( platforms=['xiaoniangao'], start_time=datetime(2026, 8, 4), end_time=datetime(2026, 8, 5), filters=[FilterCondition(field='like_cnt', operator='>', value=3)], cursor={'id': 100}, limit=50, ) sql, params = build_query(request) # 筛选条件只出现一次;HAVING作用于MAX(id),防止旧重复记录跨页再次出现。 assert sql.count('`source`.`like_cnt` > %s') == 1 assert 'HAVING MAX(`source`.`id`) < %s' in sql assert params[-2:] == [100, 51] def test_query_limit_above_server_max_is_rejected(): with pytest.raises(ValueError): VideoQueryRequest( start_time=datetime(2026, 8, 4), end_time=datetime(2026, 8, 5), limit=settings.API_MAX_LIMIT + 1, ) def test_id_cursor_builds_stable_group_pagination_and_fetches_one_extra_row(): request = VideoQueryRequest( start_time=datetime(2026, 8, 4), end_time=datetime(2026, 8, 5), limit=100, cursor={'id': 123}, ) sql, params = build_query(request) assert 'HAVING MAX(`source`.`id`) < %s' in sql assert params[-2:] == [123, 101] assert 'OFFSET' not in sql def test_millisecond_timestamps_are_converted_to_china_time(): china_timezone = timezone(timedelta(hours=8)) start_time = datetime(2026, 8, 4, tzinfo=china_timezone) end_time = datetime(2026, 8, 5, tzinfo=china_timezone) request = VideoQueryRequest( start_time=int(start_time.timestamp() * 1000), end_time=int(end_time.timestamp() * 1000), ) assert request.start_time == datetime(2026, 8, 4) assert request.end_time == datetime(2026, 8, 5) def test_missing_time_defaults_to_latest_three_days(): before = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None) request = VideoQueryRequest() after = datetime.now(timezone(timedelta(hours=8))).replace(tzinfo=None) assert before <= request.end_time <= after assert request.end_time - request.start_time == timedelta(days=3) sql, sql_params = build_query(request) assert '`create_time` >= %s' in sql assert '`create_time` < %s' in sql assert request.start_time in sql_params assert request.end_time in sql_params def test_rejects_unsupported_filter_field(): with pytest.raises(ValueError): FilterCondition(field='unknown_field', operator='>', value=1) def test_rejects_unsupported_platform_and_unknown_filter_parameter(): with pytest.raises(ValueError, match='不支持的平台'): VideoQueryRequest(platforms=['douyin']) with pytest.raises(ValueError): FilterCondition(field='like_cnt', operator='>', value=1, unknown='value') @pytest.mark.parametrize( 'field', [ 'like_cnt', 'collection_cnt', 'comment_cnt', 'share_cnt', 'play_cnt', 'duration', ], ) def test_supported_numeric_filter_mapping(field): condition = FilterCondition(field=field, operator='>', value=10) sql, params = compile_filter_condition(condition) assert f'`{field}` > %s' == sql assert params == [10] def test_empty_filters_are_not_added_to_query(): request = VideoQueryRequest(filters=[]) sql, _ = build_query(request) where_sql = sql.split('WHERE', 1)[1] assert '`like_cnt` >' not in where_sql def test_publish_time_filter_mapping(): condition = FilterCondition( field='publish_time', operator='>=', value='2026-08-01 00:00:00', ) sql, params = compile_filter_condition(condition) assert sql == '`publish_time` >= %s' assert params == ['2026-08-01 00:00:00'] def test_filter_range_and_set_are_parameterized(): range_condition = FilterCondition(field='duration', operator='between', value=[60, 300]) set_condition = FilterCondition(field='play_cnt', operator='in', value=[10, 20]) range_sql, range_params = compile_filter_condition(range_condition) set_sql, set_params = compile_filter_condition(set_condition) assert range_sql == '`duration` BETWEEN %s AND %s' assert range_params == [60, 300] assert set_sql == '`play_cnt` IN (%s, %s)' assert set_params == [10, 20] def test_rejects_invalid_filter_range_and_non_finite_number(): with pytest.raises(BusinessValidationError, match='起始值不能大于结束值'): compile_filter_condition( FilterCondition(field='duration', operator='between', value=[300, 60]) ) with pytest.raises(BusinessValidationError, match='有限数字'): compile_filter_condition( FilterCondition(field='like_cnt', operator='>', value='NaN') ) def test_rejects_raw_sql_in_filter_value(): with pytest.raises(BusinessValidationError, match='必须是数字'): condition = FilterCondition(field='like_cnt', operator='>', value='0 OR 1=1') compile_filter_condition(condition) @pytest.mark.asyncio async def test_api_requires_token_and_returns_expected_contract(monkeypatch): local_logs = [] cloud_logs = [] class FakeMySQL: async def fetch_all(self, sql, params): return [ {'id': 1, 'platform': params[0], 'create_time': '2026-08-04 12:00:00'}, {'id': 2, 'platform': params[0], 'create_time': '2026-08-04 11:00:00'}, ] class FakeLogger: def info(self, message): local_logs.append(('info', message)) def error(self, message): local_logs.append(('error', message)) def exception(self, message): raise AssertionError(message) class FakeAliyunLogger: def logging_batch(self, events): cloud_logs.extend(events) monkeypatch.setattr(settings, 'CHUI_ZHI_API_TOKEN', 'test-token') app = create_app() app.cleanup_ctx.clear() app[MYSQL_KEY] = FakeMySQL() app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(10) app[LOGGER_KEY] = FakeLogger() app[ALIYUN_LOGGER_KEY] = FakeAliyunLogger() client = TestClient(TestServer(app)) await client.start_server() payload = { 'platforms': ['xiaoniangao'], 'start_time': 1785772800000, 'end_time': 1785859200000, 'filters': [ {'field': 'like_cnt', 'operator': '>', 'value': 3}, ], 'limit': 1, } try: unauthorized = await client.post(API_PATH, json=payload) assert unauthorized.status == 401 authorized = await client.post( API_PATH, json=payload, headers={'X-API-Key': 'test-token', 'X-Request-ID': 'scheduler-request-1'}, ) body = await authorized.json() assert authorized.status == 200 assert body['code'] == 0 assert body['data']['count'] == 1 assert body['data']['has_more'] is True assert body['data']['next_cursor'] == {'id': 1} assert body['data']['start_time'] == payload['start_time'] assert body['data']['end_time'] == payload['end_time'] assert authorized.headers['X-Request-ID'] == 'scheduler-request-1' invalid_payload = { **payload, 'filters': [{'field': 'like_cnt', 'operator': '>', 'value': 'invalid'}], } failed = await client.post( API_PATH, json=invalid_payload, headers={'X-API-Key': 'test-token'}, ) assert failed.status == 422 assert len(cloud_logs) == 3 assert all('event_type' not in log['data'] for log in cloud_logs) assert all('request_count' not in log['data'] for log in cloud_logs) assert all(log['data']['path'] == API_PATH for log in cloud_logs) assert all(log['data']['request_duration_ms'] >= 0 for log in cloud_logs) assert cloud_logs[0]['data']['status_code'] == 401 assert cloud_logs[0]['data']['request_params']['body'] is None assert cloud_logs[0]['data']['message'] == 'unauthorized' assert cloud_logs[0]['data']['failure_stage'] == 'authentication' assert cloud_logs[0]['data']['error_type'] == 'AuthenticationError' assert cloud_logs[1]['data']['status_code'] == 200 assert cloud_logs[1]['trace_id'] == 'scheduler-request-1' assert cloud_logs[1]['data']['request_id'] == 'scheduler-request-1' assert cloud_logs[1]['data']['request_params']['body'] == payload assert 'response' not in cloud_logs[1]['data'] assert cloud_logs[1]['data']['success'] is True assert cloud_logs[1]['data']['query_result_count'] == 2 assert cloud_logs[1]['data']['query_duration_ms'] >= 0 assert cloud_logs[1]['data']['query_success'] is True assert cloud_logs[2]['data']['status_code'] == 422 assert cloud_logs[2]['data']['success'] is False assert cloud_logs[2]['data']['query_result_count'] is None assert cloud_logs[2]['data']['query_duration_ms'] is None assert cloud_logs[2]['data']['failure_stage'] == 'business_validation' assert cloud_logs[2]['data']['error_type'] == 'BusinessValidationError' assert 'Traceback (most recent call last)' in cloud_logs[2]['data']['message'] assert '必须是数字' in cloud_logs[2]['data']['message'] assert cloud_logs[2]['message'] == cloud_logs[2]['data']['message'] assert any(level == 'error' and 'status=401' in message for level, message in local_logs) assert any(level == 'info' and 'status=200' in message for level, message in local_logs) assert all('duration_ms=' in message for _, message in local_logs) assert any(level == 'error' and '必须是数字' in message for level, message in local_logs) assert all('response=' not in message for _, message in local_logs) finally: await client.close() @pytest.mark.asyncio async def test_cloud_log_failure_does_not_change_api_response(monkeypatch): exception_logs = [] class FakeMySQL: async def fetch_all(self, sql, params): return [] class FakeLogger: def info(self, message): pass def error(self, message): pass def exception(self, message): exception_logs.append(message) class BrokenAliyunLogger: def logging_batch(self, events): raise RuntimeError('SLS unavailable') monkeypatch.setattr(settings, 'CHUI_ZHI_API_TOKEN', 'test-token') app = create_app() app.cleanup_ctx.clear() app[MYSQL_KEY] = FakeMySQL() app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(10) app[LOGGER_KEY] = FakeLogger() app[ALIYUN_LOGGER_KEY] = BrokenAliyunLogger() client = TestClient(TestServer(app)) await client.start_server() try: response = await client.post( API_PATH, json={}, headers={'X-API-Key': 'test-token'}, ) assert response.status == 200 assert (await response.json())['code'] == 0 assert any('阿里云API日志上报失败' in message for message in exception_logs) finally: await client.close() @pytest.mark.asyncio async def test_unsupported_parameters_return_specific_error_message(monkeypatch): class FakeMySQL: async def fetch_all(self, sql, params): raise AssertionError('参数校验失败时不应查询数据库') class FakeLogger: def info(self, message): pass def error(self, message): pass def exception(self, message): pass class FakeAliyunLogger: def logging_batch(self, events): pass monkeypatch.setattr(settings, 'CHUI_ZHI_API_TOKEN', 'test-token') app = create_app() app.cleanup_ctx.clear() app[MYSQL_KEY] = FakeMySQL() app[QUERY_SEMAPHORE_KEY] = asyncio.Semaphore(10) app[LOGGER_KEY] = FakeLogger() app[ALIYUN_LOGGER_KEY] = FakeAliyunLogger() client = TestClient(TestServer(app)) await client.start_server() try: unknown_parameter = await client.post( API_PATH, json={'unknown_parameter': 1}, headers={'X-API-Key': 'test-token'}, ) unknown_body = await unknown_parameter.json() assert unknown_parameter.status == 400 assert '不支持的参数: unknown_parameter' in unknown_body['msg'] unsupported_filter = await client.post( API_PATH, json={'filters': [{'field': 'unknown_field', 'operator': '>', 'value': 1}]}, headers={'X-API-Key': 'test-token'}, ) filter_body = await unsupported_filter.json() assert unsupported_filter.status == 400 assert 'filters.0.field不支持值 unknown_field' in filter_body['msg'] unknown_filter_key = await client.post( API_PATH, json={ 'filters': [ { 'field': 'like_cnt', 'operator': '>', 'value': 1, 'column': 'like_cnt', } ] }, headers={'X-API-Key': 'test-token'}, ) unknown_key_body = await unknown_filter_key.json() assert unknown_filter_key.status == 400 assert '不支持的参数: filters.0.column' in unknown_key_body['msg'] missing_filter_key = await client.post( API_PATH, json={'filters': [{'field': 'like_cnt', 'value': 1}]}, headers={'X-API-Key': 'test-token'}, ) missing_key_body = await missing_filter_key.json() assert missing_filter_key.status == 400 assert 'filters.0.operator不能为空' in missing_key_body['msg'] finally: await client.close() def test_api_routes_are_registered(): app = create_app() route_names = {route.name for route in app.router.routes()} assert 'chui_zhi_videos' in route_names assert {'health', 'ready'} <= route_names @pytest.mark.asyncio async def test_health_and_readiness_do_not_require_business_token(): cloud_logs = [] class FakeMySQL: async def fetch_one(self, sql): assert sql == 'SELECT 1 AS ok' return {'ok': 1} class FakeLogger: def info(self, message): pass def error(self, message): pass def exception(self, message): raise AssertionError(message) class FakeAliyunLogger: def logging_batch(self, events): cloud_logs.extend(events) app = create_app() app.cleanup_ctx.clear() app[MYSQL_KEY] = FakeMySQL() app[LOGGER_KEY] = FakeLogger() app[ALIYUN_LOGGER_KEY] = FakeAliyunLogger() client = TestClient(TestServer(app)) await client.start_server() try: health_response = await client.get('/health') ready_response = await client.get('/ready') assert health_response.status == 200 assert ready_response.status == 200 assert (await health_response.json())['data']['status'] == 'ok' assert (await ready_response.json())['data']['status'] == 'ready' assert len(cloud_logs) == 2 finally: await client.close()