RuyiPage 网络能力

# =============================
# RuyiPage 示例注释版
# 标题: 网络拦截入门：mock、fail、改请求头、队列模式
# 说明: 通过本地 TestServer 让请求拦截、mock 和 fail 行为可重复验证。
# 你会学到:
#   1. 拦截请求并直接 mock 响应
#   2. 主动 fail 请求
#   3. 修改请求头后继续请求
#   4. 无 handler 队列模式手动处理请求
# 相关 API: page.intercept.start, page.intercept.wait, req.mock, req.fail, req.continue_request, page.run_js
# =============================

# -*- coding: utf-8 -*-
"""
示例11: 网络拦截（本地可重复版）

测试覆盖：
1) beforeRequestSent + mock 响应
2) beforeRequestSent + fail 请求
3) beforeRequestSent + 修改请求头并继续
4) 无 handler 队列模式 wait()

说明：
- 本示例全部使用本地 TestServer，避免外网波动导致结果不稳定。
- 所有拦截都在当前页面 context 内触发，便于复现与调试。
"""

import io
import os
import sys


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.dirname(__file__))

from ruyipage import FirefoxOptions, FirefoxPage
from test_server import TestServer


def test_network_intercept():
    """网络拦截演示主流程（适合初学者逐步阅读）。"""
    print("=" * 60)
    print("测试11: 网络拦截")
    print("=" * 60)

    server = TestServer(port=8888)
    server.start()

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)

    try:
        # 准备一个干净页面，确保后续 fetch 都在当前 context 内触发。
        page.get("about:blank")
        page.wait(0.5)

        # ==========================================================
        # 1) mock 响应
        # ==========================================================
        print("\n1. 拦截并 Mock 本地 API:")

        def mock_handler(req):
            if "/api/data" in req.url:
                print(f"   拦截到请求: {req.method} {req.url}")
                # 注意：mock 需要带上 CORS 头，不然浏览器会把跨域响应拦掉，
                # fetch 侧会看到 NetworkError。
                req.mock(
                    '{"status":"ok","data":{"message":"mocked-by-interceptor"}}',
                    status_code=200,
                    headers=[
                        {
                            "name": "content-type",
                            "value": {"type": "string", "value": "application/json"},
                        },
                        {
                            "name": "access-control-allow-origin",
                            "value": {"type": "string", "value": "*"},
                        },
                    ],
                )
                print("   ✓ 已返回 Mock 响应")
            else:
                # 不处理的请求必须显式继续，否则会一直挂起。
                req.continue_request()

        page.intercept.start(mock_handler, phases=["beforeRequestSent"])
        mocked = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.data.message).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   mock结果: {mocked}")
        page.intercept.stop()

        # ==========================================================
        # 2) fail 请求
        # ==========================================================
        print("\n2. 阻止本地 API 请求:")

        def fail_handler(req):
            if "/api/data" in req.url:
                print(f"   阻止请求: {req.url}")
                req.fail()
            else:
                req.continue_request()

        page.intercept.start(fail_handler, phases=["beforeRequestSent"])
        blocked = page.run_js(
            """
            return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   fail结果: {blocked}")
        page.intercept.stop()

        # ==========================================================
        # 3) 修改请求头并继续
        # ==========================================================
        print("\n3. 修改请求头并继续:")

        def header_handler(req):
            if "/api/headers" in req.url:
                # continue_request(headers=...) 会覆盖本次请求头。
                req.continue_request(
                    headers=[
                        {
                            "name": "X-Ruyi-Demo",
                            "value": {"type": "string", "value": "yes"},
                        },
                        {
                            "name": "User-Agent",
                            "value": {"type": "string", "value": "RuyiPage/Example11"},
                        },
                    ]
                )
                print("   ✓ 已注入自定义请求头")
            else:
                req.continue_request()

        page.intercept.start(header_handler, phases=["beforeRequestSent"])
        headers_json = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
            """,
            server.get_url("/api/headers"),
            as_expr=False,
        )
        x_header = ""
        if isinstance(headers_json, dict):
            x_header = headers_json.get("X-Ruyi-Demo") or headers_json.get(
                "x-ruyi-demo"
            )
        print(f"   服务器看到的 X-Ruyi-Demo: {x_header}")
        page.intercept.stop()

        # ==========================================================
        # 4) 无 handler 队列模式
        # ==========================================================
        print("\n4. 队列模式 wait() 捕获拦截请求:")
        # handler=None 表示不在回调里处理，而是进入队列，供 wait() 手动取出。
        page.intercept.start(handler=None, phases=["beforeRequestSent"])
        page.run_js(
            """
            fetch(arguments[0]).catch(() => null);
            return 'sent';
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        req = page.intercept.wait(timeout=5)
        if req:
            print(f"   wait捕获: {req.method} {req.url}")
            # 队列模式下记得手动继续/中止请求，否则请求会悬挂。
            req.continue_request()
        else:
            print("   ⚠ 未在超时内捕获请求")
        page.intercept.stop()

        print("\n" + "=" * 60)
        print("✓ 网络拦截测试通过！")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 测试失败: {e}")
        import traceback

        traceback.print_exc()
    finally:
        try:
            page.intercept.stop()
        except Exception:
            pass
        page.wait(1)
        page.quit()
        server.stop()


if __name__ == "__main__":
    test_network_intercept()

# -*- coding: utf-8 -*-
"""
示例11: 网络拦截（本地可重复版）

测试覆盖：
1) beforeRequestSent + mock 响应
2) beforeRequestSent + fail 请求
3) beforeRequestSent + 修改请求头并继续
4) 无 handler 队列模式 wait()

说明：
- 本示例全部使用本地 TestServer，避免外网波动导致结果不稳定。
- 所有拦截都在当前页面 context 内触发，便于复现与调试。
"""

import io
import os
import sys


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.dirname(__file__))

from ruyipage import FirefoxOptions, FirefoxPage
from test_server import TestServer


def test_network_intercept():
    """网络拦截演示主流程（适合初学者逐步阅读）。"""
    print("=" * 60)
    print("测试11: 网络拦截")
    print("=" * 60)

    server = TestServer(port=8888)
    server.start()

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)

    try:
        # 准备一个干净页面，确保后续 fetch 都在当前 context 内触发。
        page.get("about:blank")
        page.wait(0.5)

        # ==========================================================
        # 1) mock 响应
        # ==========================================================
        print("\n1. 拦截并 Mock 本地 API:")

        def mock_handler(req):
            if "/api/data" in req.url:
                print(f"   拦截到请求: {req.method} {req.url}")
                # 注意：mock 需要带上 CORS 头，不然浏览器会把跨域响应拦掉，
                # fetch 侧会看到 NetworkError。
                req.mock(
                    '{"status":"ok","data":{"message":"mocked-by-interceptor"}}',
                    status_code=200,
                    headers=[
                        {
                            "name": "content-type",
                            "value": {"type": "string", "value": "application/json"},
                        },
                        {
                            "name": "access-control-allow-origin",
                            "value": {"type": "string", "value": "*"},
                        },
                    ],
                )
                print("   ✓ 已返回 Mock 响应")
            else:
                # 不处理的请求必须显式继续，否则会一直挂起。
                req.continue_request()

        page.intercept.start(mock_handler, phases=["beforeRequestSent"])
        mocked = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.data.message).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   mock结果: {mocked}")
        page.intercept.stop()

        # ==========================================================
        # 2) fail 请求
        # ==========================================================
        print("\n2. 阻止本地 API 请求:")

        def fail_handler(req):
            if "/api/data" in req.url:
                print(f"   阻止请求: {req.url}")
                req.fail()
            else:
                req.continue_request()

        page.intercept.start(fail_handler, phases=["beforeRequestSent"])
        blocked = page.run_js(
            """
            return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   fail结果: {blocked}")
        page.intercept.stop()

        # ==========================================================
        # 3) 修改请求头并继续
        # ==========================================================
        print("\n3. 修改请求头并继续:")

        def header_handler(req):
            if "/api/headers" in req.url:
                # continue_request(headers=...) 会覆盖本次请求头。
                req.continue_request(
                    headers=[
                        {
                            "name": "X-Ruyi-Demo",
                            "value": {"type": "string", "value": "yes"},
                        },
                        {
                            "name": "User-Agent",
                            "value": {"type": "string", "value": "RuyiPage/Example11"},
                        },
                    ]
                )
                print("   ✓ 已注入自定义请求头")
            else:
                req.continue_request()

        page.intercept.start(header_handler, phases=["beforeRequestSent"])
        headers_json = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
            """,
            server.get_url("/api/headers"),
            as_expr=False,
        )
        x_header = ""
        if isinstance(headers_json, dict):
            x_header = headers_json.get("X-Ruyi-Demo") or headers_json.get(
                "x-ruyi-demo"
            )
        print(f"   服务器看到的 X-Ruyi-Demo: {x_header}")
        page.intercept.stop()

        # ==========================================================
        # 4) 无 handler 队列模式
        # ==========================================================
        print("\n4. 队列模式 wait() 捕获拦截请求:")
        # handler=None 表示不在回调里处理，而是进入队列，供 wait() 手动取出。
        page.intercept.start(handler=None, phases=["beforeRequestSent"])
        page.run_js(
            """
            fetch(arguments[0]).catch(() => null);
            return 'sent';
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        req = page.intercept.wait(timeout=5)
        if req:
            print(f"   wait捕获: {req.method} {req.url}")
            # 队列模式下记得手动继续/中止请求，否则请求会悬挂。
            req.continue_request()
        else:
            print("   ⚠ 未在超时内捕获请求")
        page.intercept.stop()

        print("\n" + "=" * 60)
        print("✓ 网络拦截测试通过！")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 测试失败: {e}")
        import traceback

        traceback.print_exc()
    finally:
        try:
            page.intercept.stop()
        except Exception:
            pass
        page.wait(1)
        page.quit()
        server.stop()


if __name__ == "__main__":
    test_network_intercept()

# =============================
# RuyiPage 示例注释版
# 标题: 高级网络功能：请求/响应拦截、mock、阻止、fetchError
# 说明: 展示 intercept 与 events 组合后的完整网络控制能力。
# 你会学到:
#   1. 在 beforeRequestSent 阶段继续请求
#   2. 修改请求头
#   3. mock JSON 响应
#   4. 阻止请求
#   5. 在 responseStarted 阶段修改响应状态
# 相关 API: page.intercept.start_requests, page.intercept.start_responses, req.continue_request, req.mock, req.fail, req.continue_response, page.events.start, page.events.wait
# =============================

# -*- coding: utf-8 -*-
"""示例18: 高级网络功能（完整场景）

覆盖：
1) beforeRequestSent 拦截 + 继续请求
2) 修改请求头
3) mock 响应
4) 阻止请求
5) responseStarted 阶段修改响应状态
6) 队列模式 wait() 手动处理
7) fetchError 事件监听
"""

import io
import os
import sys


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.dirname(__file__))

from ruyipage import FirefoxOptions, FirefoxPage
from ruyipage._functions.tools import find_free_port
from test_server import TestServer


def test_advanced_network():
    print("=" * 60)
    print("测试18: 高级网络功能")
    print("=" * 60)

    server = TestServer(port=find_free_port(8889, 8989))
    server.start()

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)

    fetch_error_list = []

    try:
        page.get("about:blank")
        page.wait(0.6)

        # 1) 请求拦截
        print("\n1. beforeRequestSent 拦截 + 继续请求:")
        intercepted = []

        def req_handler(req):
            if "/api/data" in req.url:
                intercepted.append(req.url)
                print(f"   拦截到请求: {req.method} {req.url}")
            req.continue_request()

        page.intercept.start_requests(req_handler)
        data = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.status).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        page.wait(0.5)
        print(f"   请求结果: {data}")
        print(f"   拦截计数: {len(intercepted)}")
        page.intercept.stop()

        # 2) 修改请求头
        print("\n2. 修改请求头:")

        def header_handler(req):
            if "/api/headers" in req.url:
                req.continue_request(
                    headers=[
                        {
                            "name": "X-Ruyi-Demo",
                            "value": {"type": "string", "value": "yes"},
                        },
                        {
                            "name": "User-Agent",
                            "value": {"type": "string", "value": "RuyiPage/Example18"},
                        },
                    ]
                )
                print("   ✓ 已注入请求头")
            else:
                req.continue_request()

        page.intercept.start_requests(header_handler)
        headers = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
            """,
            server.get_url("/api/headers"),
            as_expr=False,
        )
        print(
            f"   X-Ruyi-Demo: {headers.get('X-Ruyi-Demo') or headers.get('x-ruyi-demo')}"
        )
        page.intercept.stop()

        # 3) mock 响应
        print("\n3. mock 响应:")

        def mock_handler(req):
            if "/api/data" in req.url:
                req.mock(
                    '{"status":"mocked","data":{"message":"这是Mock数据"}}',
                    status_code=200,
                    headers=[
                        {
                            "name": "content-type",
                            "value": {"type": "string", "value": "application/json"},
                        },
                        {
                            "name": "access-control-allow-origin",
                            "value": {"type": "string", "value": "*"},
                        },
                    ],
                )
                print("   ✓ 已返回Mock响应")
            else:
                req.continue_request()

        page.intercept.start_requests(mock_handler)
        mock_msg = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.status + ':' + d.data.message).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   Mock结果: {mock_msg}")
        page.intercept.stop()

        # 4) 阻止请求
        print("\n4. 阻止请求:")

        def block_handler(req):
            if "/api/error" in req.url:
                req.fail()
                print("   ✓ 请求已阻止")
            else:
                req.continue_request()

        page.intercept.start_requests(block_handler)
        blocked = page.run_js(
            """
            return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
            """,
            server.get_url("/api/error"),
            as_expr=False,
        )
        print(f"   阻止结果: {blocked}")
        page.intercept.stop()

        # 5) 响应阶段修改状态码
        print("\n5. responseStarted 阶段修改响应状态码:")

        def response_handler(req):
            if "/api/data" in req.url:
                req.continue_response(status_code=299, reason_phrase="RuyiModified")
                print("   ✓ 响应状态码已改为 299")
            else:
                req.continue_response()

        page.intercept.start_responses(response_handler)
        modified_status = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.status + ':' + r.statusText).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   响应状态: {modified_status}")
        page.intercept.stop()

        # 6) 队列模式
        print("\n6. 队列模式 wait() 手动处理:")
        page.intercept.start(handler=None, phases=["beforeRequestSent"])
        page.run_js(
            """
            fetch(arguments[0]).catch(() => null);
            return 'sent';
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        queued = page.intercept.wait(timeout=5)
        if queued:
            print(f"   wait捕获: {queued.method} {queued.url}")
            queued.continue_request()
        else:
            print("   ⚠ 未捕获到请求")
        page.intercept.stop()

        # 7) fetchError 事件监听
        print("\n7. fetchError 事件监听:")
        page.events.start(["network.fetchError"], contexts=[page.tab_id])
        page.events.clear()

        page.intercept.start_requests(block_handler)
        page.run_js(
            """
            fetch(arguments[0]).catch(() => null);
            return 'sent';
            """,
            server.get_url("/api/error"),
            as_expr=False,
        )
        page.wait(1)
        page.intercept.stop()
        while True:
            evt = page.events.wait("network.fetchError", timeout=0.2)
            if not evt:
                break
            fetch_error_list.append(evt)
        print(f"   fetchError 事件数量: {len(fetch_error_list)}")

        print("\n" + "=" * 60)
        print("✓ 所有高级网络功能测试通过！")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 测试失败: {e}")
        import traceback

        traceback.print_exc()
    finally:
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        page.wait(1)
        page.quit()
        server.stop()


if __name__ == "__main__":
    test_advanced_network()

# -*- coding: utf-8 -*-
"""示例18: 高级网络功能（完整场景）

覆盖：
1) beforeRequestSent 拦截 + 继续请求
2) 修改请求头
3) mock 响应
4) 阻止请求
5) responseStarted 阶段修改响应状态
6) 队列模式 wait() 手动处理
7) fetchError 事件监听
"""

import io
import os
import sys


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.dirname(__file__))

from ruyipage import FirefoxOptions, FirefoxPage
from ruyipage._functions.tools import find_free_port
from test_server import TestServer


def test_advanced_network():
    print("=" * 60)
    print("测试18: 高级网络功能")
    print("=" * 60)

    server = TestServer(port=find_free_port(8889, 8989))
    server.start()

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)

    fetch_error_list = []

    try:
        page.get("about:blank")
        page.wait(0.6)

        # 1) 请求拦截
        print("\n1. beforeRequestSent 拦截 + 继续请求:")
        intercepted = []

        def req_handler(req):
            if "/api/data" in req.url:
                intercepted.append(req.url)
                print(f"   拦截到请求: {req.method} {req.url}")
            req.continue_request()

        page.intercept.start_requests(req_handler)
        data = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.status).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        page.wait(0.5)
        print(f"   请求结果: {data}")
        print(f"   拦截计数: {len(intercepted)}")
        page.intercept.stop()

        # 2) 修改请求头
        print("\n2. 修改请求头:")

        def header_handler(req):
            if "/api/headers" in req.url:
                req.continue_request(
                    headers=[
                        {
                            "name": "X-Ruyi-Demo",
                            "value": {"type": "string", "value": "yes"},
                        },
                        {
                            "name": "User-Agent",
                            "value": {"type": "string", "value": "RuyiPage/Example18"},
                        },
                    ]
                )
                print("   ✓ 已注入请求头")
            else:
                req.continue_request()

        page.intercept.start_requests(header_handler)
        headers = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
            """,
            server.get_url("/api/headers"),
            as_expr=False,
        )
        print(
            f"   X-Ruyi-Demo: {headers.get('X-Ruyi-Demo') or headers.get('x-ruyi-demo')}"
        )
        page.intercept.stop()

        # 3) mock 响应
        print("\n3. mock 响应:")

        def mock_handler(req):
            if "/api/data" in req.url:
                req.mock(
                    '{"status":"mocked","data":{"message":"这是Mock数据"}}',
                    status_code=200,
                    headers=[
                        {
                            "name": "content-type",
                            "value": {"type": "string", "value": "application/json"},
                        },
                        {
                            "name": "access-control-allow-origin",
                            "value": {"type": "string", "value": "*"},
                        },
                    ],
                )
                print("   ✓ 已返回Mock响应")
            else:
                req.continue_request()

        page.intercept.start_requests(mock_handler)
        mock_msg = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.status + ':' + d.data.message).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   Mock结果: {mock_msg}")
        page.intercept.stop()

        # 4) 阻止请求
        print("\n4. 阻止请求:")

        def block_handler(req):
            if "/api/error" in req.url:
                req.fail()
                print("   ✓ 请求已阻止")
            else:
                req.continue_request()

        page.intercept.start_requests(block_handler)
        blocked = page.run_js(
            """
            return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
            """,
            server.get_url("/api/error"),
            as_expr=False,
        )
        print(f"   阻止结果: {blocked}")
        page.intercept.stop()

        # 5) 响应阶段修改状态码
        print("\n5. responseStarted 阶段修改响应状态码:")

        def response_handler(req):
            if "/api/data" in req.url:
                req.continue_response(status_code=299, reason_phrase="RuyiModified")
                print("   ✓ 响应状态码已改为 299")
            else:
                req.continue_response()

        page.intercept.start_responses(response_handler)
        modified_status = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.status + ':' + r.statusText).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   响应状态: {modified_status}")
        page.intercept.stop()

        # 6) 队列模式
        print("\n6. 队列模式 wait() 手动处理:")
        page.intercept.start(handler=None, phases=["beforeRequestSent"])
        page.run_js(
            """
            fetch(arguments[0]).catch(() => null);
            return 'sent';
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        queued = page.intercept.wait(timeout=5)
        if queued:
            print(f"   wait捕获: {queued.method} {queued.url}")
            queued.continue_request()
        else:
            print("   ⚠ 未捕获到请求")
        page.intercept.stop()

        # 7) fetchError 事件监听
        print("\n7. fetchError 事件监听:")
        page.events.start(["network.fetchError"], contexts=[page.tab_id])
        page.events.clear()

        page.intercept.start_requests(block_handler)
        page.run_js(
            """
            fetch(arguments[0]).catch(() => null);
            return 'sent';
            """,
            server.get_url("/api/error"),
            as_expr=False,
        )
        page.wait(1)
        page.intercept.stop()
        while True:
            evt = page.events.wait("network.fetchError", timeout=0.2)
            if not evt:
                break
            fetch_error_list.append(evt)
        print(f"   fetchError 事件数量: {len(fetch_error_list)}")

        print("\n" + "=" * 60)
        print("✓ 所有高级网络功能测试通过！")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 测试失败: {e}")
        import traceback

        traceback.print_exc()
    finally:
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        page.wait(1)
        page.quit()
        server.stop()


if __name__ == "__main__":
    test_advanced_network()

#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 网络数据采集器：响应体抓取、读取与释放
# 说明: 用于说明 add_data_collector 在数据抓取和协议观测中的价值。
# 你会学到:
#   1. 注入额外请求头
#   2. 设置缓存 bypass
#   3. 添加并使用 data collector 采集 response 数据
#   4. 按 request_id 读取采集结果
#   5. 释放采集句柄并移除 collector
# 相关 API: page.network.set_extra_headers, page.network.set_cache_behavior, page.network.add_data_collector, collector.get, collector.disown, collector.remove, page.listen.start
# =============================

# -*- coding: utf-8 -*-
"""示例28: Network Data Collector（严格结果版）"""

import io
import sys
from typing import Dict, List

if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")

from ruyipage import DataCollector, DataPacket, FirefoxPage, InterceptedRequest
from ruyipage._functions.tools import find_free_port

from test_server import TestServer


def add_result(
    results: List[Dict[str, str]], item: str, status: str, note: str
) -> None:
    results.append({"item": item, "status": status, "note": note})


def print_results(results: List[Dict[str, str]]) -> None:
    print("\n| 项目 | 状态 | 说明 |")
    print("| --- | --- | --- |")
    for row in results:
        print(f"| {row['item']} | {row['status']} | {row['note']} |")


def main() -> None:
    print("=" * 70)
    print("测试 28: Network Data Collector")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    collector: DataCollector | None = None
    server = None

    try:
        server = TestServer(port=find_free_port(9230, 9330)).start()
        base_url = server.get_url("")[:-1]

        page.get("about:blank")
        add_result(
            results, "页面加载", "成功", f"about:blank 已加载，服务地址: {base_url}"
        )

        page.network.set_extra_headers({"X-Test-Collector": "yes"})
        add_result(
            results, "network.setExtraHeaders", "成功", "已设置 X-Test-Collector"
        )

        try:
            page.network.set_cache_behavior("bypass")
            add_result(
                results, "network.setCacheBehavior", "成功", "缓存行为已设为 bypass"
            )
        except Exception as e:
            add_result(results, "network.setCacheBehavior", "不支持", str(e)[:120])

        collector = page.network.add_data_collector(
            ["responseCompleted"],
            data_types=["response"],
        )
        if collector.id:
            add_result(
                results, "network.addDataCollector", "成功", f"collector={collector.id}"
            )
        else:
            add_result(
                results, "network.addDataCollector", "失败", "未返回 collector ID"
            )

        page.intercept.start_requests()
        page.listen.start("/api/collector")
        page.run_js(
            f"fetch('{base_url}/api/collector').catch(()=>null); return true;",
            as_expr=False,
        )

        req: InterceptedRequest | None = page.intercept.wait(timeout=8)

        if req:
            has_header = "X-Test-Collector" in req.headers
            add_result(
                results,
                "request header injected",
                "成功" if has_header else "失败",
                f"request_id={req.request_id}, X-Test-Collector={has_header}",
            )
            req.continue_request()
        else:
            add_result(
                results,
                "request header injected",
                "失败",
                "未捕获到 beforeRequestSent 请求",
            )

        packet: DataPacket | None = page.listen.wait(timeout=8)

        if packet:
            add_result(
                results,
                "network.responseCompleted observed",
                "成功",
                f"status={packet.status} url={packet.url}",
            )
        else:
            add_result(
                results,
                "network.responseCompleted observed",
                "失败",
                "未在超时内观察到 responseCompleted",
            )

        if collector and req:
            got_response_data = False
            try:
                data = collector.get(req.request_id, data_type="response")
                if data.has_data:
                    got_response_data = True
                    add_result(
                        results,
                        "network.getData response",
                        "成功",
                        f"bytes={data.bytes} base64={'yes' if data.base64 else 'no'}",
                    )
                else:
                    add_result(
                        results, "network.getData response", "失败", f"raw={data.raw}"
                    )
            except Exception as e:
                add_result(results, "network.getData response", "失败", str(e)[:120])

            if got_response_data:
                try:
                    collector.disown(req.request_id, data_type="response")
                    add_result(
                        results, "network.disownData", "成功", "已释放 response 数据"
                    )
                except Exception as e:
                    add_result(results, "network.disownData", "失败", str(e)[:120])

                try:
                    data_after = collector.get(req.request_id, data_type="response")
                    if not data_after.has_data:
                        add_result(
                            results,
                            "network.getData after disown",
                            "成功",
                            "释放后已无可用数据",
                        )
                    else:
                        add_result(
                            results,
                            "network.getData after disown",
                            "跳过",
                            f"释放后仍可读取: {data_after.raw}",
                        )
                except Exception as e:
                    add_result(
                        results,
                        "network.getData after disown",
                        "成功",
                        f"释放后读取报错: {str(e)[:100]}",
                    )
            else:
                add_result(
                    results,
                    "network.disownData",
                    "跳过",
                    "未成功拿到 response 数据，跳过释放验证",
                )
                add_result(
                    results,
                    "network.getData after disown",
                    "跳过",
                    "未成功拿到 response 数据，跳过释放后验证",
                )

        if collector:
            try:
                collector.remove()
                add_result(
                    results,
                    "network.removeDataCollector",
                    "成功",
                    f"已移除 {collector.id}",
                )
                collector = None
            except Exception as e:
                add_result(results, "network.removeDataCollector", "失败", str(e)[:120])

        page.network.clear_extra_headers()
        add_result(results, "clear extra headers", "成功", "额外请求头已清理")

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.listen.stop()
        except Exception:
            pass
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        if collector:
            try:
                collector.remove()
            except Exception:
                pass
        try:
            page.network.clear_extra_headers()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""示例28: Network Data Collector（严格结果版）"""

import io
import sys
from typing import Dict, List

if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")

from ruyipage import DataCollector, DataPacket, FirefoxPage, InterceptedRequest
from ruyipage._functions.tools import find_free_port

from test_server import TestServer


def add_result(
    results: List[Dict[str, str]], item: str, status: str, note: str
) -> None:
    results.append({"item": item, "status": status, "note": note})


def print_results(results: List[Dict[str, str]]) -> None:
    print("\n| 项目 | 状态 | 说明 |")
    print("| --- | --- | --- |")
    for row in results:
        print(f"| {row['item']} | {row['status']} | {row['note']} |")


def main() -> None:
    print("=" * 70)
    print("测试 28: Network Data Collector")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    collector: DataCollector | None = None
    server = None

    try:
        server = TestServer(port=find_free_port(9230, 9330)).start()
        base_url = server.get_url("")[:-1]

        page.get("about:blank")
        add_result(
            results, "页面加载", "成功", f"about:blank 已加载，服务地址: {base_url}"
        )

        page.network.set_extra_headers({"X-Test-Collector": "yes"})
        add_result(
            results, "network.setExtraHeaders", "成功", "已设置 X-Test-Collector"
        )

        try:
            page.network.set_cache_behavior("bypass")
            add_result(
                results, "network.setCacheBehavior", "成功", "缓存行为已设为 bypass"
            )
        except Exception as e:
            add_result(results, "network.setCacheBehavior", "不支持", str(e)[:120])

        collector = page.network.add_data_collector(
            ["responseCompleted"],
            data_types=["response"],
        )
        if collector.id:
            add_result(
                results, "network.addDataCollector", "成功", f"collector={collector.id}"
            )
        else:
            add_result(
                results, "network.addDataCollector", "失败", "未返回 collector ID"
            )

        page.intercept.start_requests()
        page.listen.start("/api/collector")
        page.run_js(
            f"fetch('{base_url}/api/collector').catch(()=>null); return true;",
            as_expr=False,
        )

        req: InterceptedRequest | None = page.intercept.wait(timeout=8)

        if req:
            has_header = "X-Test-Collector" in req.headers
            add_result(
                results,
                "request header injected",
                "成功" if has_header else "失败",
                f"request_id={req.request_id}, X-Test-Collector={has_header}",
            )
            req.continue_request()
        else:
            add_result(
                results,
                "request header injected",
                "失败",
                "未捕获到 beforeRequestSent 请求",
            )

        packet: DataPacket | None = page.listen.wait(timeout=8)

        if packet:
            add_result(
                results,
                "network.responseCompleted observed",
                "成功",
                f"status={packet.status} url={packet.url}",
            )
        else:
            add_result(
                results,
                "network.responseCompleted observed",
                "失败",
                "未在超时内观察到 responseCompleted",
            )

        if collector and req:
            got_response_data = False
            try:
                data = collector.get(req.request_id, data_type="response")
                if data.has_data:
                    got_response_data = True
                    add_result(
                        results,
                        "network.getData response",
                        "成功",
                        f"bytes={data.bytes} base64={'yes' if data.base64 else 'no'}",
                    )
                else:
                    add_result(
                        results, "network.getData response", "失败", f"raw={data.raw}"
                    )
            except Exception as e:
                add_result(results, "network.getData response", "失败", str(e)[:120])

            if got_response_data:
                try:
                    collector.disown(req.request_id, data_type="response")
                    add_result(
                        results, "network.disownData", "成功", "已释放 response 数据"
                    )
                except Exception as e:
                    add_result(results, "network.disownData", "失败", str(e)[:120])

                try:
                    data_after = collector.get(req.request_id, data_type="response")
                    if not data_after.has_data:
                        add_result(
                            results,
                            "network.getData after disown",
                            "成功",
                            "释放后已无可用数据",
                        )
                    else:
                        add_result(
                            results,
                            "network.getData after disown",
                            "跳过",
                            f"释放后仍可读取: {data_after.raw}",
                        )
                except Exception as e:
                    add_result(
                        results,
                        "network.getData after disown",
                        "成功",
                        f"释放后读取报错: {str(e)[:100]}",
                    )
            else:
                add_result(
                    results,
                    "network.disownData",
                    "跳过",
                    "未成功拿到 response 数据，跳过释放验证",
                )
                add_result(
                    results,
                    "network.getData after disown",
                    "跳过",
                    "未成功拿到 response 数据，跳过释放后验证",
                )

        if collector:
            try:
                collector.remove()
                add_result(
                    results,
                    "network.removeDataCollector",
                    "成功",
                    f"已移除 {collector.id}",
                )
                collector = None
            except Exception as e:
                add_result(results, "network.removeDataCollector", "失败", str(e)[:120])

        page.network.clear_extra_headers()
        add_result(results, "clear extra headers", "成功", "额外请求头已清理")

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.listen.stop()
        except Exception:
            pass
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        if collector:
            try:
                collector.remove()
            except Exception:
                pass
        try:
            page.network.clear_extra_headers()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()