Skip to content

expect

BaseRequestExpectation

Base class for handling request and response expectations. This class provides a context manager to wait for specific network requests and responses based on a URL pattern. It sets up handlers for request and response events and provides properties to access the request, response, and response body.

Parameters:

Name Type Description Default
tab Connection

The Tab instance to monitor.

 required
url_pattern Union[str, Pattern[str]]

The URL pattern to match requests and responses.

 required
Source code in zendriver/core/expect.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class BaseRequestExpectation:
    """
    Base class for handling request and response expectations.
    This class provides a context manager to wait for specific network requests and responses
    based on a URL pattern. It sets up handlers for request and response events and provides
    properties to access the request, response, and response body.
    :param tab: The Tab instance to monitor.
    :param url_pattern: The URL pattern to match requests and responses.
    """

    def __init__(self, tab: Connection, url_pattern: Union[str, re.Pattern[str]]):
        self.tab = tab
        self.url_pattern = url_pattern
        self.request_future: asyncio.Future[cdp.network.RequestWillBeSent] = (
            asyncio.Future()
        )
        self.response_future: asyncio.Future[cdp.network.ResponseReceived] = (
            asyncio.Future()
        )
        self.loading_finished_future: asyncio.Future[cdp.network.LoadingFinished] = (
            asyncio.Future()
        )
        self.request_id: Union[cdp.network.RequestId, None] = None

    async def _request_handler(self, event: cdp.network.RequestWillBeSent) -> None:
        """
        Internal handler for request events.
        :param event: The request event.
        """
        if re.fullmatch(self.url_pattern, event.request.url):
            self._remove_request_handler()
            self.request_id = event.request_id
            self.request_future.set_result(event)

    async def _response_handler(self, event: cdp.network.ResponseReceived) -> None:
        """
        Internal handler for response events.
        :param event: The response event.
        """
        if event.request_id == self.request_id:
            self._remove_response_handler()
            self.response_future.set_result(event)

    async def _loading_finished_handler(
        self, event: cdp.network.LoadingFinished
    ) -> None:
        """
        Internal handler for loading finished events.
        :param event: The loading finished event.
        """
        if event.request_id == self.request_id:
            self._remove_loading_finished_handler()
            self.loading_finished_future.set_result(event)

    def _remove_request_handler(self) -> None:
        """
        Remove the request event handler.
        """
        self.tab.remove_handlers(cdp.network.RequestWillBeSent, self._request_handler)

    def _remove_response_handler(self) -> None:
        """
        Remove the response event handler.
        """
        self.tab.remove_handlers(cdp.network.ResponseReceived, self._response_handler)

    def _remove_loading_finished_handler(self) -> None:
        """
        Remove the loading finished event handler.
        """
        self.tab.remove_handlers(
            cdp.network.LoadingFinished, self._loading_finished_handler
        )

    async def __aenter__(self):  # type: ignore
        """
        Enter the context manager, adding request and response handlers.
        """
        await self._setup()
        return self

    async def __aexit__(self, *args: Any) -> None:
        """
        Exit the context manager, removing request and response handlers.
        """
        self._teardown()

    async def _setup(self) -> None:
        self.tab.add_handler(cdp.network.RequestWillBeSent, self._request_handler)
        self.tab.add_handler(cdp.network.ResponseReceived, self._response_handler)
        self.tab.add_handler(
            cdp.network.LoadingFinished, self._loading_finished_handler
        )

    def _teardown(self) -> None:
        self._remove_request_handler()
        self._remove_response_handler()
        self._remove_loading_finished_handler()

    async def reset(self) -> None:
        """
        Resets the internal state, allowing the expectation to be reused.
        """
        self.request_future = asyncio.Future()
        self.response_future = asyncio.Future()
        self.loading_finished_future = asyncio.Future()
        self.request_id = None
        self._teardown()
        await self._setup()

    @property
    async def request(self) -> cdp.network.Request:
        """
        Get the matched request.
        :return: The matched request.
        :rtype: cdp.network.Request
        """
        return (await self.request_future).request

    @property
    async def response(self) -> cdp.network.Response:
        """
        Get the matched response.
        :return: The matched response.
        :rtype: cdp.network.Response
        """
        return (await self.response_future).response

    @property
    async def response_body(self) -> tuple[str, bool]:
        """
        Get the body of the matched response.
        :return: The response body.
        :rtype: str
        """
        request_id = (await self.response_future).request_id
        await (
            self.loading_finished_future
        )  # Ensure the loading is finished before fetching the body
        body = await self.tab.send(cdp.network.get_response_body(request_id=request_id))
        return body

loading_finished_future: asyncio.Future[cdp.network.LoadingFinished] = asyncio.Future() instance-attribute

request: cdp.network.Request async property

Get the matched request.

Returns:

Type Description
cdp.network.Request

The matched request.

request_future: asyncio.Future[cdp.network.RequestWillBeSent] = asyncio.Future() instance-attribute

request_id: Union[cdp.network.RequestId, None] = None instance-attribute

response: cdp.network.Response async property

Get the matched response.

Returns:

Type Description
cdp.network.Response

The matched response.

response_body: tuple[str, bool] async property

Get the body of the matched response.

Returns:

Type Description
str

The response body.

response_future: asyncio.Future[cdp.network.ResponseReceived] = asyncio.Future() instance-attribute

tab = tab instance-attribute

url_pattern = url_pattern instance-attribute

__aenter__() async

Enter the context manager, adding request and response handlers.

Source code in zendriver/core/expect.py 
83
84
85
86
87
88
async def __aenter__(self):  # type: ignore
    """
    Enter the context manager, adding request and response handlers.
    """
    await self._setup()
    return self

__aexit__(*args) async

Exit the context manager, removing request and response handlers.

Source code in zendriver/core/expect.py 
90
91
92
93
94
async def __aexit__(self, *args: Any) -> None:
    """
    Exit the context manager, removing request and response handlers.
    """
    self._teardown()

__init__(tab, url_pattern)

Source code in zendriver/core/expect.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, tab: Connection, url_pattern: Union[str, re.Pattern[str]]):
    self.tab = tab
    self.url_pattern = url_pattern
    self.request_future: asyncio.Future[cdp.network.RequestWillBeSent] = (
        asyncio.Future()
    )
    self.response_future: asyncio.Future[cdp.network.ResponseReceived] = (
        asyncio.Future()
    )
    self.loading_finished_future: asyncio.Future[cdp.network.LoadingFinished] = (
        asyncio.Future()
    )
    self.request_id: Union[cdp.network.RequestId, None] = None

reset() async

Resets the internal state, allowing the expectation to be reused.

Source code in zendriver/core/expect.py 
108
109
110
111
112
113
114
115
116
117
async def reset(self) -> None:
    """
    Resets the internal state, allowing the expectation to be reused.
    """
    self.request_future = asyncio.Future()
    self.response_future = asyncio.Future()
    self.loading_finished_future = asyncio.Future()
    self.request_id = None
    self._teardown()
    await self._setup()

DownloadExpectation

Source code in zendriver/core/expect.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
class DownloadExpectation:
    def __init__(self, tab: Connection):
        self.tab = tab
        self.future: asyncio.Future[cdp.browser.DownloadWillBegin] = asyncio.Future()
        # TODO: Improve
        self.default_behavior = (
            self.tab._download_behavior[0] if self.tab._download_behavior else "default"
        )
        self.download_path = (
            self.tab._download_behavior[1]
            if self.tab._download_behavior and len(self.tab._download_behavior) > 1
            else None
        )

    async def _handler(self, event: cdp.browser.DownloadWillBegin) -> None:
        self._remove_handler()
        self.future.set_result(event)

    def _remove_handler(self) -> None:
        self.tab.remove_handlers(cdp.browser.DownloadWillBegin, self._handler)

    async def __aenter__(self) -> "DownloadExpectation":
        """
        Enter the context manager, adding download handler, set download behavior to deny.
        """
        await self.tab.send(
            cdp.browser.set_download_behavior(behavior="deny", events_enabled=True)
        )
        self.tab.add_handler(cdp.browser.DownloadWillBegin, self._handler)
        return self

    async def __aexit__(self, *args: Any) -> None:
        """
        Exit the context manager, removing handler, set download behavior to default.
        """
        await self.tab.send(
            cdp.browser.set_download_behavior(
                behavior=self.default_behavior, download_path=self.download_path
            )
        )
        self._remove_handler()

    @property
    async def value(self) -> cdp.browser.DownloadWillBegin:
        return await self.future

default_behavior = self.tab._download_behavior[0] if self.tab._download_behavior else 'default' instance-attribute

download_path = self.tab._download_behavior[1] if self.tab._download_behavior and len(self.tab._download_behavior) > 1 else None instance-attribute

future: asyncio.Future[cdp.browser.DownloadWillBegin] = asyncio.Future() instance-attribute

tab = tab instance-attribute

value: cdp.browser.DownloadWillBegin async property

__aenter__() async

Enter the context manager, adding download handler, set download behavior to deny.

Source code in zendriver/core/expect.py 
209
210
211
212
213
214
215
216
217
async def __aenter__(self) -> "DownloadExpectation":
    """
    Enter the context manager, adding download handler, set download behavior to deny.
    """
    await self.tab.send(
        cdp.browser.set_download_behavior(behavior="deny", events_enabled=True)
    )
    self.tab.add_handler(cdp.browser.DownloadWillBegin, self._handler)
    return self

__aexit__(*args) async

Exit the context manager, removing handler, set download behavior to default.

Source code in zendriver/core/expect.py 
219
220
221
222
223
224
225
226
227
228
async def __aexit__(self, *args: Any) -> None:
    """
    Exit the context manager, removing handler, set download behavior to default.
    """
    await self.tab.send(
        cdp.browser.set_download_behavior(
            behavior=self.default_behavior, download_path=self.download_path
        )
    )
    self._remove_handler()

__init__(tab)

Source code in zendriver/core/expect.py 
189
190
191
192
193
194
195
196
197
198
199
200
def __init__(self, tab: Connection):
    self.tab = tab
    self.future: asyncio.Future[cdp.browser.DownloadWillBegin] = asyncio.Future()
    # TODO: Improve
    self.default_behavior = (
        self.tab._download_behavior[0] if self.tab._download_behavior else "default"
    )
    self.download_path = (
        self.tab._download_behavior[1]
        if self.tab._download_behavior and len(self.tab._download_behavior) > 1
        else None
    )

RequestExpectation

Bases: BaseRequestExpectation

Class for handling request expectations. This class extends BaseRequestExpectation and provides a property to access the matched request.

Parameters:

Name Type Description Default
tab Connection

The Tab instance to monitor.

 required
url_pattern Union[str, Pattern[str]]

The URL pattern to match requests.

 required
Source code in zendriver/core/expect.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class RequestExpectation(BaseRequestExpectation):
    """
    Class for handling request expectations.
    This class extends `BaseRequestExpectation` and provides a property to access the matched request.
    :param tab: The Tab instance to monitor.
    :param url_pattern: The URL pattern to match requests.
    """

    @property
    async def value(self) -> cdp.network.RequestWillBeSent:
        """
        Get the matched request event.
        :return: The matched request event.
        :rtype: cdp.network.RequestWillBeSent
        """
        return await self.request_future

value: cdp.network.RequestWillBeSent async property

Get the matched request event.

Returns:

Type Description
cdp.network.RequestWillBeSent

The matched request event.

ResponseExpectation

Bases: BaseRequestExpectation

Class for handling response expectations. This class extends BaseRequestExpectation and provides a property to access the matched response.

Parameters:

Name Type Description Default
tab Connection

The Tab instance to monitor.

 required
url_pattern Union[str, Pattern[str]]

The URL pattern to match responses.

 required
Source code in zendriver/core/expect.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class ResponseExpectation(BaseRequestExpectation):
    """
    Class for handling response expectations.
    This class extends `BaseRequestExpectation` and provides a property to access the matched response.
    :param tab: The Tab instance to monitor.
    :param url_pattern: The URL pattern to match responses.
    """

    @property
    async def value(self) -> cdp.network.ResponseReceived:
        """
        Get the matched response event.
        :return: The matched response event.
        :rtype: cdp.network.ResponseReceived
        """
        return await self.response_future

value: cdp.network.ResponseReceived async property

Get the matched response event.

Returns:

Type Description
cdp.network.ResponseReceived

The matched response event.