Skip to content
This repository has been archived by the owner on Jan 3, 2024. It is now read-only.

Add methods for deleting/clearing a single request only; add check_response parameter to wait_for_request #639

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -230,8 +230,8 @@ Selenium Wire captures all HTTP/HTTPS traffic made by the browser [1]_. The foll
``driver.last_request``
Convenience attribute for retrieving the most recently captured request. This is more efficient than using ``driver.requests[-1]``.

``driver.wait_for_request(pat, timeout=10)``
This method will wait until it sees a request matching a pattern. The ``pat`` attribute will be matched within the request URL. ``pat`` can be a simple substring or a regular expression. Note that ``driver.wait_for_request()`` doesn't *make* a request, it just *waits* for a previous request made by some other action and it will return the first request it finds. Also note that since ``pat`` can be a regular expression, you must escape special characters such as question marks with a slash. A ``TimeoutException`` is raised if no match is found within the timeout period.
``driver.wait_for_request(pat, timeout=10, check_response=False)``
This method will wait until it sees a request matching a pattern. The ``pat`` attribute will be matched within the request URL. ``pat`` can be a simple substring or a regular expression. Note that ``driver.wait_for_request()`` doesn't *make* a request, it just *waits* for a previous request made by some other action and it will return the first request it finds. Also note that since ``pat`` can be a regular expression, you must escape special characters such as question marks with a slash. The ``check_response`` parameter controlls whether this method should wait until a request has received a response. A ``TimeoutException`` is raised if no match is found within the timeout period.

For example, to wait for an AJAX request to return after a button is clicked:

Expand Down Expand Up @@ -263,6 +263,12 @@ To clear previously captured requests and HAR entries, use ``del``:

del driver.requests

Alternatively, you can discard a specific captured request using ``.delete_request()``:

.. code:: python

driver.delete_request(request.id)

.. [1] Selenium Wire ignores OPTIONS requests by default, as these are typically uninteresting and just add overhead. If you want to capture OPTIONS requests, you need to set the ``ignore_http_methods`` `option`_ to ``[]``.

.. _`option`: #all-options
Expand Down
19 changes: 17 additions & 2 deletions seleniumwire/inspect.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ def last_request(self) -> Optional[Request]:
"""
return self.backend.storage.load_last_request()

def wait_for_request(self, pat: str, timeout: Union[int, float] = 10) -> Request:
def wait_for_request(self, pat: str, timeout: Union[int, float] = 10, check_response: bool = False) -> Request:
"""Wait up to the timeout period for a request matching the specified
pattern to be seen.

Expand All @@ -63,6 +63,7 @@ def wait_for_request(self, pat: str, timeout: Union[int, float] = 10) -> Request
Args:
pat: The pat of the request to look for. A regex can be supplied.
timeout: The maximum time to wait in seconds. Default 10s.
check_response: If nonzero, only match requests with a response.

Returns:
The request.
Expand All @@ -73,7 +74,7 @@ def wait_for_request(self, pat: str, timeout: Union[int, float] = 10) -> Request
start = time.time()

while time.time() - start < timeout:
request = self.backend.storage.find(pat)
request = self.backend.storage.find(pat, check_response=check_response)

if request is None:
time.sleep(1 / 5)
Expand All @@ -82,6 +83,20 @@ def wait_for_request(self, pat: str, timeout: Union[int, float] = 10) -> Request

raise TimeoutException('Timed out after {}s waiting for request matching {}'.format(timeout, pat))

def delete_request(self, request_id) -> None:
"""Clear a specific request.

Args:
request_id: id of request to clear.
Raises:
KeyError if no request matching the specified id could be found
"""
# I believe that changing the 'requests* property to return an
# instance supporting __delitem__ would provide a better API for
# users, but this may break some backwards compatibilty or lead
# to unexpected behavior.
self.backend.storage.clear_request_by_id(request_id)

@property
def har(self) -> str:
"""Get a HAR archive of HTTP transactions that have taken place.
Expand Down
31 changes: 31 additions & 0 deletions seleniumwire/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,24 @@ def clear_requests(self) -> None:
for indexed_request in index:
shutil.rmtree(self._get_request_dir(indexed_request.id), ignore_errors=True)

def clear_request_by_id(self, request_id: str) -> None:
"""Clear a specific request.

Args:
request_id: id of request to clear.
Raises:
KeyError if no request matching the specified id could be found
"""
with self._lock:
for i, indexed_request in enumerate(self._index):
if indexed_request.id == request_id:
break
else:
raise KeyError("Could not find any request with the specified id '{}'!".format(request_id))
del self._index[i] # i will be index at break

shutil.rmtree(self._get_request_dir(indexed_request.id), ignore_errors=True)

def find(self, pat: str, check_response: bool = True) -> Optional[Request]:
"""Find the first request that matches the specified pattern.

Expand Down Expand Up @@ -498,6 +516,19 @@ def clear_requests(self) -> None:
with self._lock:
self._requests.clear()

def clear_request_by_id(self, request_id: str) -> None:
"""Clear a specific request.

Args:
request_id: id of request to clear.
Raises:
KeyError if no request matching the specified id could be found
"""
with self._lock:
if request_id not in self._requests:
raise KeyError("Could not find any request with the specified id '{}'!".format(request_id))
del self._requests[request_id]

def find(self, pat: str, check_response: bool = True) -> Optional[Request]:
"""Find the first request that matches the specified pattern.

Expand Down
15 changes: 14 additions & 1 deletion tests/seleniumwire/test_inspect.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,11 @@ def test_delete_requests(self):

self.mock_backend.storage.clear_requests.assert_called_once_with()

def test_delete_request(self):
self.driver.delete_request("foo")

self.mock_backend.storage.clear_request_by_id.assert_called_once_with("foo")

def test_iter_requests(self):
self.mock_backend.storage.iter_requests.return_value = iter([Mock()])

Expand Down Expand Up @@ -59,7 +64,15 @@ def test_wait_for_request(self):
request = self.driver.wait_for_request('/some/path')

self.assertIsNotNone(request)
self.mock_backend.storage.find.assert_called_once_with('/some/path')
self.mock_backend.storage.find.assert_called_once_with('/some/path', check_response=False)

def test_wait_for_request_with_response(self):
self.mock_backend.storage.find.return_value = Mock()

request = self.driver.wait_for_request('/some/path', check_response=True)

self.assertIsNotNone(request)
self.mock_backend.storage.find.assert_called_once_with('/some/path', check_response=True)

def test_wait_for_request_timeout(self):
self.mock_backend.storage.find.return_value = None
Expand Down
25 changes: 25 additions & 0 deletions tests/seleniumwire/test_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,19 @@ def test_clear_requests(self):
self.assertFalse(requests)
self.assertFalse(glob.glob(os.path.join(self.base_dir, '.seleniumwire', 'storage-*', '*')))

def test_clear_request_by_id(self):
request_1 = self._create_request()
request_2 = self._create_request()
self.storage.save_request(request_1)
self.storage.save_request(request_2)

self.storage.clear_request_by_id(request_2.id)
requests = self.storage.load_requests()

self.assertEqual(len(requests), 1) # only request_2 should have been cleared
self.assertEqual(requests[0].id, request_1.id)
self.assertEqual(len(glob.glob(os.path.join(self.base_dir, '.seleniumwire', 'storage-*', '*'))), 1)

def test_get_home_dir(self):
self.assertEqual(os.path.join(self.base_dir, '.seleniumwire'), self.storage.home_dir)

Expand Down Expand Up @@ -490,6 +503,18 @@ def test_clear_requests(self):

self.assertFalse(requests)

def test_clear_request_by_id(self):
request_1 = self._create_request()
request_2 = self._create_request()
self.storage.save_request(request_1)
self.storage.save_request(request_2)

self.storage.clear_request_by_id(request_2.id)
requests = self.storage.load_requests()

self.assertEqual(len(requests), 1) # only request_2 should have been cleared
self.assertEqual(requests[0].id, request_1.id)

def test_cleanup(self):
request_1 = self._create_request()
request_2 = self._create_request()
Expand Down