diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 000000000..73e36596d --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,84 @@ +import sys +import socket +from threading import Thread +import json + +import stripe +import pytest + +if sys.version_info[0] < 3: + from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer +else: + from http.server import BaseHTTPRequestHandler, HTTPServer + + +def get_free_port(): + s = socket.socket(socket.AF_INET, type=socket.SOCK_STREAM) + s.bind(("localhost", 0)) + address, port = s.getsockname() + s.close() + return port + + +class TestIntegration(object): + @pytest.fixture(autouse=True) + def close_mock_server(self): + yield + if self.mock_server: + self.mock_server.shutdown() + self.mock_server_thread.join() + + def setup_mock_server(self, handler): + # Configure mock server. + self.mock_server_port = get_free_port() + self.mock_server = HTTPServer( + ("localhost", self.mock_server_port), handler + ) + + # Start running mock server in a separate thread. + # Daemon threads automatically shut down when the main process exits. + self.mock_server_thread = Thread(target=self.mock_server.serve_forever) + self.mock_server_thread.setDaemon(True) + self.mock_server_thread.start() + + def test_hits_api_base(self): + class MockServerRequestHandler(BaseHTTPRequestHandler): + num_requests = 0 + + def do_GET(self): + self.__class__.num_requests += 1 + + self.send_response(200) + self.send_header( + "Content-Type", "application/json; charset=utf-8" + ) + self.end_headers() + self.wfile.write(json.dumps({}).encode("utf-8")) + return + + self.setup_mock_server(MockServerRequestHandler) + + stripe.api_base = "http://localhost:%s" % self.mock_server_port + stripe.Balance.retrieve() + assert MockServerRequestHandler.num_requests == 1 + + def test_hits_stripe_proxy(self): + class MockServerRequestHandler(BaseHTTPRequestHandler): + num_requests = 0 + + def do_GET(self): + self.__class__.num_requests += 1 + + self.send_response(200) + self.send_header( + "Content-Type", "application/json; charset=utf-8" + ) + self.end_headers() + self.wfile.write(json.dumps({}).encode("utf-8")) + return + + self.setup_mock_server(MockServerRequestHandler) + + stripe.proxy = "http://localhost:%s" % self.mock_server_port + stripe.Balance.retrieve() + assert MockServerRequestHandler.num_requests == 1