diff --git a/scripts/run_tests.py b/scripts/run_tests.py new file mode 100644 index 0000000..1964773 --- /dev/null +++ b/scripts/run_tests.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Run project tests consistently on Windows, Linux, and WSL. + +First run in an existing virtual environment: + python scripts/run_tests.py --setup + +Subsequent runs: + python scripts/run_tests.py + python scripts/run_tests.py --quick + +The --setup option installs this repository's declared [dev] dependencies +into the *active virtual environment*, then runs the tests. +""" +from __future__ import annotations + +import argparse +import importlib.util +import os +from pathlib import Path +import subprocess +import sys + + +REQUIRED_MODULES = ("pytest", "fastapi", "httpx", "uvicorn", "requests") + + +def run(command: list[str], *, cwd: Path, env: dict[str, str]) -> int: + print("\n$ " + " ".join(command), flush=True) + try: + result = subprocess.run(command, cwd=str(cwd), env=env, check=False) + except OSError as exc: + print(f"Unable to execute command: {exc}", file=sys.stderr) + return 127 + print(f"Exit code: {result.returncode}", flush=True) + return result.returncode + + +def missing_modules() -> list[str]: + return [module for module in REQUIRED_MODULES if importlib.util.find_spec(module) is None] + + +def main() -> int: + parser = argparse.ArgumentParser(description="Run repository tests with isolated Python dependencies") + parser.add_argument("--quick", action="store_true", help="Run only test_login_headers.py") + parser.add_argument( + "--setup", + action="store_true", + help="Install the project's [dev] dependencies into the active virtual environment before testing", + ) + args = parser.parse_args() + + root = Path(__file__).resolve().parent.parent + src = root / "src" + if not (src / "codebuddy_proxy").is_dir(): + print(f"ERROR: missing package directory: {src / 'codebuddy_proxy'}", file=sys.stderr) + return 2 + + env = os.environ.copy() + existing_path = env.get("PYTHONPATH", "") + env["PYTHONPATH"] = os.pathsep.join(p for p in (str(src), str(root), existing_path) if p) + + print(f"Project: {root}", flush=True) + print(f"Python: {sys.executable}", flush=True) + print(f"PYTHONPATH: {env['PYTHONPATH']}", flush=True) + + if args.setup: + if sys.prefix == sys.base_prefix: + print( + "ERROR: --setup requires a virtual environment; refusing to install into system Python.", + file=sys.stderr, + ) + print("Create one first: python3 -m venv .venv", file=sys.stderr) + return 2 + print("Installing project dependencies, including pytest, from pyproject.toml.", flush=True) + install = run( + [sys.executable, "-m", "pip", "install", "-e", ".[dev]"], + cwd=root, + env=env, + ) + if install != 0: + print("RESULT: SETUP FAILED (see pip output above)", file=sys.stderr) + return install + + missing = missing_modules() + if missing: + print(f"\nERROR: missing Python modules: {', '.join(missing)}", file=sys.stderr) + print("Within your virtual environment, run this one-time setup:", file=sys.stderr) + print(f' "{sys.executable}" scripts/run_tests.py --setup', file=sys.stderr) + return 2 + + test_options = [["test_login_headers.py", "-vv", "--tb=long"]] + if not args.quick: + test_options.append(["-q", "--tb=short"]) + + failures = 0 + for options in test_options: + if run([sys.executable, "-m", "pytest", *options], cwd=root, env=env) != 0: + failures += 1 + + print(f"\nRESULT: {'PASS' if failures == 0 else f'FAIL ({failures} test command(s) failed)'}") + return 0 if failures == 0 else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test_backend_profiles.py b/test_backend_profiles.py index 8e540c1..1ae67db 100644 --- a/test_backend_profiles.py +++ b/test_backend_profiles.py @@ -167,7 +167,14 @@ def test_root_model_config_matches_packaged_domestic_resource(): ) root_copy = Path(__file__).with_name("models_config.json") - assert root_copy.read_bytes() == package_resource.read_bytes() + # Git may check out text files with CRLF on Windows and LF on Linux/WSL. + # Normalize only CRLF vs LF; all other bytes must remain identical. + def normalize_newlines(value: bytes) -> bytes: + return value.replace(b"\r\n", b"\n") + + assert normalize_newlines(root_copy.read_bytes()) == normalize_newlines( + package_resource.read_bytes() + ) def test_refresh_preserves_session_metadata_and_machine_id(tmp_path, monkeypatch): diff --git a/test_login_headers.py b/test_login_headers.py index d045a64..d365603 100644 --- a/test_login_headers.py +++ b/test_login_headers.py @@ -4,14 +4,17 @@ import json from src.codebuddy_proxy.codebuddy_client_demo import CodeBuddyClient -def test_enterprise_headers(): +def test_enterprise_headers(tmp_path): """测试企业认证头生成""" print("=" * 60) print("测试企业认证头生成") print("=" * 60) # 场景1: token 无 domain 字段（最常见的场景） - client = CodeBuddyClient(endpoint="https://copilot.tencent.com") + client = CodeBuddyClient( + endpoint="https://copilot.tencent.com", + session_file=tmp_path / "domestic-session.json", + ) token_no_domain = {"accessToken": "test_token_123"} headers1 = client._enterprise_headers(token_no_domain) @@ -38,7 +41,10 @@ def test_enterprise_headers(): print(" ✓ 正确使用 token.domain 覆盖默认值") # 场景3: 带端口的 endpoint - client_with_port = CodeBuddyClient(endpoint="https://copilot.example.com:8443") + client_with_port = CodeBuddyClient( + endpoint="https://copilot.example.com:8443", + session_file=tmp_path / "custom-session.json", + ) headers3 = client_with_port._enterprise_headers(token_no_domain) print("\n场景3: endpoint 包含端口") @@ -53,13 +59,13 @@ def test_enterprise_headers(): print("✓ 所有测试通过！") print("=" * 60) -def test_login_account_headers(): +def test_login_account_headers(tmp_path): """验证 /login/account 请求会正确组合所有必需的 headers""" print("\n" + "=" * 60) print("验证 /login/account 请求头组合") print("=" * 60) - client = CodeBuddyClient() + client = CodeBuddyClient(session_file=tmp_path / "account-session.json") mock_token = {"accessToken": "bearer_token_xyz"} # 模拟 login 方法中的 headers 组合逻辑 diff --git a/test_proxy_headers.py b/test_proxy_headers.py index 4dd2e10..f6469aa 100755 --- a/test_proxy_headers.py +++ b/test_proxy_headers.py @@ -1,48 +1,72 @@ #!/usr/bin/env python3 -"""测试代理实际发送的headers""" +"""Optional live-proxy integration smoke test. + +By default pytest skips this test because it requires a running local proxy, +which may in turn make authenticated upstream requests. + +Opt in from a prepared environment: + WORKBUDDY_RUN_PROXY_INTEGRATION=1 python -m pytest test_proxy_headers.py -vv -s + +Standalone execution remains available: + python test_proxy_headers.py +""" +from __future__ import annotations -import httpx import json -import sys +import os + +import httpx +import pytest + +@pytest.mark.skipif( + os.environ.get("WORKBUDDY_RUN_PROXY_INTEGRATION") != "1", + reason="requires a running proxy; set WORKBUDDY_RUN_PROXY_INTEGRATION=1 to opt in", +) def test_proxy_headers(): - """测试代理转发时的实际headers""" - proxy_url = "http://localhost:8787/v1/chat/completions" - + """Manually inspect real proxy header forwarding with a running server.""" + proxy_url = os.environ.get( + "WORKBUDDY_PROXY_TEST_URL", + "http://127.0.0.1:8787/v1/chat/completions", + ) test_body = { "model": "gpt-4", "messages": [{"role": "user", "content": "测试headers"}], "stream": False, - "max_tokens": 10 + "max_tokens": 10, } - + print("=" * 60) print("测试代理Headers") print("=" * 60) - + try: resp = httpx.post( proxy_url, json=test_body, headers={"Content-Type": "application/json"}, - timeout=30.0 + timeout=30.0, ) - - print(f"状态码: {resp.status_code}") - print(f"\n响应:") - if resp.status_code == 200: + except httpx.ConnectError as exc: + pytest.fail( + f"无法连接代理服务器 {proxy_url}。请先启动服务，然后重试：{exc}" + ) + except httpx.HTTPError as exc: + pytest.fail(f"代理请求失败: {exc}") + + print(f"状态码: {resp.status_code}") + if resp.status_code == 200: + try: print(json.dumps(resp.json(), indent=2, ensure_ascii=False)[:500]) - print("\n✅ 请求成功！检查日志确认headers") - else: - print(f"错误: {resp.text[:500]}") - - except httpx.ConnectError: - print("❌ 无法连接到代理服务器") - print("请先启动代理: uv run --with workbuddy2api python -m codebuddy_proxy") - sys.exit(1) - except Exception as e: - print(f"❌ 请求失败: {e}") - sys.exit(1) + except ValueError: + print(resp.text[:500]) + print("请求成功；请结合服务端日志检查转发的 headers") + else: + print(f"错误: {resp.text[:500]}") + assert resp.status_code == 200, f"代理响应 HTTP {resp.status_code}: {resp.text[:500]}" + if __name__ == "__main__": + # Direct invocation is an explicit opt-in; pytest's skip marker applies + # only when collected through pytest, not when calling this function here. test_proxy_headers()