feat(init): 完成 Phase 1 基础架构搭建
- 完成 T-001A: 前端项目初始化 (Next.js 14 + TypeScript + Tailwind CSS) - 完成 T-001B: 后端项目初始化 (FastAPI + SQLAlchemy + asyncpg) - 完成 T-002: 数据库配置 (KolVideo 模型 + 索引 + 测试) - 完成 T-003: 基础 UI 框架 (Header/Footer 组件 + 品牌色系) - 完成 T-004: 环境变量配置 (前后端环境变量) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
zfc
@@ -0,0 +1,58 @@
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||||
|
||||
from app.database import Base
|
||||
from app.models import KolVideo
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_video_data():
|
||||
"""Sample video data for testing."""
|
||||
return {
|
||||
"item_id": "test_item_001",
|
||||
"title": "测试视频标题",
|
||||
"viral_type": "爆款",
|
||||
"video_url": "https://example.com/video/001",
|
||||
"star_id": "star_001",
|
||||
"star_unique_id": "unique_001",
|
||||
"star_nickname": "测试达人",
|
||||
"natural_play_cnt": 100000,
|
||||
"heated_play_cnt": 50000,
|
||||
"total_play_cnt": 150000,
|
||||
"total_interact": 5000,
|
||||
"like_cnt": 3000,
|
||||
"share_cnt": 1000,
|
||||
"comment_cnt": 1000,
|
||||
"new_a3_rate": 0.05,
|
||||
"after_view_search_uv": 500,
|
||||
"return_search_cnt": 200,
|
||||
"industry_id": "ind_001",
|
||||
"industry_name": "美妆",
|
||||
"brand_id": "brand_001",
|
||||
"estimated_video_cost": 10000.0,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def test_engine():
|
||||
"""Create a test database engine using SQLite."""
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
echo=False,
|
||||
)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield engine
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def test_session(test_engine):
|
||||
"""Create a test database session."""
|
||||
async_session = async_sessionmaker(
|
||||
test_engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False,
|
||||
)
|
||||
async with async_session() as session:
|
||||
yield session
|
||||
@@ -0,0 +1,165 @@
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.models import KolVideo
|
||||
|
||||
|
||||
class TestKolVideoModel:
|
||||
"""Tests for KolVideo model."""
|
||||
|
||||
async def test_create_video(self, test_session, sample_video_data):
|
||||
"""Test creating a video record."""
|
||||
video = KolVideo(**sample_video_data)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == sample_video_data["item_id"])
|
||||
)
|
||||
saved_video = result.scalar_one()
|
||||
|
||||
assert saved_video.item_id == sample_video_data["item_id"]
|
||||
assert saved_video.title == sample_video_data["title"]
|
||||
assert saved_video.star_id == sample_video_data["star_id"]
|
||||
|
||||
async def test_query_by_star_id(self, test_session, sample_video_data):
|
||||
"""Test querying videos by star_id."""
|
||||
video = KolVideo(**sample_video_data)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.star_id == sample_video_data["star_id"])
|
||||
)
|
||||
videos = result.scalars().all()
|
||||
|
||||
assert len(videos) == 1
|
||||
assert videos[0].star_id == sample_video_data["star_id"]
|
||||
|
||||
async def test_query_by_star_unique_id(self, test_session, sample_video_data):
|
||||
"""Test querying videos by star_unique_id."""
|
||||
video = KolVideo(**sample_video_data)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(
|
||||
KolVideo.star_unique_id == sample_video_data["star_unique_id"]
|
||||
)
|
||||
)
|
||||
videos = result.scalars().all()
|
||||
|
||||
assert len(videos) == 1
|
||||
assert videos[0].star_unique_id == sample_video_data["star_unique_id"]
|
||||
|
||||
async def test_query_by_nickname_like(self, test_session, sample_video_data):
|
||||
"""Test querying videos by nickname using LIKE."""
|
||||
video = KolVideo(**sample_video_data)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.star_nickname.like("%测试%"))
|
||||
)
|
||||
videos = result.scalars().all()
|
||||
|
||||
assert len(videos) == 1
|
||||
assert "测试" in videos[0].star_nickname
|
||||
|
||||
async def test_batch_query_by_star_ids(self, test_session, sample_video_data):
|
||||
"""Test batch querying videos by multiple star_ids."""
|
||||
video1 = KolVideo(**sample_video_data)
|
||||
video2_data = sample_video_data.copy()
|
||||
video2_data["item_id"] = "test_item_002"
|
||||
video2_data["star_id"] = "star_002"
|
||||
video2 = KolVideo(**video2_data)
|
||||
|
||||
test_session.add_all([video1, video2])
|
||||
await test_session.commit()
|
||||
|
||||
star_ids = ["star_001", "star_002"]
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.star_id.in_(star_ids))
|
||||
)
|
||||
videos = result.scalars().all()
|
||||
|
||||
assert len(videos) == 2
|
||||
|
||||
async def test_video_default_values(self, test_session):
|
||||
"""Test that default values are set correctly."""
|
||||
video = KolVideo(
|
||||
item_id="test_defaults",
|
||||
star_id="star_test",
|
||||
star_unique_id="unique_test",
|
||||
star_nickname="测试默认值",
|
||||
)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == "test_defaults")
|
||||
)
|
||||
saved_video = result.scalar_one()
|
||||
|
||||
assert saved_video.natural_play_cnt == 0
|
||||
assert saved_video.heated_play_cnt == 0
|
||||
assert saved_video.total_play_cnt == 0
|
||||
assert saved_video.estimated_video_cost == 0
|
||||
|
||||
async def test_video_nullable_fields(self, test_session):
|
||||
"""Test that nullable fields can be None."""
|
||||
video = KolVideo(
|
||||
item_id="test_nullable",
|
||||
star_id="star_nullable",
|
||||
star_unique_id="unique_nullable",
|
||||
star_nickname="测试可空字段",
|
||||
)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == "test_nullable")
|
||||
)
|
||||
saved_video = result.scalar_one()
|
||||
|
||||
assert saved_video.title is None
|
||||
assert saved_video.video_url is None
|
||||
assert saved_video.brand_id is None
|
||||
assert saved_video.new_a3_rate is None
|
||||
|
||||
async def test_update_video(self, test_session, sample_video_data):
|
||||
"""Test updating a video record."""
|
||||
video = KolVideo(**sample_video_data)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == sample_video_data["item_id"])
|
||||
)
|
||||
saved_video = result.scalar_one()
|
||||
saved_video.title = "更新后的标题"
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == sample_video_data["item_id"])
|
||||
)
|
||||
updated_video = result.scalar_one()
|
||||
assert updated_video.title == "更新后的标题"
|
||||
|
||||
async def test_delete_video(self, test_session, sample_video_data):
|
||||
"""Test deleting a video record."""
|
||||
video = KolVideo(**sample_video_data)
|
||||
test_session.add(video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == sample_video_data["item_id"])
|
||||
)
|
||||
saved_video = result.scalar_one()
|
||||
await test_session.delete(saved_video)
|
||||
await test_session.commit()
|
||||
|
||||
result = await test_session.execute(
|
||||
select(KolVideo).where(KolVideo.item_id == sample_video_data["item_id"])
|
||||
)
|
||||
assert result.scalar_one_or_none() is None
|
||||
Reference in New Issue
Block a user