AutoScraperX/test/test_video_query_api.py

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()