How to listen newest window's XHR?

[muffinweb]

After using sb.switch_to_newest_window() method to listen XHR of newest window, i recall listenXHR(tab) method to listen xhr logs.

# Switching newest window
sb.switch_to_newest_window()

# Recalling 
tab = sb.cdp.tab
listenXHR(tab)
.
.
loop = sb.cdp.get_event_loop()
xhr_responses = loop.run_until_complete(receiveXHR(tab, xhr_requests))
.
.

i also made a simple demo website that fits my scenario while i'm trying to solve it.

W3School's service for static web apps

https://myedgecase.w3spaces.com/index.html

I mention this issue before here #3782, but i guess i couldn't point actual problem

[mdmintz]

Since the new window doesn't have CDP Mode loaded, you would need to grab the URL from the new tab, and then load it in your original CDP Mode window with the XHR listener:

"""CDP.network.ResponseReceived with CDP.network.ResourceType.XHR."""
import ast
import colorama
import mycdp
import sys
import time
from seleniumbase import SB

xhr_requests = []
last_xhr_request = None
c1 = colorama.Fore.BLUE + colorama.Back.LIGHTYELLOW_EX
c2 = colorama.Fore.BLUE + colorama.Back.LIGHTGREEN_EX
cr = colorama.Style.RESET_ALL
if "linux" in sys.platform:
    c1 = c2 = cr = ""

def listenXHR(page):
    async def handler(evt):
        # Get AJAX requests
        if evt.type_ is mycdp.network.ResourceType.XHR:
            xhr_requests.append([evt.response.url, evt.request_id])
            global last_xhr_request
            last_xhr_request = time.time()
    page.add_handler(mycdp.network.ResponseReceived, handler)

async def receiveXHR(page, requests):
    responses = []
    retries = 0
    max_retries = 5
    # Wait at least 2 seconds after last XHR request for more
    while True:
        if last_xhr_request is None or retries > max_retries:
            break
        if time.time() - last_xhr_request <= 2:
            retries = retries + 1
            time.sleep(2)
            continue
        else:
            break
    await page
    # Loop through gathered requests and get response body
    for request in requests:
        try:
            res = await page.send(mycdp.network.get_response_body(request[1]))
            if res is None:
                continue
            responses.append({
                "url": request[0],
                "body": res[0],
                "is_base64": res[1],
            })
        except Exception as e:
            print("Error getting response:", e)
    return responses

with SB(uc=True, test=True, locale="en") as sb:
    sb.activate_cdp_mode("https://myedgecase.w3spaces.com/index.html")
    sb.sleep(1)
    sb.cdp.click("#continue-btn")
    sb.sleep(2)
    sb.click("#new_window_button")
    sb.sleep(2)
    sb.switch_to_newest_window()
    tab_url = sb.get_current_url()
    sb.switch_to_default_window()
    tab = sb.cdp.page
    listenXHR(tab)
    sb.open(tab_url)
    sb.sleep(7)

    loop = sb.cdp.get_event_loop()
    xhr_responses = loop.run_until_complete(receiveXHR(tab, xhr_requests))
    for response in xhr_responses:
        print(c1 + "*** ==> XHR Request URL <== ***" + cr)
        print(f'{response["url"]}')
        is_base64 = response["is_base64"]
        b64_data = "Base64 encoded data"
        try:
            headers = ast.literal_eval(response["body"])["headers"]
            print(c2 + "*** ==> XHR Response Headers <== ***" + cr)
            print(headers if not is_base64 else b64_data)
        except Exception:
            response_body = response["body"]
            print(c2 + "*** ==> XHR Response Body <== ***" + cr)
            print(response_body if not is_base64 else b64_data)

Output:

*** ==> XHR Request URL <== ***
https://jsonplaceholder.typicode.com/todos/1
*** ==> XHR Response Body <== ***
{
  "userId": 1,
  "id": 1,
  "title": "delectus aut autem",
  "completed": false
}