Files
yunke/yunke-server/tests/test_smoke_script.py
T
浅念 8dc8caa749 chore: wechat end-to-end smoke script
Wires the existing pieces together into a runnable integration check:
scrape ~10 WeChat friends + ~50 of their messages (last 7 days) from the
real yunke CRM (read-only), upsert into a local SQLite smoke DB, then
print a summary that verifies rows landed.

Two auth modes:
* Existing-session (preferred): YK_USERID/YK_COMPANY[/YK_COOKIE] -> YunkeClient
  constructed from a captured session, login bypassed.
* Login mode (Phase-0 blocked): YUNKE_USERNAME/YUNKE_PASSWORD -> attempts
  client.login(), which the real CRM will reject until the placeholder
  password encryption is replaced. Prints a clear message and exits 3.

build_client_from_env() is split out as a pure, network-free helper so
auth-mode selection is unit-testable. main() is async (asyncio.run in
__main__). HTTPStatusError from a 401/expired session is wrapped into a
helpful message + exit 4 (no traceback). client.aclose() + engine.dispose()
always run in finally.

Defaults to SMOKE_DB_URL=sqlite+aiosqlite:///./smoke.db so no MySQL is
required. Run from yunke-server/ with the venv active:

    python -m scripts.smoke_wechat

Tests cover the entrypoint shape and auth-mode selection (no network).
Suite: 77 -> 87 passing.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-01 12:16:02 +08:00

165 lines
5.6 KiB
Python

"""Light unit tests for the end-to-end smoke script (Task 1.9).
These DO NOT hit the network — they verify the testable surface of
:mod:`scripts.smoke_wechat`:
* the module exposes an async ``main`` entrypoint,
* :func:`build_client_from_env` picks the right auth mode from env vars,
* :func:`_parse_cookie_env` accepts both raw-header and JSON cookie formats.
The full :func:`main` flow is an integration check against the live CRM and is
exercised manually (see the module docstring of ``scripts.smoke_wechat``).
"""
from __future__ import annotations
import asyncio
from app.session.yunke_client import YunkeClient
from scripts.smoke_wechat import _parse_cookie_env, build_client_from_env, main
USERID = "a" * 32
COMPANY = "8m0atd"
# ---------------------------------------------------------------------------
# Entrypoint shape
# ---------------------------------------------------------------------------
def test_main_is_async_entrypoint():
"""``main`` must be a coroutine function so ``asyncio.run(main())`` works."""
assert asyncio.iscoroutinefunction(main)
# ---------------------------------------------------------------------------
# build_client_from_env — auth-mode selection
# ---------------------------------------------------------------------------
async def test_build_client_prefers_existing_session(monkeypatch):
"""YK_USERID/YK_COMPANY/YK_COOKIE set => existing-session mode."""
# Clear any conflicting env so the test is hermetic.
for k in ("YUNKE_USERNAME", "YUNKE_PASSWORD"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("YK_USERID", USERID)
monkeypatch.setenv("YK_COMPANY", COMPANY)
monkeypatch.setenv("YK_COOKIE", "JSESSIONID=deadbeef; foo=bar")
client = build_client_from_env()
try:
assert isinstance(client, YunkeClient)
# Session mode => already logged in (login() must NOT be required).
assert client.logged_in
assert client.login_result is not None
assert client.login_result.userid == USERID
assert client.login_result.company == COMPANY
# Cookies from YK_COOKIE landed in the underlying httpx cookie jar.
assert client.client.cookies.get("JSESSIONID") == "deadbeef"
assert client.client.cookies.get("foo") == "bar"
# And login mode was NOT chosen (no creds attached).
assert client.username is None
assert client.password is None
finally:
# build_client_from_env constructs an owned httpx.AsyncClient; close
# it so the test does not leak the connection pool.
await client.aclose()
async def test_build_client_login_mode_when_only_creds(monkeypatch):
"""YUNKE_USERNAME + YUNKE_PASSWORD (no session env) => credentialed client."""
for k in ("YK_USERID", "YK_COMPANY", "YK_COOKIE"):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("YUNKE_USERNAME", "alice")
monkeypatch.setenv("YUNKE_PASSWORD", "secret")
client = build_client_from_env()
try:
assert isinstance(client, YunkeClient)
assert client.username == "alice"
assert client.password == "secret"
# Login mode => NOT logged in yet; the caller must call login().
assert not client.logged_in
assert client.login_result is None
finally:
await client.aclose()
async def test_build_client_session_takes_priority_over_creds(monkeypatch):
"""If both session and credential env are present, session mode wins."""
monkeypatch.setenv("YK_USERID", USERID)
monkeypatch.setenv("YK_COMPANY", COMPANY)
monkeypatch.setenv("YUNKE_USERNAME", "alice")
monkeypatch.setenv("YUNKE_PASSWORD", "secret")
client = build_client_from_env()
try:
assert isinstance(client, YunkeClient)
assert client.logged_in
# Session mode chosen; cred fields left unset.
assert client.username is None
finally:
await client.aclose()
def test_build_client_returns_none_when_no_env(monkeypatch):
"""No auth env set => None (the caller prints usage and exits 2)."""
for k in (
"YK_USERID",
"YK_COMPANY",
"YK_COOKIE",
"YUNKE_USERNAME",
"YUNKE_PASSWORD",
):
monkeypatch.delenv(k, raising=False)
assert build_client_from_env() is None
def test_build_client_raises_on_partial_session_env(monkeypatch):
"""Only one of YK_USERID/YK_COMPANY set => ValueError (clear caller bug)."""
for k in (
"YK_COMPANY",
"YK_COOKIE",
"YUNKE_USERNAME",
"YUNKE_PASSWORD",
):
monkeypatch.delenv(k, raising=False)
monkeypatch.setenv("YK_USERID", USERID)
import pytest
with pytest.raises(ValueError, match="YK_USERID and YK_COMPANY"):
build_client_from_env()
# ---------------------------------------------------------------------------
# _parse_cookie_env — cookie string parsing
# ---------------------------------------------------------------------------
def test_parse_cookie_env_raw_header():
assert _parse_cookie_env("JSESSIONID=abc; foo=bar") == {
"JSESSIONID": "abc",
"foo": "bar",
}
def test_parse_cookie_env_json():
assert _parse_cookie_env('{"JSESSIONID": "abc", "x": "y"}') == {
"JSESSIONID": "abc",
"x": "y",
}
def test_parse_cookie_env_empty_returns_empty_dict():
assert _parse_cookie_env("") == {}
assert _parse_cookie_env(" ") == {}
def test_parse_cookie_env_malformed_returns_empty_dict():
# Non-JSON, no equals signs -> nothing parseable -> empty.
assert _parse_cookie_env("not a cookie") == {}
# Malformed JSON -> empty (no raise).
assert _parse_cookie_env('{bad json') == {}