Method setUp

tests/test_exposition.py:223–273

(self)

class TestPushGateway(unittest.TestCase):
    def setUp(self):
        redirect_flag = 'testFlag'
        self.redirect_flag = redirect_flag  # preserve a copy for downstream test assertions
        self.registry = CollectorRegistry()
        self.counter = Gauge('g', 'help', registry=self.registry)
        self.requests = requests = []

        class TestHandler(BaseHTTPRequestHandler):
            def do_PUT(self):
                if 'with_basic_auth' in self.requestline and self.headers['authorization'] != 'Basic Zm9vOmJhcg==':
                    self.send_response(401)
                elif 'redirect' in self.requestline and redirect_flag not in self.requestline:
                    # checks for an initial test request with 'redirect' but without the redirect_flag,
                    # and simulates a redirect to a url with the redirect_flag (which will produce a 201)
                    self.send_response(301)
                    self.send_header('Location', getattr(self, 'redirect_address', None))
                else:
                    self.send_response(201)
                length = int(self.headers['content-length'])
                requests.append((self, self.rfile.read(length)))
                self.end_headers()

            do_POST = do_PUT
            do_DELETE = do_PUT

        # set up a separate server to serve a fake redirected request.
        # the redirected URL will have `redirect_flag` added to it,
        # which will cause the request handler to return 201.
        httpd_redirect = HTTPServer(('localhost', 0), TestHandler)
        self.redirect_address = TestHandler.redirect_address = \
            f'http://localhost:{httpd_redirect.server_address[1]}/{redirect_flag}'

        class TestRedirectServer(threading.Thread):
            def run(self):
                httpd_redirect.handle_request()

        self.redirect_server = TestRedirectServer()
        self.redirect_server.daemon = True
        self.redirect_server.start()

        # set up the normal server to serve the example requests across test cases.
        httpd = HTTPServer(('localhost', 0), TestHandler)
        self.address = f'http://localhost:{httpd.server_address[1]}'

        class TestServer(threading.Thread):
            def run(self):
                httpd.handle_request()

        self.server = TestServer()
        self.server.daemon = True
        self.server.start()


    def test_push(self):