Pine
31f8703595
- scripts/db/seed.py：培训种子与 accounts→users 迁移改经 training 数据层（双方言） - scripts/db/seed_hall_community.py：人才/岗位/服务/帖子演示数据（幂等，离线执行） - tests/test_hall.py：人才档案/帖子点赞评论计数/register 报名状态机覆盖
106 lines
4.3 KiB
Python
106 lines
4.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""大厅与社区自动化验证（0035）。
|
|
|
|
覆盖：人才档案 upsert / 帖子点赞评论计数 / TaskService.register 报名状态机
|
|
（报名成功、重复拒绝、名额上限、选定转 assigned）。
|
|
临时 SQLite 隔离，不依赖外部库。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from app.infrastructure.hall_repositories import (
|
|
CommunityCommentRepository,
|
|
CommunityLikeRepository,
|
|
CommunityPostRepository,
|
|
TalentProfileRepository,
|
|
)
|
|
from app.infrastructure.models import Task
|
|
from app.infrastructure.repositories import Database
|
|
from app.services.task_service import TaskService
|
|
|
|
|
|
@pytest_asyncio.fixture()
|
|
async def db(tmp_path, monkeypatch):
|
|
url = f"sqlite+aiosqlite:///{tmp_path}/hall.db"
|
|
d = Database(db_url=url)
|
|
from app.infrastructure.db import Base
|
|
async with d._engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield d
|
|
await d.close()
|
|
|
|
|
|
async def _mk_task(db: Database, **kw) -> dict:
|
|
t = Task(id="task_hall_1", title="报名测试任务", mode="register", status="published",
|
|
register_quota=kw.pop("register_quota", 2), c_visible=True,
|
|
publisher_name="测试发包方")
|
|
for k, v in kw.items():
|
|
setattr(t, k, v)
|
|
db.session.add(t)
|
|
await db.session.commit()
|
|
return {"id": t.id, "status": t.status, "mode": t.mode, "register_quota": t.register_quota,
|
|
"visibility": "public", "park_id": "", "assign_type": "", "eligibility": "{}"}
|
|
|
|
|
|
ACTOR = {"id": "u_actor", "username": "actor", "nickname": "接单人"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talent_profile_upsert_and_list(db):
|
|
repo = TalentProfileRepository(db.session)
|
|
await repo.upsert("u_t1", {"displayName": "阿岩", "talentType": "opc",
|
|
"headline": "AI 开发", "fields": ["AI应用"], "skills": ["Python"]})
|
|
got = await repo.get("u_t1")
|
|
assert got["displayName"] == "阿岩" and got["fields"] == ["AI应用"]
|
|
rows = await repo.list(q="AI")
|
|
assert any(r["userId"] == "u_t1" for r in rows)
|
|
# 下架后列表不可见
|
|
await repo.set_published("u_t1", False)
|
|
assert not any(r["userId"] == "u_t1" for r in await repo.list())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_community_post_like_comment_counts(db):
|
|
posts = CommunityPostRepository(db.session)
|
|
likes = CommunityLikeRepository(db.session)
|
|
cmts = CommunityCommentRepository(db.session)
|
|
p = await posts.create(author_id="u_a", author_name="作者", title="第一帖",
|
|
content="内容", topic="交流", status="published")
|
|
pid = p["id"]
|
|
await likes.add("post", pid, "u1")
|
|
await likes.add("post", pid, "u2")
|
|
await posts.inc_likes(pid, 2)
|
|
assert (await posts.get(pid))["likeCount"] == 2
|
|
await likes.remove("post", pid, "u1")
|
|
await posts.inc_likes(pid, -1)
|
|
assert (await posts.get(pid))["likeCount"] == 1
|
|
await cmts.create(pid, "u2", "评论人", "赞一个")
|
|
await posts.inc_comments(pid, 1)
|
|
assert (await posts.get(pid))["commentCount"] == 1
|
|
assert len(await cmts.list_for_post(pid)) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_flow(db):
|
|
svc = TaskService(db)
|
|
task = await _mk_task(db, register_quota=2)
|
|
claim = await svc.register(task["id"], ACTOR)
|
|
assert claim["status"] == "registered" and claim["claim_source"] == "register"
|
|
with pytest.raises(Exception) as e:
|
|
await svc.register(task["id"], ACTOR)
|
|
assert getattr(e.value, "status_code", 0) == 400 # 重复报名拒绝
|
|
other = {"id": "u_b", "username": "b", "nickname": "乙"}
|
|
await svc.register(task["id"], other)
|
|
third = {"id": "u_c", "username": "c", "nickname": "丙"}
|
|
with pytest.raises(Exception) as e2:
|
|
await svc.register(task["id"], third)
|
|
assert getattr(e2.value, "status_code", 0) == 400 # 名额已满
|
|
# 选定乙 → assigned + 任务 claimed，其余 withdrawn
|
|
claims = {c["claimer_user_id"]: c for c in await db.task_claims.list_by_task(task["id"])}
|
|
await svc.register_select(task["id"], claims["u_b"]["id"], ACTOR)
|
|
assert (await db.task_claims.get(claims["u_b"]["id"]))["status"] == "assigned"
|
|
assert (await db.task_claims.get(claims["u_actor"]["id"]))["status"] == "withdrawn"
|
|
assert (await db.tasks.get(task["id"]))["status"] == "claimed"
|