test(stream-gears): pytest smoke suite for the wheel in CI (#1723)

* test(stream-gears): pytest smoke suite for the wheel in CI

Offline checks against the installed wheel: exported names, UploadLine
members, PySegment/config_bindings basics, `biliup --version` and
`python -m biliup --help` in a subprocess. OPS-19 (login functions
require `proxy`) is recorded as a strict xfail.

The linux x86_64 job installs the abi3 wheel into a venv and runs the
suite before the free-threaded build.

Ref #1721

* chore(stream-gears): move manual scripts to examples/

Drop test_server.py (only printed the default config; now covered by
the smoke suite). Move login.py, test_upload.py and test_download.py to
examples/ so pytest never collects them, and take the video/cookie
paths and stream URL from argv/env instead of hard-coded E:\ paths.
Also pass proxy explicitly (OPS-19), use PySegment (OPS-20) and catch
ValueError from str.index (OPS-38) so the scripts run on this version.

Ref #1721

* test(stream-gears): drop internal bug id from xfail reason

.github/workflows/ci.yml

@@ -60,6 +60,12 @@ jobs:
           args: --release --out dist
           sccache: ${{ !startsWith(github.ref, 'refs/tags/') }}
           manylinux: ${{ matrix.platform.target == 'aarch64' && '2_28' || 'auto' }}
+      - name: Smoke test wheel
+        if: matrix.platform.target == 'x86_64'
+        run: |
+          python -m venv /tmp/smoke-venv
+          /tmp/smoke-venv/bin/pip install dist/biliup-*-abi3-*.whl pytest
+          /tmp/smoke-venv/bin/pytest crates/stream-gears/tests -v
       - name: Build free-threaded wheels
         uses: PyO3/maturin-action@v1
         with:

crates/stream-gears/examples/download.py

@@ -1,21 +1,20 @@
+"""下载一路直播流，按时长/大小分段。
+
+用法：python download.py <直播流 URL>
+"""
 import stream_gears
 import sys
 import multiprocessing
 import time
 
 
-class Segment:
-    pass
-
-
-def download_worker():
+def download_worker(url):
     """在子进程中执行下载任务"""
     try:
-        segment = Segment()
+        segment = stream_gears.PySegment()
         segment.time = 60 * 1 # secs(1 min)
         # segment.size = 6000 * 1024 * 1024
         segment.size = 1000 * 1000 * 10 # bytes(10MB)
-        url = ""
 
         stream_gears.download(
             url = url,
@@ -33,12 +32,12 @@ def download_worker():
         return False
 
 
-def download_with_signal_handling():
+def download_with_signal_handling(url):
     """使用多进程包装下载函数，支持 Ctrl+C 中断"""
     print("开始下载，按 Ctrl+C 可以退出...")
 
     # 创建下载进程
-    download_process = multiprocessing.Process(target=download_worker)
+    download_process = multiprocessing.Process(target=download_worker, args=(url,))
     download_process.start()
 
     try:
@@ -70,4 +69,6 @@ def download_with_signal_handling():
 if __name__ == '__main__':
     # Windows 上需要这个来支持 multiprocessing
     multiprocessing.freeze_support()
-    download_with_signal_handling()
+    if len(sys.argv) != 2:
+        sys.exit(__doc__)
+    download_with_signal_handling(sys.argv[1])

crates/stream-gears/examples/login.py

@@ -0,0 +1,16 @@
+"""扫码登录，把凭据写到 `<mid>.json`（或第一个参数指定的路径）。
+
+用法：python login.py [输出文件]
+"""
+import json
+import sys
+
+import stream_gears
+if __name__ == '__main__':
+    res = stream_gears.get_qrcode(proxy=None)
+    print(res)
+    res = stream_gears.login_by_qrcode(res, proxy=None)
+    out = sys.argv[1] if len(sys.argv) > 1 else f'{json.loads(res)["token_info"]["mid"]}.json'
+    with open(out, 'w', encoding='utf-8') as file:
+        file.write(res)
+    print(res)

crates/stream-gears/examples/upload.py

@@ -1,27 +1,36 @@
 """
-stream_gears 上传功能测试
+stream_gears 上传示例
 
-这个测试文件展示了如何使用 stream_gears.upload() 函数的各种功能：
+这个示例展示了如何使用 stream_gears.upload() 函数的各种功能：
 1. 基础上传 - 使用关键字参数让代码更清晰
 2. 多文件上传 - 上传多P视频
 3. 定时发布 - 设置未来时间发布
 4. 高级功能 - 使用富文本简介和额外字段
 5. 最小参数 - 只使用必需参数，其他使用默认值
 
+用法：python upload.py 视频1 [视频2 ...]
+cookie 文件默认 cookies.json，可用环境变量 BILIUP_COOKIE_FILE 指定。
+
 注意：运行前请确保：
 - 存在有效的 cookies.json 文件
 - 视频文件路径正确
 - 网络连接正常
 """
 
+import os
+import sys
+
 import stream_gears
 from typing import List, Dict, Any
 
-video_path = ["E:\\Projects\\biliup\\target\\x86_64-pc-windows-msvc\\debug\\test.flv"]
-cookie_file = "E:\\Projects\\biliup\\target\\x86_64-pc-windows-msvc\\debug\\cookies.json"
+video_path = sys.argv[1:]
+cookie_file = os.environ.get("BILIUP_COOKIE_FILE", "cookies.json")
 # submit_api = "BCutAndroid"
 
 if __name__ == '__main__':
+    if not video_path:
+        sys.exit(__doc__)
+
     # full kwargs 上传测试
     # stream_gears.upload(
     #     video_path=["examples/test.mp4"],
@@ -70,7 +79,7 @@ if __name__ == '__main__':
                 desc = desc.replace(
                     "@credit", "@" + credit["username"] + "  ", 1)
                 desc_v2_tmp = desc_v2_tmp[num + 7:]
-            except IndexError:
+            except ValueError:
                 print('简介中的@credit占位符少于credits的数量,替换失败')
         desc_v2.append({
             "raw_text": str(desc_v2_tmp),

crates/stream-gears/tests/login.py

@@ -1,10 +0,0 @@
-import json
-
-import stream_gears
-if __name__ == '__main__':
-    res = stream_gears.get_qrcode()
-    print(res)
-    res = stream_gears.login_by_qrcode(res)
-    with open(f'{json.loads(res)["token_info"]["mid"]}.json', 'w', encoding='utf-8') as file:
-        file.write(res)
-    print(res)

crates/stream-gears/tests/test_server.py

@@ -1,17 +0,0 @@
-import asyncio
-
-import stream_gears
-
-async def main():
-    config = stream_gears.config_bindings()
-    print(config)
-    print(config.get("twitcasting_password", "313213"))
-    print(config.get("file_size"))
-    print(config.get("file_size"))
-    print(config.get("streamers"))
-    # await loop.run_in_executor(None, stream_gears.main_loop)
-    # stream_gears.main_loop()
-
-
-if __name__ == '__main__':
-    asyncio.run(main())

crates/stream-gears/tests/test_smoke.py

@@ -0,0 +1,136 @@
+"""Smoke tests for the installed biliup wheel.
+
+Run against an installed wheel (``pip install dist/biliup-*.whl``), not the
+source tree: ``pytest crates/stream-gears/tests``. All tests are offline.
+"""
+
+import importlib.metadata
+import inspect
+import subprocess
+import sys
+import sysconfig
+from pathlib import Path
+
+import pytest
+import stream_gears
+
+EXPORTS = {
+    "PySegment",
+    "UploadLine",
+    "config_bindings",
+    "download",
+    "download_with_callback",
+    "get_qrcode",
+    "login_by_cookies",
+    "login_by_qrcode",
+    "login_by_sms",
+    "login_by_web_cookies",
+    "login_by_web_qrcode",
+    "main_loop",
+    "send_sms",
+    "stream_gears",
+    "upload",
+}
+
+UPLOAD_LINES = [
+    "Bldsa",
+    "Cnbldsa",
+    "Andsa",
+    "Atdsa",
+    "Bda2",
+    "Cnbd",
+    "Anbd",
+    "Atbd",
+    "Tx",
+    "Cntx",
+    "Antx",
+    "Attx",
+    "Txa",
+    "Alia",
+    "Estx",
+    "Akbd",
+]
+
+LOGIN_FUNCTIONS = [
+    "get_qrcode",
+    "login_by_cookies",
+    "login_by_qrcode",
+    "login_by_sms",
+    "login_by_web_cookies",
+    "login_by_web_qrcode",
+    "send_sms",
+]
+
+# The CLI exits the process (clap's `--help` / `--version`), so it must run in
+# a subprocess. cwd is a temp dir so the repo's `biliup/` source package can't
+# shadow the installed one.
+TIMEOUT = 60
+
+
+def run(args, cwd):
+    return subprocess.run(
+        args, cwd=cwd, capture_output=True, text=True, timeout=TIMEOUT
+    )
+
+
+def biliup_script():
+    name = "biliup.exe" if sys.platform == "win32" else "biliup"
+    path = Path(sysconfig.get_path("scripts")) / name
+    assert path.is_file(), f"console script not installed: {path}"
+    return str(path)
+
+
+def test_exports():
+    public = {name for name in dir(stream_gears) if not name.startswith("_")}
+    assert public == EXPORTS
+
+
+def test_upload_line_members():
+    members = [
+        name
+        for name in dir(stream_gears.UploadLine)
+        if isinstance(getattr(stream_gears.UploadLine, name), stream_gears.UploadLine)
+    ]
+    assert sorted(members) == sorted(UPLOAD_LINES)
+    assert len(members) == 16
+
+
+def test_py_segment():
+    # Fields are write-only (`#[pyclass(set_all)]` without `get_all`).
+    segment = stream_gears.PySegment()
+    segment.time = 60
+    segment.size = 10 * 1000 * 1000
+    with pytest.raises(TypeError):
+        segment.time = "60"
+
+
+def test_config_bindings():
+    config = stream_gears.config_bindings()
+    assert config.get("streamers") == {}
+    assert config.get("no_such_key", "fallback") == "fallback"
+    with pytest.raises(AttributeError):
+        config.get("no_such_key")
+
+
+@pytest.mark.parametrize("name", LOGIN_FUNCTIONS)
+@pytest.mark.xfail(
+    strict=True,
+    raises=AssertionError,
+    reason="PyO3 >= 0.24 no longer defaults trailing Option args to None; login functions need proxy=None",
+)
+def test_login_proxy_is_optional(name):
+    params = inspect.signature(getattr(stream_gears, name)).parameters
+    assert params["proxy"].default is None
+
+
+def test_biliup_version(tmp_path):
+    result = run([biliup_script(), "--version"], tmp_path)
+    assert result.returncode == 0, result.stderr
+    assert importlib.metadata.version("biliup") in result.stdout
+
+
+def test_python_m_biliup_help(tmp_path):
+    result = run([sys.executable, "-m", "biliup", "--help"], tmp_path)
+    assert result.returncode == 0, result.stderr
+    assert "Usage:" in result.stdout
+    assert "server" in result.stdout