ai-swap/backend/tests/api/test_auto_post_api.py

import os
import json
import pytest
from datetime import datetime, timedelta
from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock

# 导入被测试的API
import sys
current_dir = os.path.dirname(os.path.abspath(__file__))
backend_dir = os.path.dirname(os.path.dirname(current_dir))
if backend_dir not in sys.path:
    sys.path.insert(0, backend_dir)

from api.auto_post_api import router
from fastapi import FastAPI

# 创建测试应用
app = FastAPI()
app.include_router(router)
client = TestClient(app)

# 测试数据
TEST_IMAGE_DIR = os.path.join(os.path.dirname(__file__), "test_data")
os.makedirs(TEST_IMAGE_DIR, exist_ok=True)

def create_test_image(filename, content=b"fake image content"):
    """创建测试图片文件"""
    filepath = os.path.join(TEST_IMAGE_DIR, filename)
    with open(filepath, "wb") as f:
        f.write(content)
    return filepath

@pytest.fixture(autouse=True)
def setup_teardown():
    """测试前创建必要的目录和文件，测试后清理"""
    # 设置
    os.makedirs("temp", exist_ok=True)
    test_image1 = create_test_image("flower.jpg")
    test_image2 = create_test_image("test_img.png")
    
    yield
    
    # 清理
    import shutil
    if os.path.exists("temp"):
        shutil.rmtree("temp")
    if os.path.exists(TEST_IMAGE_DIR):
        shutil.rmtree(TEST_IMAGE_DIR)

def get_test_data(future_time=None):
    """获取测试数据"""
    if future_time is None:
        future_time = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d %H:%M")
    
    return {
        "title": "💙被问爆的蓝色仙女裙，美到犯规！",
        "description": "👗宝子们，挖到一条超绝的蓝色长裙！温柔的浅蓝色，仿佛把天空穿在了身上~V领设计巧妙修饰颈部线条，增添了一丝小性感。泡泡袖又带着点复古甜美感，谁穿谁是在逃公主！",
        "topics": json.dumps(["每日穿搭", "今日分享", "蓝色长裙", "仙女裙", "复古甜美", "小性感", "氛围感", "闭眼入不亏"]),
        "schedule_time": future_time
    }

@patch("api.auto_post_api.AutoPostService")
def test_successful_post(mock_service):
    """测试成功发帖场景"""
    # 模拟服务层返回
    mock_instance = mock_service.return_value
    mock_instance.post_to_xiaohongshu.return_value = {"status": "success"}
    
    # 准备测试数据
    with open(os.path.join(TEST_IMAGE_DIR, "flower.jpg"), "rb") as f1, \
         open(os.path.join(TEST_IMAGE_DIR, "test_img.png"), "rb") as f2:
        files = [
            ("files", ("flower.jpg", f1, "image/jpeg")),
            ("files", ("test_img.png", f2, "image/png"))
        ]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=get_test_data()
        )
    
    # 验证响应
    assert response.status_code == 200
    assert response.json()["status"] == "success"
    
    # 验证服务层调用
    mock_instance.post_to_xiaohongshu.assert_called_once()

def test_invalid_schedule_time():
    """测试无效的定时发布时间"""
    past_time = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d %H:%M")
    data = get_test_data(past_time)
    
    with open(os.path.join(TEST_IMAGE_DIR, "flower.jpg"), "rb") as f1:
        files = [("files", ("flower.jpg", f1, "image/jpeg"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=data
        )
    
    assert response.status_code == 422
    assert "定时发布时间不能早于当前时间" in response.text

def test_invalid_topics_format():
    """测试无效的话题格式"""
    data = get_test_data()
    data["topics"] = "invalid json"
    
    with open(os.path.join(TEST_IMAGE_DIR, "flower.jpg"), "rb") as f1:
        files = [("files", ("flower.jpg", f1, "image/jpeg"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=data
        )
    
    assert response.status_code == 400
    assert "话题必须是数组格式" in response.text

def test_too_many_topics():
    """测试超过话题数量限制"""
    data = get_test_data()
    # 创建21个话题（超过20个的限制）
    data["topics"] = json.dumps([f"话题{i}" for i in range(21)])
    
    with open(os.path.join(TEST_IMAGE_DIR, "flower.jpg"), "rb") as f1:
        files = [("files", ("flower.jpg", f1, "image/jpeg"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=data
        )
    
    assert response.status_code == 422
    assert "topics" in response.text

def test_too_many_images():
    """测试超过最大图片数量限制"""
    # 创建10张测试图片（超过9张的限制）
    files = []
    for i in range(10):
        create_test_image(f"test{i}.jpg")
        with open(os.path.join(TEST_IMAGE_DIR, f"test{i}.jpg"), "rb") as f:
            files.append(("files", (f"test{i}.jpg", f, "image/jpeg")))
    
    response = client.post(
        "/xiaohongshu/post",
        files=files,
        data=get_test_data()
    )
    
    assert response.status_code == 422
    assert "files" in response.text

def test_missing_required_fields():
    """测试缺少必填字段"""
    data = get_test_data()
    del data["title"]  # 删除必填字段
    
    with open(os.path.join(TEST_IMAGE_DIR, "flower.jpg"), "rb") as f1:
        files = [("files", ("flower.jpg", f1, "image/jpeg"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=data
        )
    
    assert response.status_code == 422
    assert "title" in response.text

@patch("api.auto_post_api.AutoPostService")
def test_service_error_handling(mock_service):
    """测试服务层错误处理"""
    # 模拟服务层抛出异常
    mock_instance = mock_service.return_value
    mock_instance.post_to_xiaohongshu.side_effect = Exception("发布失败")
    
    with open(os.path.join(TEST_IMAGE_DIR, "flower.jpg"), "rb") as f1:
        files = [("files", ("flower.jpg", f1, "image/jpeg"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=get_test_data()
        )
    
    assert response.status_code == 500
    assert "Failed to post to Xiaohongshu" in response.text

def test_empty_image():
    """测试空图片文件"""
    create_test_image("empty.jpg", content=b"")
    data = get_test_data()
    
    with open(os.path.join(TEST_IMAGE_DIR, "empty.jpg"), "rb") as f:
        files = [("files", ("empty.jpg", f, "image/jpeg"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=data
        )
    
    assert response.status_code == 400
    assert "empty file" in response.text.lower()

def test_invalid_image_format():
    """测试无效的图片格式"""
    create_test_image("invalid.txt", content=b"this is not an image")
    data = get_test_data()
    
    with open(os.path.join(TEST_IMAGE_DIR, "invalid.txt"), "rb") as f:
        files = [("files", ("invalid.txt", f, "text/plain"))]
        response = client.post(
            "/xiaohongshu/post",
            files=files,
            data=data
        )
    
    assert response.status_code == 400
    assert "image format" in response.text.lower()