retry helper for BrokenPipeError in selenium

July 19, 2018, 7:49 a.m.   wschaub   django TDD  

While going over chapter 21 of Test-Driven Development with Python I kept getting failing tests with 

BrokenPipeError: [Errno 32] Broken pipe exception

This article describes my workaround for this problem.

This error usually happens randomly on self.browser.get calls. I wanted a way to catch the error and retry and this is what I have come up with that seems to work so far. I created a decorator similar to the existing @wait decorator we already use at this point in the book.

functional_tests/base.py

import sys

...

def brokenpipe_retry(fn):
    """Catch BrokenPipeErrors and restart the function up to 3 times"""
    def modified_fn(*args, **kwargs):
        retrycount = 1
        while True:
            try:
                return fn(*args, **kwargs)
            except BrokenPipeError as e:
                if retrycount > 3:
                    raise e #give up and let the exception bubble up.
                print("Caught BrokenPipeError. Retry", retrycount,
                      "of 3", file=sys.stderr)
                retrycount += 1
            else: #Called only when no exceptions caught.
                break
    return modified_fn

def wait(fn):
...

class FunctionalTest(StaticLiveServerTestCase):
...

    @brokenpipe_retry
    def retry_helper(self, fn):
        return fn()

    @wait
    def wait_for(self, fn):
        return fn()

...

functional_tests/test_login.py

class LoginTest(FunctionalTest):
...
        #she clicks it
        ## get around intermittent selenium BrokenPipeError exceptions
        self.retry_helper(lambda: self.browser.get(url))

And finally since this is about tests I wrote my very first test that wasn't guided by the book:

functiional_tests/test_brokenpipe_retry.py

from unittest import TestCase
from .base import brokenpipe_retry

class TestBrokenPipeRetry(TestCase):

    def throws_error_always(self):
        raise BrokenPipeError

    @brokenpipe_retry                                                                                                                                                   
    def retry_helper(self, fn):                                                                                                                                         
        return fn()                                                                                                                                                     

    def test_throws_error_on_max_retry_count(self):
        with self.assertRaises(BrokenPipeError):
            self.retry_helper(
                lambda: self.throws_error_always()
            )

    def test_successfuly_retries(self):
        retry = 2
        def fail_twice():
            nonlocal retry
            if retry > 0:
                retry -= 1
                raise BrokenPipeError
            return True

        is_true = self.retry_helper(lambda: fail_twice())
        self.assertEqual(is_true, True)
TDD python python-tdd-book
William Schaub
Software Engineer

Blog Search