Daniel's Blog

Creating a Mock ZXY Server for testing

Issue

The client's unit tests are running against Open Street Map's servers. In the code you can even see a comment with an issue with the OSM:

def Test(TestCase): 
    
    ...

    def setUp(self): 

        ...

        # OSM servers some times give 429 errors
        self.url = ('https://cartodb-basemaps-a.global.ssl.fastly.net'
                     '/dark_all/{Z}/{X}/{Y}.png')
        self.test_base_layer = BaseLayer(
            url=self.url, name='Open Street Map',
            project=self.test_project, layer_name='url',
            added_by=self.test_user, source='zxy',
            last_updated_by=self.test_user
        )

This is problematic for several reasons:

  1. Using an external service that may fail
  2. It has failed and that's known due to the code comment about 429 (Too Many Requests) errors
  3. With 429 errors it means that the tests potentially could abuse the external server (not nice!)
  4. The tests were failing so often, that the test was commented them out so the code isn't tested.

The fix

The tests simply need a Mock ZXY server to run. The tests don't care if the correct image is returned, just that their code is able to retrieve the image. So the easiest case is to simply create a mock server that returns a 256x256 image (which is the normal size of an image). A png will work fine for this use case and that can be done in memory using PIL or Pillow as PIL was deprecated.

class MockZXYRequestHandler(BaseHTTPRequestHandler):
    """
    This mock ZXY server will always return a white 256x256 image regardless of the
    request made. It is used in place of a real image server.
    """

    def do_GET(self):
        # Process an HTTP GET request and return a response with an HTTP 200 status and a White png tile
        # as an image

        heir_path = self.path
        if '?' in self.path:
            heir_path = self.path.split('?')[0]

        if heir_path.startswith('/zxy'):
            self.send_response(requests.codes.ok)
            self.send_header('Content-type', 'image/png')
            self.end_headers()

            new_image = Image.new("RGB", (256, 256), (255, 255, 255))

            temp_buffer = BytesIO()
            new_image.save(temp_buffer, format="PNG")
            self.wfile.write(temp_buffer.getvalue())

            return

        heir_path = heir_path.rstrip('/')

        return

Any request made to this server in the /zxy path is going to return an image. The server doesn't care about the actual ZXY values as it's not the test's job to ensure it's sending a correct URL. Generating the ZXY Url correctly can be another test, requesting a tile should only be tested once the URL is generated correctly. So it just doesn't care and that's fine.

Using the server is rather straightforward.

First a function go get a free port on the local host is needed.

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

Then launch the server in another thread during the test:

def TestSomething(TestCase): 

    def setUp(self)
        # Configure mock server.
        self.mock_server_port = get_free_port()
        self.mock_server = HTTPServer(('localhost', self.mock_server_port), MockZXYServerRequestHandler)

        # Start a mock ZXY server to use instead of an external server.
        self.mock_server_thread = Thread(target=self.mock_server.serve_forever)
        self.mock_server_thread.setDaemon(True)
        self.mock_server_thread.start()

    def tearDown(self):

        self.mock_server.shutdown()
        self.mock_server_thread.join()

Once that is setup a test can be written using the server

def TestSomething(TestCase): 

    ...

    def test_request_to_zxy_server_returns_image(self): 
        """ 
        Request the 0 0 0  tile from the ZXY server and ensure a succesful response and image are returned
        """ 
        url = 'http://localhost:{port}/zxy/{{Z}}/{{X}}/{{Y}}.png'.format(port=self.mock_server_port)
        preview_url = url.format(Z='0', X='0', Y='0')
        response = requests.get(preview_url)
        self.assertEqual(response.status_code, 200)

        img = Image.open(BytesIO(response.content))
        self.assertEqual(img.height, 256)
        self.assertEqual(img.width, 256)

Conclusion

Now the tests can run, don't abuse external services, run quicker and more stable, and will run even if the external service is down.