dingdanquanliucheng/backend/scripts/smoke_check.py

from __future__ import annotations

from pathlib import Path
import sys

CURRENT_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = CURRENT_DIR.parent.parent
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from backend.app.core.error_codes import ErrorCode
from backend.app.core.exceptions import AppException
from backend.app.services.auth_service import auth_service
from backend.app.services.export_service import export_service
from backend.app.services.file_service import file_service
from backend.app.services.logistics_service import logistics_service
from backend.app.services.storage_service import storage_service
from backend.app.services.bootstrap import DEFAULT_ROLE_MENU_CODES


def main() -> None:
    checks: list[tuple[str, bool, str]] = []

    export_meta = export_service.build_performance_export(
        {
            "stat_type": "month",
            "list": [
                {
                    "stat_period": "2026-05",
                    "order_count": 1,
                    "order_amount": 100.0,
                    "commission_amount": 10.0,
                    "category_amounts": [{"category_id": 1, "category_name": "演示分类", "amount": 100.0}],
                }
            ],
        },
        "xlsx",
    )
    checks.append(("报表导出", Path(export_meta["file_path"]).exists(), export_meta["file_name"]))

    upload_meta = storage_service.create_upload_token("smoke", 1, "demo.png", 1024)
    checks.append(("OSS 上传签名", upload_meta["upload_url"].startswith("https://"), upload_meta["object_key"]))

    checks.append(("token 黑名单文件目录", auth_service.token_store_path.parent.exists(), str(auth_service.token_store_path)))
    checks.append(
        (
            "系统管理权限颗粒度",
            all(
                code in DEFAULT_ROLE_MENU_CODES["admin"]
                for code in (
                    "system:view",
                    "system:user:create",
                    "system:user:update",
                    "system:user:reset-password",
                    "system:role:create",
                    "system:role:update",
                    "system:role:assign-menus",
                    "system:menu:create",
                    "system:menu:update",
                )
            ),
            "admin 权限码已细化到写接口级别",
        )
    )
    checks.append(
        (
            "报表与AI权限颗粒度",
            all(
                code in DEFAULT_ROLE_MENU_CODES["manager"]
                for code in (
                    "report:performance:view",
                    "report:performance:export",
                    "ai:recognize",
                    "ai:correct",
                    "logistics:trace:list",
                    "logistics:trace:create",
                )
            ),
            "manager 已具备报表/AI/物流轨迹权限码",
        )
    )

    attachment_meta = file_service.build_attachment_payload(
        {
            "biz_type": "logistics_task",
            "biz_id": 1,
            "file_name": "pickup.jpg",
            "file_url": "https://example.com/logistics_task/1/pickup.jpg",
            "file_type": "image/jpeg",
            "file_size": None,
        },
        require_file_size=False,
    )
    checks.append(("司机附件留痕兼容无文件大小", attachment_meta["file_category"] == "image", attachment_meta["object_key"]))

    pickup_photo_required = False
    try:
        logistics_service.pickup_task(
            task_id=5001,
            payload={"photo_files": [], "video_files": [], "remark": "smoke"},
            session=None,
            current_user={"user_id": 10, "role_code": "driver"},
        )
    except AppException as exc:
        pickup_photo_required = exc.code == ErrorCode.PARAM_ERROR
    checks.append(("司机揽货照片必填", pickup_photo_required, "pickup requires photo files"))

    failed = [item for item in checks if not item[1]]
    for name, ok, detail in checks:
        print(f"[{'PASS' if ok else 'FAIL'}] {name}: {detail}")

    if failed:
        raise SystemExit(1)


if __name__ == "__main__":
    main()