AutoScraperX/test/test_fastapi_video_query_api.py

import asyncio

from fastapi.testclient import TestClient

from api.base import BaseApi
from api.chui_zhi.videos import VideoQueryApi
from api.fastapi_app import API_PATH, create_app, print_api_routes


class FakeLogger:
    def __init__(self):
        self.records = []

    def info(self, message):
        self.records.append(('info', message))

    def error(self, message):
        self.records.append(('error', message))

    def exception(self, message):
        self.records.append(('exception', message))


class FakeAliyunLogger:
    def __init__(self):
        self.events = []

    def logging_batch(self, events):
        self.events.extend(events)


def build_test_app(mysql):
    app = create_app(manage_resources=False)
    app.state.mysql = mysql
    app.state.query_semaphore = asyncio.Semaphore(10)
    app.state.logger = FakeLogger()
    app.state.aliyun_logger = FakeAliyunLogger()
    return app


def test_fastapi_query_keeps_existing_contract_and_unified_logging():
    assert issubclass(VideoQueryApi, BaseApi)

    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'},
            ]

    app = build_test_app(FakeMySQL())
    payload = {
        'platforms': ['xiaoniangao'],
        'filters': [
            {'field': 'create_time', 'operator': '>=', 'value': 1785772800000},
            {'field': 'create_time', 'operator': '<', 'value': 1785859200000},
            {'field': 'like_cnt', 'operator': '>', 'value': 3},
        ],
        'limit': 1,
    }
    with TestClient(app, raise_server_exceptions=False) as client:
        response = client.post(
            API_PATH,
            json=payload,
            headers={'X-Request-ID': 'scheduler-request-1'},
        )

    body = response.json()
    assert response.status_code == 200
    assert response.headers['X-Request-ID'] == 'scheduler-request-1'
    assert response.headers['Cache-Control'] == 'no-store'
    assert body['code'] == 0
    assert body['data']['count'] == 1
    assert body['data']['has_more'] is True
    assert body['data']['next_cursor'] == {'id': 1}
    event = app.state.aliyun_logger.events[0]
    assert event['trace_id'] == 'scheduler-request-1'
    assert event['data']['request_params'] == payload
    assert event['data']['status_code'] == 200
    assert event['data']['query_result_count'] == 2


def test_default_time_range_is_pinned_in_next_cursor():
    class FakeMySQL:
        async def fetch_all(self, sql, params):
            return [{'id': 2}, {'id': 1}]

    app = build_test_app(FakeMySQL())
    with TestClient(app, raise_server_exceptions=False) as client:
        first_page = client.post(API_PATH, json={'limit': 1}).json()['data']
        cursor = first_page['next_cursor']
        second_page = client.post(API_PATH, json={'limit': 1, 'cursor': cursor}).json()['data']

    assert cursor['id'] == 2
    assert isinstance(cursor['query_time'], int)
    assert second_page['next_cursor']['query_time'] == cursor['query_time']


def test_fastapi_validation_and_business_errors_use_same_exit_contract():
    class FakeMySQL:
        async def fetch_all(self, sql, params):
            raise AssertionError('参数错误时不应查询数据库')

    app = build_test_app(FakeMySQL())
    with TestClient(app, raise_server_exceptions=False) as client:
        validation = client.post(API_PATH, json={'unknown_parameter': 1})
        business = client.post(
            API_PATH,
            json={'filters': [{'field': 'like_cnt', 'operator': '>', 'value': 'invalid'}]},
        )
        wrong_method = client.get(API_PATH)

    assert validation.status_code == 400
    assert '不支持的参数: unknown_parameter' in validation.json()['msg']
    assert business.status_code == 422
    assert '必须是数字' in business.json()['msg']
    assert wrong_method.status_code == 405
    assert wrong_method.json() == {'code': 405, 'msg': 'Method Not Allowed', 'data': None}
    assert len(app.state.aliyun_logger.events) == 3
    business_event = app.state.aliyun_logger.events[1]
    assert business_event['data']['failure_stage'] == 'business_validation'
    assert business_event['data']['error_type'] == 'BusinessValidationError'
    assert 'Traceback (most recent call last)' in business_event['data']['message']
    assert '必须是数字' in business_event['data']['message']
    assert all('X-Request-ID' in response.headers for response in (validation, business, wrong_method))


def test_fastapi_health_readiness_and_documentation_routes():
    class FakeMySQL:
        async def fetch_one(self, sql):
            assert sql == 'SELECT 1 AS ok'
            return {'ok': 1}

    app = build_test_app(FakeMySQL())
    with TestClient(app, raise_server_exceptions=False) as client:
        health = client.get('/health')
        ready = client.get('/ready')
        docs = client.get('/openapi.json')

    assert health.json()['data']['status'] == 'ok'
    assert ready.json()['data']['status'] == 'ready'
    assert API_PATH in docs.json()['paths']


def test_startup_route_list_contains_business_source_location(capsys):
    app = create_app(manage_resources=False)

    print_api_routes(app)

    output = capsys.readouterr().out
    assert 'api.chui_zhi.videos.VideoQueryApi.logic' in output
    assert 'api/chui_zhi/videos.py:' in output
    assert 'api.fastapi_app.health' in output
    assert 'api/fastapi_app.py:' in output


def test_fastapi_sls_failure_does_not_change_business_response():
    class FakeMySQL:
        async def fetch_all(self, sql, params):
            return []

    class BrokenAliyunLogger:
        def logging_batch(self, events):
            raise RuntimeError('SLS unavailable')

    app = build_test_app(FakeMySQL())
    app.state.aliyun_logger = BrokenAliyunLogger()
    with TestClient(app, raise_server_exceptions=False) as client:
        response = client.post(API_PATH, json={})

    assert response.status_code == 200
    assert response.json()['code'] == 0
    assert any('阿里云API日志上报失败' in message for _, message in app.state.logger.records)