aios/tests/test_wechat_send_backend.py

from __future__ import annotations

import importlib.util
import io
import os
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
from types import SimpleNamespace
from unittest import TestCase
from unittest.mock import Mock, call, patch


BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"


def _load_backend():
    spec = importlib.util.spec_from_file_location("wechat_send_backend", BACKEND_PATH)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"unable to load backend from {BACKEND_PATH}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


class WechatSendBackendTests(TestCase):
    @classmethod
    def setUpClass(cls):
        cls.backend = _load_backend()

    def test_select_backend_prefers_ucc_adapter_over_openclaw_channel(self):
        with (
            patch.object(self.backend, "SEND_BACKEND", ""),
            patch.object(self.backend, "ucc_adapter_alive", return_value=True),
            patch.object(self.backend, "discover_openclaw_wechat_channel", return_value="some-wechat-channel"),
        ):
            selected = self.backend.select_backend()

        self.assertEqual(selected["backend"], "ucc-adapter")
        self.assertEqual(selected["reason"], "ucc_adapter_alive")

    def test_openclaw_chat_only_is_not_wechat_channel(self):
        completed = SimpleNamespace(returncode=0, stdout='{"chat": {}}')
        with (
            patch.object(self.backend, "WECHAT_CHANNEL", ""),
            patch.object(self.backend, "_run_openclaw", return_value=completed),
        ):
            self.assertIsNone(self.backend.discover_openclaw_wechat_channel())

    def test_send_ucc_adapter_uses_canonical_endpoint_and_sent_status(self):
        with (
            patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
            patch.object(
                self.backend,
                "_post_ucc",
                return_value=(
                    200,
                    {
                        "ok": True,
                        "status": "sent",
                        "provider_message_id": "openclaw-weixin:1-abc",
                    },
                ),
            ) as post,
        ):
            result = self.backend.send_ucc_adapter(
                target="chat-001",
                content="hello",
                account_id="acct-001",
                message_id="msg-001",
                metadata={"source": "test"},
            ).to_dict()

        self.assertTrue(result["ok"])
        self.assertEqual(result["backend"], "ucc-adapter")
        self.assertEqual(result["status"], "sent")
        self.assertEqual(result["endpoint"], "/api/internal/wechat/send")
        post.assert_called_once()
        args, _ = post.call_args
        self.assertEqual(args[0], "/api/internal/wechat/send")
        self.assertEqual(args[1]["target"], "chat-001")
        self.assertEqual(args[1]["content"], "hello")
        self.assertFalse(args[1]["dry_run"])

    def test_ucc_auth_is_authorization_header_not_argv_or_output(self):
        secret = "ucc-super-secret-token"

        class Response:
            status = 200

            def __enter__(self):
                return self

            def __exit__(self, *_args):
                return False

            def read(self, _limit=-1):
                return b'{"ok":true,"status":"sent"}'

        urlopen = Mock(return_value=Response())
        output = io.StringIO()
        with (
            patch.object(self.backend, "UCC_ADAPTER_TOKEN", secret),
            patch.object(self.backend, "_open_no_redirect", urlopen),
            redirect_stdout(output),
            redirect_stderr(output),
        ):
            status, payload = self.backend._post_ucc(
                "/api/internal/wechat/send", {"target": "chat-001", "content": "hello"}
            )

        request = urlopen.call_args.args[0]
        self.assertEqual(request.get_header("Authorization"), f"Bearer {secret}")
        self.assertEqual(status, 200)
        self.assertEqual(payload, {"ok": True, "status": "sent"})
        self.assertNotIn(secret, " ".join(self.backend._openclaw_base_args()))
        self.assertNotIn(secret, output.getvalue())
        self.assertFalse({"--gateway", "--token", "--password"} & set(self.backend._openclaw_base_args()))

    def test_ucc_paths_are_canonical_first_unique_and_bounded(self):
        with patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"):
            paths = self.backend._ucc_send_paths()

        self.assertEqual(paths[0], "/api/internal/wechat/send")
        self.assertLessEqual(len(paths), 2)
        self.assertEqual(len(paths), len(set(paths)))
        self.assertTrue(all(path.startswith("/") for path in paths))

    def test_ucc_accepts_only_explicit_true_and_sent_when_not_dry_run(self):
        self.assertTrue(self.backend._ucc_response_sent({"ok": True, "status": "sent"}))
        for response in (
            {"ok": 1, "status": "sent"},
            {"ok": True, "status": "accepted"},
            {"ok": False, "status": "sent"},
            {"status": "sent"},
        ):
            with self.subTest(response=response):
                self.assertFalse(self.backend._ucc_response_sent(response))

    def test_ambiguous_canonical_failures_never_fallback(self):
        secret = "raw-upstream-secret"
        failures = (
            TimeoutError(secret),
            ConnectionError(secret),
            self.backend._BackendProtocolError(secret),
            RuntimeError(secret),
        )
        for failure in failures:
            with self.subTest(failure=type(failure).__name__):
                with (
                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
                    patch.object(self.backend, "_post_ucc", side_effect=failure) as post,
                ):
                    result = self.backend.send_ucc_adapter(
                        target="chat-001", content="hello", message_id="msg-001"
                    ).to_dict()

                self.assertFalse(result["ok"])
                self.assertEqual(result["code"], "ucc_adapter_unavailable")
                self.assertEqual(post.call_count, 1)
                self.assertNotIn(secret, repr(result))

    def test_only_explicit_http_404_or_405_uses_legacy_fallback(self):
        for http_status in (404, 405):
            with self.subTest(http_status=http_status):
                missing = urllib.error.HTTPError(
                    "http://127.0.0.1/api/internal/wechat/send",
                    http_status,
                    "not found",
                    {},
                    None,
                )
                with (
                    patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
                    patch.object(
                        self.backend,
                        "_post_ucc",
                        side_effect=[missing, (200, {"ok": True, "status": "sent"})],
                    ) as post,
                ):
                    result = self.backend.send_ucc_adapter(
                        target="chat-001", content="hello", message_id="msg-001"
                    ).to_dict()

                self.assertTrue(result["ok"])
                self.assertEqual(result["endpoint"], "/internal/channels/wechat/send")
                self.assertEqual(post.call_count, 2)

    def test_non_not_found_http_error_never_falls_back(self):
        denied = urllib.error.HTTPError(
            "http://127.0.0.1/api/internal/wechat/send", 401, "denied", {}, None
        )
        with patch.object(self.backend, "_post_ucc", side_effect=denied) as post:
            result = self.backend.send_ucc_adapter(target="chat-001", content="hello").to_dict()

        self.assertFalse(result["ok"])
        self.assertEqual(result["code"], "ucc_adapter_http_error")
        post.assert_called_once()

    def test_ucc_rejection_does_not_expose_raw_error_or_secret(self):
        secret = "provider-password=secret-value"
        with patch.object(
            self.backend,
            "_post_ucc",
            return_value=(200, {"ok": False, "status": "failed", "error": secret}),
        ):
            result = self.backend.send_ucc_adapter(
                target="chat-001", content="hello", message_id="msg-001"
            ).to_dict()

        self.assertFalse(result["ok"])
        self.assertEqual(result["code"], "ucc_adapter_rejected")
        self.assertNotIn(secret, repr(result))

    def test_dry_run_requires_dry_run_status_and_never_reports_sent(self):
        with patch.object(
            self.backend,
            "_post_ucc",
            return_value=(200, {"ok": True, "status": "dry_run"}),
        ):
            result = self.backend.send_ucc_adapter(
                target="chat-001", content="hello", dry_run=True
            ).to_dict()

        self.assertTrue(result["ok"])
        self.assertEqual(result["status"], "dry_run")
        self.assertNotEqual(result["status"], "sent")

        for dry_run, upstream_status in ((True, "sent"), (False, "dry_run")):
            with self.subTest(dry_run=dry_run, upstream_status=upstream_status):
                with patch.object(
                    self.backend,
                    "_post_ucc",
                    return_value=(200, {"ok": True, "status": upstream_status}),
                ):
                    rejected = self.backend.send_ucc_adapter(
                        target="chat-001", content="hello", dry_run=dry_run
                    ).to_dict()
                self.assertFalse(rejected["ok"])
                self.assertEqual(rejected["status"], "failed")

    def test_invalid_provider_message_id_never_reverses_confirmed_send(self):
        invalid_values = (
            "\ud800",
            "contains\x00control",
            "x" * (self.backend.MAX_IDENTIFIER_BYTES + 1),
            {"not": "a string"},
        )
        for provider_message_id in invalid_values:
            with self.subTest(value=repr(provider_message_id)[:40]):
                with patch.object(
                    self.backend,
                    "_post_ucc",
                    return_value=(
                        200,
                        {
                            "ok": True,
                            "status": "sent",
                            "provider_message_id": provider_message_id,
                        },
                    ),
                ):
                    result = self.backend.send_ucc_adapter(
                        target="chat-001", content="hello"
                    ).to_dict()

                self.assertTrue(result["ok"])
                self.assertEqual(result["status"], "sent")
                self.assertNotIn("provider_message_id", result)

    def test_openclaw_2026_6_11_commands_have_no_unsupported_global_gateway(self):
        completed = SimpleNamespace(returncode=0, stdout='{"channels": []}')
        with (
            patch.object(self.backend, "OPENCLAW_BIN", "/usr/local/bin/openclaw"),
            patch.object(self.backend, "_run_openclaw", return_value=completed) as run,
        ):
            self.assertIsNone(self.backend.discover_openclaw_wechat_channel())

        command = run.call_args.args[0]
        self.assertEqual(command, ["/usr/local/bin/openclaw", "channels", "list", "--json"])
        self.assertFalse({"--gateway", "--token", "--password"} & set(command))

    def test_openclaw_output_overflow_terminates_while_process_is_running(self):
        writer = (
            "import os,time;"
            f"os.write(1,b'x'*{self.backend.MAX_RESPONSE_BYTES + 4096});"
            "time.sleep(10)"
        )
        started = time.monotonic()
        completed = self.backend._run_openclaw([sys.executable, "-c", writer], timeout=4)
        elapsed = time.monotonic() - started

        self.assertLess(elapsed, 2.0)
        self.assertNotEqual(completed.returncode, 0)
        self.assertEqual(completed.stdout, "")
        self.assertEqual(completed.stderr, "")

    def test_openclaw_timeout_returns_failure_and_terminates_promptly(self):
        started = time.monotonic()
        completed = self.backend._run_openclaw(
            [sys.executable, "-c", "import time;time.sleep(10)"], timeout=1
        )
        elapsed = time.monotonic() - started

        self.assertLess(elapsed, 3.0)
        self.assertNotEqual(completed.returncode, 0)
        self.assertEqual(completed.stdout, "")
        self.assertEqual(completed.stderr, "")

    def test_timeout_kills_inherited_pipe_descendant_after_parent_exits(self):
        if os.name != "posix" or not Path("/proc").is_dir():
            self.skipTest("POSIX process-group and procfs contract")

        def process_is_running(pid):
            try:
                state = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[2]
            except (FileNotFoundError, IndexError, OSError):
                return False
            return state not in {"X", "Z"}

        with tempfile.TemporaryDirectory() as directory:
            pid_path = Path(directory) / "descendant.pid"
            descendant = "import time;time.sleep(30)"
            launcher = (
                "import pathlib,subprocess,sys;"
                f"child=subprocess.Popen([sys.executable,'-c',{descendant!r}]);"
                f"pathlib.Path({str(pid_path)!r}).write_text(str(child.pid),encoding='utf-8')"
            )
            started = time.monotonic()
            completed = self.backend._run_openclaw(
                [sys.executable, "-c", launcher], timeout=1
            )
            elapsed = time.monotonic() - started
            self.assertTrue(pid_path.is_file())
            descendant_pid = int(pid_path.read_text(encoding="utf-8"))
            try:
                deadline = time.monotonic() + 2
                while process_is_running(descendant_pid) and time.monotonic() < deadline:
                    time.sleep(0.02)
                self.assertFalse(process_is_running(descendant_pid))
            finally:
                if process_is_running(descendant_pid):
                    os.kill(descendant_pid, signal.SIGKILL)

        self.assertLess(elapsed, 3.0)
        self.assertNotEqual(completed.returncode, 0)

    def test_process_tree_termination_uses_posix_process_group(self):
        if os.name != "posix":
            self.skipTest("POSIX process-group contract")

        class Process:
            pid = 43210

            def __init__(self):
                self.wait_count = 0

            def poll(self):
                return None

            def wait(self, timeout):
                self.wait_count += 1
                if self.wait_count == 1:
                    raise subprocess.TimeoutExpired(["openclaw"], timeout)
                return -signal.SIGKILL

        process = Process()
        with patch.object(self.backend.os, "killpg") as killpg:
            self.backend._terminate_process_tree(process)

        self.assertEqual(
            killpg.call_args_list,
            [
                call(process.pid, signal.SIGTERM),
                call(process.pid, signal.SIGKILL),
            ],
        )

    def test_unsupported_and_generic_chat_backends_fail_closed(self):
        with (
            patch.object(self.backend, "SEND_BACKEND", "openclaw-chat"),
            patch.object(self.backend, "ucc_adapter_alive", return_value=False),
        ):
            selected = self.backend.select_backend()
            result = self.backend.send_message(target="chat-001", content="hello")

        self.assertEqual(selected["backend"], "unavailable")
        self.assertEqual(selected["reason"], "unsupported_backend")
        self.assertFalse(result["ok"])
        self.assertEqual(result["code"], "unsupported_backend")

    def test_invalid_or_oversized_input_is_rejected_before_network(self):
        with patch.object(self.backend, "_post_ucc") as post:
            empty = self.backend.send_ucc_adapter(target="", content="hello").to_dict()
            oversized = self.backend.send_ucc_adapter(
                target="chat-001", content="x" * (self.backend.MAX_CONTENT_BYTES + 1)
            ).to_dict()

        self.assertEqual(empty["code"], "invalid_input")
        self.assertEqual(oversized["code"], "input_too_large")
        post.assert_not_called()

    def test_unpaired_surrogate_inputs_fail_closed_before_network(self):
        cases = (
            {"target": "\ud800", "content": "hello"},
            {"target": "chat-001", "content": "\ud800"},
            {"target": "chat-001", "content": "hello", "account_id": "\ud800"},
            {"target": "chat-001", "content": "hello", "message_id": "\ud800"},
            {"target": "chat-001", "content": "hello", "metadata": {"bad": "\ud800"}},
        )
        with patch.object(self.backend, "_post_ucc") as post:
            for kwargs in cases:
                with self.subTest(field=next(reversed(kwargs))):
                    result = self.backend.send_ucc_adapter(**kwargs).to_dict()
                    self.assertEqual(result["code"], "invalid_input")
        post.assert_not_called()

    def test_deeply_nested_metadata_is_bounded_before_network(self):
        metadata = {}
        for _ in range(1200):
            metadata = {"nested": metadata}

        with patch.object(self.backend, "_post_ucc") as post:
            result = self.backend.send_ucc_adapter(
                target="chat-001", content="hello", metadata=metadata
            ).to_dict()

        self.assertEqual(result["code"], "input_too_large")
        post.assert_not_called()

    def test_adapter_origin_must_be_allowlisted(self):
        with (
            patch.object(self.backend, "UCC_ADAPTER_URL", "https://evil.example:8443"),
            patch.object(
                self.backend,
                "UCC_ADAPTER_ALLOWED_ORIGINS",
                ("http://127.0.0.1:18800",),
            ),
        ):
            self.assertIsNone(self.backend._adapter_base_url())

        with (
            patch.object(self.backend, "UCC_ADAPTER_URL", "https://adapter.example:8443"),
            patch.object(
                self.backend,
                "UCC_ADAPTER_ALLOWED_ORIGINS",
                ("https://adapter.example:8443",),
            ),
        ):
            self.assertEqual(
                self.backend._adapter_base_url(), "https://adapter.example:8443"
            )

    def test_adapter_origin_rejects_scheme_userinfo_fragment_and_path(self):
        invalid_origins = (
            "ftp://127.0.0.1:18800",
            "http://user:password@127.0.0.1:18800",
            "http://127.0.0.1:18800#fragment",
            "http://127.0.0.1:18800/base-path",
        )
        for origin in invalid_origins:
            with self.subTest(origin=origin):
                with (
                    patch.object(self.backend, "UCC_ADAPTER_URL", origin),
                    patch.object(
                        self.backend,
                        "UCC_ADAPTER_ALLOWED_ORIGINS",
                        ("http://127.0.0.1:18800",),
                    ),
                ):
                    self.assertIsNone(self.backend._adapter_base_url())

    def test_adapter_path_rejects_decoded_traversal(self):
        for path in (
            "/api/%2e%2e/send",
            "/api/%2E%2E/send",
            "/api/%2525252e%2525252e/send",
            "/api/%252525252e%252525252e/send",
            "/api/../send",
            "/api/%zz/send",
        ):
            with self.subTest(path=path):
                self.assertIsNone(self.backend._safe_path(path))

    def test_adapter_path_keeps_intended_safe_canonical_paths(self):
        for path in (
            "/api/internal/wechat/send",
            "/internal/channels/wechat/send",
            "/api/%7Ehealth/status",
        ):
            with self.subTest(path=path):
                self.assertEqual(self.backend._safe_path(path), path)

    def test_redirects_are_refused_and_authorization_cannot_cross_origin(self):
        handler = self.backend._NoRedirectHandler()
        self.assertIsNone(
            handler.redirect_request(
                None,
                None,
                302,
                "redirect",
                {"Location": "https://evil.example/steal"},
                "https://evil.example/steal",
            )
        )

        redirected = urllib.error.HTTPError(
            "http://127.0.0.1:18800/api/internal/wechat/send",
            302,
            "redirect",
            {"Location": "https://evil.example/steal"},
            None,
        )
        with (
            patch.object(self.backend, "UCC_ADAPTER_TOKEN", "header-secret"),
            patch.object(self.backend, "_open_no_redirect", side_effect=redirected) as opened,
        ):
            result = self.backend.send_ucc_adapter(
                target="chat-001", content="hello"
            ).to_dict()

        request = opened.call_args.args[0]
        self.assertEqual(urllib.parse.urlsplit(request.full_url).netloc, "127.0.0.1:18800")
        self.assertEqual(request.get_header("Authorization"), "Bearer header-secret")
        opened.assert_called_once()
        self.assertFalse(result["ok"])
        self.assertEqual(result["code"], "ucc_adapter_http_error")


if __name__ == "__main__":
    import unittest

    unittest.main()