# -*- coding: utf-8 -*- """Shared fixtures for integration tests. These fixtures start a real QwenPaw app subprocess with isolated workspace directories and a sanitized environment to avoid touching local secrets. Subprocess coverage (optional): QWENPAW_INTEGRATION_COVERAGE=1 pytest tests/integration/ When set, ``pytest_sessionstart`` writes a coverage rcfile under ``.integration_coverage/`` with an **absolute** ``source=`` path (``…/src/pineagents``); the app subprocess runs with ``COVERAGE_PROCESS_START`` / ``COVERAGE_FILE`` so the child traces that tree. The fixture stops the app with **SIGINT** first so coverage can flush (SIGTERM often yields empty data). After the session, files under ``.integration_coverage/`` are combined and HTML is written to ``htmlcov-integration/``. Run integration tests without ``--cov`` from pytest-cov (or use ``--no-cov``) so the parent process does not enforce ``fail_under`` on near-zero host-process coverage. pytest-xdist compatibility: pytest tests/integration/ -n auto --dist=loadscope Each xdist worker is a separate process; ``app_server`` (module-scoped) naturally isolates per-module. Coverage data files use ``parallel=true`` with unique PID suffixes — no cross-worker collision. The final ``coverage combine`` + ``coverage html`` runs only in the controller process (or single-process mode), not in individual workers. """ from __future__ import annotations import json import os import shutil import signal import socket import subprocess import sys import threading import time from collections.abc import Iterator from dataclasses import dataclass from pathlib import Path from typing import Any import httpx import pytest _INTEGRATION_COVERAGE_DIR: Path | None = None _COVERAGE_SUBPROC_BASENAME = "integration_subproc" _COVERAGE_RCFILE_NAME = "coverage_subprocess.ini" @pytest.fixture def isolated_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: """Redirect user-home lookup so tests cannot touch developer files.""" home = tmp_path / "home" home.mkdir() monkeypatch.setenv("HOME", str(home)) monkeypatch.setenv("USERPROFILE", str(home)) return home def _write_integration_subprocess_rc(root: Path, dest_ini: Path) -> None: """Write a coverage rcfile with absolute ``source`` for the app subprocess. Relative ``source=`` paths in a checked-in rcfile are not resolved reliably when the file is loaded via ``COVERAGE_PROCESS_START``, which produced empty traces (0 files) even though the app ran. """ src_qwenpaw = (root / "src" / "qwenpaw").resolve() text = ( "[run]\n" "parallel = true\n" "branch = false\n" f"source = {src_qwenpaw}\n" "omit =\n" " */tests/*\n" " */test_*\n" " */__pycache__/*\n" ) dest_ini.write_text(text, encoding="utf-8") def _integration_coverage_requested() -> bool: return os.environ.get( "QWENPAW_INTEGRATION_COVERAGE", "", ).strip().lower() in ( "1", "true", "yes", ) def pytest_sessionstart(session: pytest.Session) -> None: """Prepare directory and config for subprocess coverage when requested.""" global _INTEGRATION_COVERAGE_DIR if not _integration_coverage_requested(): return root = Path(session.config.rootpath).resolve() _INTEGRATION_COVERAGE_DIR = root / ".integration_coverage" _INTEGRATION_COVERAGE_DIR.mkdir(parents=True, exist_ok=True) if not os.environ.get("PYTEST_XDIST_WORKER"): for p in _INTEGRATION_COVERAGE_DIR.glob( f"{_COVERAGE_SUBPROC_BASENAME}*", ): p.unlink(missing_ok=True) _write_integration_subprocess_rc( root, _INTEGRATION_COVERAGE_DIR / _COVERAGE_RCFILE_NAME, ) def pytest_sessionfinish( # pylint: disable=unused-argument session: pytest.Session, exitstatus: int, ) -> None: """Merge parallel coverage files from app subprocesses and write HTML.""" if ( not _integration_coverage_requested() or _INTEGRATION_COVERAGE_DIR is None ): return if os.environ.get("PYTEST_XDIST_WORKER"): return wd = _INTEGRATION_COVERAGE_DIR if not any(wd.glob(f"{_COVERAGE_SUBPROC_BASENAME}*")): print( "[integration coverage] No data files under " f"{wd} (no app_server tests ran?).", flush=True, ) return combine = subprocess.run( [ sys.executable, "-m", "coverage", "combine", "--data-file", _COVERAGE_SUBPROC_BASENAME, ], cwd=wd, capture_output=True, text=True, check=False, ) if combine.returncode != 0: print( "[integration coverage] coverage combine failed:\n" f"{combine.stdout}\n{combine.stderr}", flush=True, ) return root = Path(session.config.rootpath).resolve() html_dir = root / "htmlcov-integration" if html_dir.is_dir(): shutil.rmtree(html_dir) html = subprocess.run( [ sys.executable, "-m", "coverage", "html", "--data-file", _COVERAGE_SUBPROC_BASENAME, "-d", str(html_dir), ], cwd=wd, capture_output=True, text=True, check=False, ) if html.returncode != 0: print( "[integration coverage] coverage html failed:\n" f"{html.stdout}\n{html.stderr}", flush=True, ) return print( f"[integration coverage] HTML report: {html_dir / 'index.html'}", flush=True, ) _SENSITIVE_ENV_VARS = ( "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "DASHSCOPE_API_KEY", "DINGTALK_APP_KEY", "DINGTALK_APP_SECRET", "FEISHU_APP_ID", "FEISHU_APP_SECRET", "DISCORD_TOKEN", "TELEGRAM_BOT_TOKEN", "TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", ) def _find_free_port(host: str = "127.0.0.1") -> int: """Bind to port 0 and return the assigned free port.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind((host, 0)) sock.listen(1) return int(sock.getsockname()[1]) def _tee_stream(stream, buffer: list[str]) -> None: """Read subprocess output, tag and print live, keep a raw copy.""" prefix = "[app server] " try: for line in iter(stream.readline, ""): buffer.append(line) try: print(f"{prefix}{line}", end="", flush=True) except (OSError, ValueError): # pytest may close captured stdout before this daemon thread # finishes draining; keep the raw copy in `buffer` regardless. pass finally: stream.close() @dataclass class AppServer: """Handle to a running app subprocess used by tests.""" host: str port: int process: subprocess.Popen[str] client: httpx.Client logs: list[str] log_thread: threading.Thread # Working directory of the subprocess (= QWENPAW_WORKING_DIR). Tests that # need to seed file-backed stores (inbox_events.json, cron jobs_history/, # backups, etc.) write directly under this path. The subprocess re-reads # these files on each HTTP request, so no restart is needed after seeding. working_dir: Path @property def base_url(self) -> str: return f"http://{self.host}:{self.port}" def logs_tail(self, chars: int = 4000) -> str: return "".join(self.logs)[-chars:] @staticmethod def _compact(value: Any, max_len: int | None = None) -> str: """Render params/body/response for logs (single line, escaped). ``max_len`` is only applied when set; by default the full string is kept so integration logs are usable for debugging. """ if value is None: return "-" if isinstance(value, str): text = value else: try: text = json.dumps(value, ensure_ascii=False, sort_keys=True) except TypeError: text = repr(value) text = text.replace("\n", "\\n") if max_len is not None and len(text) > max_len: return f"{text[: max_len - 3]}..." return text def api_request( self, method: str, path: str, **kwargs: Any, ) -> httpx.Response: """Send a request and print the full request/response to stdout.""" url = f"{self.base_url}{path}" if path.startswith("/") else path request_payload = kwargs.get("json") if request_payload is None: request_payload = kwargs.get("data") request_params = kwargs.get("params") response = self.client.request( method=method.upper(), url=url, **kwargs, ) response_text = response.text level = "PASS" if 200 <= response.status_code < 400 else "FAIL" print( ( f"[integration][{level}] {method.upper()} {path} | " f"params={self._compact(request_params)} | " f"request={self._compact(request_payload)} | " f"status={response.status_code} | " f"response={self._compact(response_text)}" ), flush=True, ) return response @pytest.fixture(scope="module") def app_server( # pylint: disable=too-many-statements,too-many-branches tmp_path_factory: pytest.TempPathFactory, ) -> Iterator[AppServer]: """Start one isolated qwenpaw app process per test module. Module-scoped: cases in the same file share one subprocess. Cross-module isolation is preserved by re-launching with a fresh tmp dir. Cases must use unique resource ids (agent_id, chat_id, ...) to stay isolated within a module — current convention (e.g. ``integ_ws_01``) already supports this. """ tmp_path = tmp_path_factory.mktemp("app_server") host = "127.0.0.1" port = _find_free_port(host) working_dir = tmp_path / "working" secret_dir = tmp_path / "working.secret" backups_dir = tmp_path / "working.backups" working_dir.mkdir(parents=True, exist_ok=True) secret_dir.mkdir(parents=True, exist_ok=True) backups_dir.mkdir(parents=True, exist_ok=True) env = os.environ.copy() for key in _SENSITIVE_ENV_VARS: env.pop(key, None) env["QWENPAW_WORKING_DIR"] = str(working_dir) env["QWENPAW_SECRET_DIR"] = str(secret_dir) env["QWENPAW_BACKUP_DIR"] = str(backups_dir) env["QWENPAW_AUTH_ENABLED"] = "false" # Set the upload size limit used by /api/.../upload-limit and the # request-body cap. Read once at app import time from this env var, # so it must be present before the subprocess starts. env["QWENPAW_UPLOAD_MAX_SIZE_MB"] = "10" # Integration tests run in a temporary isolated workspace and must not # touch the developer's OS keychain. Force file-backed secrets so first # encryption does not block on desktop keyring discovery. env["QWENPAW_RUNNING_IN_CONTAINER"] = "true" env["NO_PROXY"] = "*" env["PYTHONUNBUFFERED"] = "1" # Force UTF-8 stdio in the subprocess so non-ASCII log lines (e.g. # 中文/emoji from skills, agentscope, etc.) don't crash the parent's # _tee_stream reader on Windows where the default console encoding # is cp1252. env["PYTHONIOENCODING"] = "utf-8" if _integration_coverage_requested(): if _INTEGRATION_COVERAGE_DIR is None: raise AssertionError( "QWENPAW_INTEGRATION_COVERAGE is set but coverage dir was not " "initialised (pytest_sessionstart should create " ".integration_coverage/).", ) rcfile = _INTEGRATION_COVERAGE_DIR / _COVERAGE_RCFILE_NAME env["COVERAGE_PROCESS_START"] = str(rcfile.resolve()) env["COVERAGE_FILE"] = str( _INTEGRATION_COVERAGE_DIR / _COVERAGE_SUBPROC_BASENAME, ) logs: list[str] = [] # Windows + subprocess coverage: create a new process group so the # child can receive CTRL_BREAK_EVENT for graceful shutdown # (TerminateProcess skips atexit and coverage data is lost). popen_kwargs: dict[str, Any] = {} if sys.platform == "win32" and _integration_coverage_requested(): popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP with subprocess.Popen( [ sys.executable, "-m", "qwenpaw", "app", "--host", host, "--port", str(port), "--log-level", "info", ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, # Decode subprocess output as UTF-8 in the parent. Without this, # Popen falls back to locale.getpreferredencoding(False) which # is cp1252 on Windows CI runners and crashes _tee_stream. encoding="utf-8", errors="replace", env=env, **popen_kwargs, ) as process: assert process.stdout is not None log_thread = threading.Thread( target=_tee_stream, args=(process.stdout, logs), daemon=True, ) log_thread.start() # 15s default lets cold-start endpoints (ACP getter, heartbeat) # finish without hiding real deadlocks; 30s in coverage mode # for tracer overhead. http_timeout = 30.0 if _integration_coverage_requested() else 15.0 client = httpx.Client(timeout=http_timeout, trust_env=False) try: max_wait_seconds = 60 start_at = time.time() last_error: str | None = None while time.time() - start_at < max_wait_seconds: if process.poll() is not None: raise AssertionError( "qwenpaw app exited during startup.\n" f"exit_code={process.returncode}\n" f"logs:\n{''.join(logs)[-4000:]}", ) try: resp = client.get(f"http://{host}:{port}/api/healthz") if resp.status_code == 200: break except (httpx.ConnectError, httpx.TimeoutException) as exc: last_error = str(exc) time.sleep(0.5) else: raise AssertionError( "qwenpaw core agents did not become ready in time.\n" f"last_error={last_error}\n" f"logs:\n{''.join(logs)[-4000:]}", ) yield AppServer( host=host, port=port, process=process, client=client, logs=logs, log_thread=log_thread, working_dir=working_dir, ) finally: client.close() if process.poll() is None: # On POSIX, SIGINT lets uvicorn shut down cleanly so # subprocess coverage data flushes (SIGTERM often skips # atexit / data-file write). On Windows, SIGINT is not # delivered reliably to subprocesses; when subprocess # coverage is enabled we create the child with # CREATE_NEW_PROCESS_GROUP and send CTRL_BREAK_EVENT so # the child can run atexit / flush coverage data. # Without coverage we use terminate() for fast shutdown. try: if sys.platform == "win32": if _integration_coverage_requested(): process.send_signal(signal.CTRL_BREAK_EVENT) else: process.terminate() else: process.send_signal(signal.SIGINT) process.wait(timeout=15) except subprocess.TimeoutExpired: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) log_thread.join(timeout=2)