Create mock API

You can program your API server. Write a test that describes the behavior:

# Third-party imports...
from nose.tools import assert_true
import requests


def test_request_response():
    url = 'http://localhost:{port}/users'.format(port=mock_server_port)

    # Send a request to the mock API server and store the response.
    response = requests.get(url)

    # Confirm that the request-response cycle completed successfully.
    assert_true(response.ok)

凡事總是沒有那麼完美

但這個 testcase 往往是需要 real 的 API 來做測試,
但假設手邊沒有現成的 API 可以測試該怎麼辦呢?
我們就必須透過我們萬能的雙手去完成
首先我們可以先參考 BaseHTTPRequestHandler

程式碼範例如下

# Standard library imports...
from http.server import BaseHTTPRequestHandler, HTTPServer
import socket
from threading import Thread
import logging

# Third-party imports...
from nose.tools import assert_true
import requests
from requests.sessions import HTTPAdapter, session
from urllib3.util import Retry


class MockServerRequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Process an HTTP GET request and return a response with an HTTP 200 status.
        self.send_response(requests.codes.ok)
        self.end_headers()
        return


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

利用繼承 BaseHTTPRequestHandler
來做到 GET 的 API 讓他回傳 OK(200)

但如果我今天想做很多不同的API呢?

假設你今天想要自己做一個 POST or 其他的就可以參考 BaseHTTPRequestHandler 的物件
在 MockServerRequestHandler 裡面去定義函式:

def do_POST(self):
        # Process an HTTP POST request and return a response with an HTTP 400 status.
        self.send_response(requests.codes.bad)
        self.end_headers()

這個函式定義了 POST 的回傳值 = bad
也就是回傳 400

回傳值的參考

如果想自己定義甚麼樣的 return value 可以參考
library requests 裡面:

# Informational.
    100: ('continue',),
    101: ('switching_protocols',),
    102: ('processing',),
    103: ('checkpoint',),
    122: ('uri_too_long', 'request_uri_too_long'),
    200: ('ok', 'okay', 'all_ok', 'all_okay', 'all_good', '\\o/', '✓'),
    201: ('created',),
    202: ('accepted',),
    203: ('non_authoritative_info', 'non_authoritative_information'),
    204: ('no_content',),
    205: ('reset_content', 'reset'),
    206: ('partial_content', 'partial'),
    207: ('multi_status', 'multiple_status', 'multi_stati', 'multiple_stati'),
    208: ('already_reported',),
    226: ('im_used',),

    # Redirection.
    300: ('multiple_choices',),
    301: ('moved_permanently', 'moved', '\\o-'),
    302: ('found',),
    303: ('see_other', 'other'),
    304: ('not_modified',),
    305: ('use_proxy',),
    306: ('switch_proxy',),
    307: ('temporary_redirect', 'temporary_moved', 'temporary'),
    308: ('permanent_redirect',
          'resume_incomplete', 'resume',),  # These 2 to be removed in 3.0

    # Client Error.
    400: ('bad_request', 'bad'),
    401: ('unauthorized',),
    402: ('payment_required', 'payment'),
    403: ('forbidden',),
    404: ('not_found', '-o-'),
    405: ('method_not_allowed', 'not_allowed'),
    406: ('not_acceptable',),
    407: ('proxy_authentication_required', 'proxy_auth', 'proxy_authentication'),
    408: ('request_timeout', 'timeout'),
    409: ('conflict',),
    410: ('gone',),
    411: ('length_required',),
    412: ('precondition_failed', 'precondition'),
    413: ('request_entity_too_large',),
    414: ('request_uri_too_large',),
    415: ('unsupported_media_type', 'unsupported_media', 'media_type'),
    416: ('requested_range_not_satisfiable', 'requested_range', 'range_not_satisfiable'),
    417: ('expectation_failed',),
    418: ('im_a_teapot', 'teapot', 'i_am_a_teapot'),
    421: ('misdirected_request',),
    422: ('unprocessable_entity', 'unprocessable'),
    423: ('locked',),
    424: ('failed_dependency', 'dependency'),
    425: ('unordered_collection', 'unordered'),
    426: ('upgrade_required', 'upgrade'),
    428: ('precondition_required', 'precondition'),
    429: ('too_many_requests', 'too_many'),
    431: ('header_fields_too_large', 'fields_too_large'),
    444: ('no_response', 'none'),
    449: ('retry_with', 'retry'),
    450: ('blocked_by_windows_parental_controls', 'parental_controls'),
    451: ('unavailable_for_legal_reasons', 'legal_reasons'),
    499: ('client_closed_request',),

    # Server Error.
    500: ('internal_server_error', 'server_error', '/o\\', '✗'),
    501: ('not_implemented',),
    502: ('bad_gateway',),
    503: ('service_unavailable', 'unavailable'),
    504: ('gateway_timeout',),
    505: ('http_version_not_supported', 'http_version'),
    506: ('variant_also_negotiates',),
    507: ('insufficient_storage',),
    509: ('bandwidth_limit_exceeded', 'bandwidth'),
    510: ('not_extended',),
    511: ('network_authentication_required', 'network_auth', 'network_authentication'),
}

Example of Customize API

因此這邊小結
我們客製化的 API 如下:

# Standard library imports...
from http.server import BaseHTTPRequestHandler, HTTPServer
import socket
from threading import Thread
import logging

# Third-party imports...
from nose.tools import assert_true
import requests
from requests.sessions import HTTPAdapter, session
from urllib3.util import Retry


class MockServerRequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Process an HTTP GET request and return a response with an HTTP 200 status.
        self.send_response(requests.codes.ok)
        self.end_headers()
        return

    def do_POST(self):
        # Process an HTTP POST request and return a response with an HTTP 200 status.
        self.send_response(requests.codes.bad)
        self.end_headers()
        return


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

Implement Testcase

再來我們可以利用nosetest的方式
把 testcase完成:

class TestMockServer(object):
    @classmethod
    def setup_class(cls):
        # Configure mock server.
        cls.mock_server_port = get_free_port()
        cls.mock_server = HTTPServer(('localhost', cls.mock_server_port), MockServerRequestHandler)

        # Start running mock server in a separate thread.
        # Daemon threads automatically shut down when the main process exits.
        cls.mock_server_thread = Thread(target=cls.mock_server.serve_forever)
        cls.mock_server_thread.setDaemon(True)
        cls.mock_server_thread.start()

    def test_request_response(self):
        url = 'http://localhost:{port}/users'.format(port=self.mock_server_port)

        # Normal version
        # Send a request to the mock API server and store the response.
        response = requests.get(url)

        # Confirm that the request-response cycle completed successfully.
        print(f"response: {response}")
        assert_true(response.ok)

開始執行測試案例囉

在下指令去跑testcase
nosetests --verbosity=2 .\api_helper.py -v -s

可以得到結果:

(.venv) PS E:\side_pro> nosetests --verbosity=2 .\api_helper.py -v -s
nose.config: INFO: Ignoring files matching ['^\\.', '^_', '^setup\\.py$']
api_helper.TestMockServer.test_request_response ... 127.0.0.1 - - [29/May/2021 23:39:43] "GET /users HTTP/1.1" 200 -
response: <Response [200]>
ok

----------------------------------------------------------------------
Ran 1 test in 2.038s

是不是成功打出去了呢!!!

Test with nosetest and auto retry

有個需求是
當今天 API 因為某種原因暫時壞掉 或是暫時維護
但又不是希望 testcase 要重跑一次
(因為這個 API 不算是測項之一)
那我們會希望
假設打的當下不如預期 應該可以自動retry

requests module 神救援

Request hooks

Often when using a third party API you want to verify that the returned response is indeed valid. Requests offers the shorthand helper raise_for_status() which asserts that the response HTTP status code is not a 4xx or a 5xx, i.e that the request didn't result in a client or a server error.

For example

response = requests.get('https://api.github.com/user/repos?page=1')
# Assert that there were no errors
response.raise_for_status()

This can get repetitive if you need to raise_for_status() for each call. Luckily the requests library offers a 'hooks' interface where you can attach callbacks on certain parts of the request process.

We can use hooks to ensure raise_for_status() is called for each response object.

For Example

# Create a custom requests object, modifying the global module throws an error
http = requests.Session()

assert_status_hook = lambda response, *args, **kwargs: response.raise_for_status()
http.hooks["response"] = [assert_status_hook]

http.get("https://api.github.com/user/repos?page=1")

> HTTPError: 401 Client Error: Unauthorized for url: https://api.github.com/user/repos?page=1

Retry on failure

Network connections are lossy, congested and servers fail. If we want to build a truly robust program we need to account for failures and have a retry strategy.

Add a retry strategy to your HTTP client is straightforward. We create a HTTPAdapter and pass our strategy to the adapter.

# Implement retry mechanism
        session = requests.Session()
        retry_strategy = Retry(total=5,
                               backoff_factor=1,
                               status_forcelist=[400],
                               method_whitelist=['POST'])

        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        response = session.post(url=url)

The default Retry class offers sane defaults, but is highly configurable so here is a rundown of the most common parameters I use.

The parameters below include the default parameters the requests library uses.

total=10
The total number of retry attempts to make. If the number of failed requests or redirects exceeds this number the client will throw the urllib3.exceptions.MaxRetryError exception. I vary this parameter based on the API I'm working with, but I usually set it to lower than 10, usually 3 retries is enough.

status_forcelist=[413, 429, 503]
The HTTP response codes to retry on. You likely want to retry on the common server errors (500, 502, 503, 504) because servers and reverse proxies don't always adhere to the HTTP spec. Always retry on 429 rate limit exceeded because the urllib library should by default incrementally backoff on failed requests.

method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]
The HTTP methods to retry on. By default this includes all HTTP methods except POST because POST can result in a new insert. Modify this parameter to include POST because most API's I deal with don't return an error code and perform an insert in the same call. And if they do, you should probably issue a bug report.

backoff_factor=0
This is an interesting one. It allows you to change how long the processes will sleep between failed requests. The algorithm is as follows:

{backoff factor} * (2 ** ({number of total retries} - 1))
For example, if the backoff factor is set to:

• 1 second the successive sleeps will be 0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256.
• 2 seconds - 1, 2, 4, 8, 16, 32, 64, 128, 256, 512
• 10 seconds - 5, 10, 20, 40, 80, 160, 320, 640, 1280, 2560

The value is exponentially increasing which is a sane default implementation for retry strategies.

This value is by default 0, meaning no exponential backoff will be set and retries will immediately execute. Make sure to set this to 1 in to avoid hammering your servers!.

The full documentation on the retry module is here

Combining timeouts and retries

Since the HTTPAdapter is comparable we can combine retries and timeouts like so:

retries = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
http.mount("https://", TimeoutHTTPAdapter(max_retries=retries))

完整程式代碼

# Standard library imports...
from http.server import BaseHTTPRequestHandler, HTTPServer
import socket
from threading import Thread
import logging

# Third-party imports...
from nose.tools import assert_true
import requests
from requests.sessions import HTTPAdapter, session
from urllib3.util import Retry


class MockServerRequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Process an HTTP GET request and return a response with an HTTP 200 status.
        self.send_response(requests.codes.ok)
        self.end_headers()
        return

    def do_POST(self):
        # Process an HTTP POST request and return a response with an HTTP 200 status.
        self.send_response(requests.codes.bad)
        self.end_headers()
        return


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 TestMockServer(object):
    @classmethod
    def setup_class(cls):
        # Configure mock server.
        cls.mock_server_port = get_free_port()
        cls.mock_server = HTTPServer(('localhost', cls.mock_server_port), MockServerRequestHandler)

        # Start running mock server in a separate thread.
        # Daemon threads automatically shut down when the main process exits.
        cls.mock_server_thread = Thread(target=cls.mock_server.serve_forever)
        cls.mock_server_thread.setDaemon(True)
        cls.mock_server_thread.start()

    def test_request_response(self):
        url = 'http://localhost:{port}/users'.format(port=self.mock_server_port)

        # Implement retry mechanism
        session = requests.Session()
        retry_strategy = Retry(total=5,
                               backoff_factor=1,
                               status_forcelist=[400],
                               method_whitelist=['POST'])

        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        response = session.post(url=url)

        # Normal version
        # Send a request to the mock API server and store the response.
        response = requests.get(url)

        # Confirm that the request-response cycle completed successfully.
        print(f"response: {response}")
        assert_true(response.ok)

執行結果

nose.config: INFO: Ignoring files matching ['^\\.', '^_', '^setup\\.py$']
api_helper.TestMockServer.test_request_response ... 127.0.0.1 - - [30/May/2021 01:28:48] "POST /users HTTP/1.1" 400 -
127.0.0.1 - - [30/May/2021 01:28:50] "POST /users HTTP/1.1" 400 -
127.0.0.1 - - [30/May/2021 01:28:54] "POST /users HTTP/1.1" 400 -
127.0.0.1 - - [30/May/2021 01:29:00] "POST /users HTTP/1.1" 400 -
127.0.0.1 - - [30/May/2021 01:29:10] "POST /users HTTP/1.1" 400 -
127.0.0.1 - - [30/May/2021 01:29:28] "POST /users HTTP/1.1" 400 -
ERROR

======================================================================
ERROR: api_helper.TestMockServer.test_request_response
----------------------------------------------------------------------
Traceback (most recent call last):
  File "e:\side_pro\.venv\lib\site-packages\nose\case.py", line 198, in runTest
    self.test(*self.arg)
  File "E:\side_pro\api_helper.py", line 62, in test_request_response
    response = session.post(url=url)
  File "e:\side_pro\.venv\lib\site-packages\requests\sessions.py", line 590, in post
    return self.request('POST', url, data=data, json=json, **kwargs)
  File "e:\side_pro\.venv\lib\site-packages\requests\sessions.py", line 542, in request
    resp = self.send(prep, **send_kwargs)
  File "e:\side_pro\.venv\lib\site-packages\requests\sessions.py", line 655, in send
    r = adapter.send(request, **kwargs)
  File "e:\side_pro\.venv\lib\site-packages\requests\adapters.py", line 507, in send
    raise RetryError(e, request=request)
requests.exceptions.RetryError: HTTPConnectionPool(host='localhost', port=65011): Max retries exceeded with url: /users (Caused by ResponseError('too many 400 error responses'))
-------------------- >> begin captured logging << --------------------
urllib3.connectionpool: DEBUG: Starting new HTTP connection (1): localhost:65011
urllib3.connectionpool: DEBUG: http://localhost:65011 "POST /users HTTP/1.1" 400 None
urllib3.util.retry: DEBUG: Incremented Retry for (url='/users'): Retry(total=4, connect=None, read=None, redirect=None, status=None)
urllib3.connectionpool: DEBUG: Retry: /users
urllib3.connectionpool: DEBUG: Resetting dropped connection: localhost
urllib3.connectionpool: DEBUG: http://localhost:65011 "POST /users HTTP/1.1" 400 None
urllib3.util.retry: DEBUG: Incremented Retry for (url='/users'): Retry(total=3, connect=None, read=None, redirect=None, status=None)
urllib3.connectionpool: DEBUG: Retry: /users
urllib3.connectionpool: DEBUG: Resetting dropped connection: localhost
urllib3.connectionpool: DEBUG: http://localhost:65011 "POST /users HTTP/1.1" 400 None
urllib3.util.retry: DEBUG: Incremented Retry for (url='/users'): Retry(total=2, connect=None, read=None, redirect=None, status=None)
urllib3.connectionpool: DEBUG: Retry: /users
urllib3.connectionpool: DEBUG: Resetting dropped connection: localhost
urllib3.connectionpool: DEBUG: http://localhost:65011 "POST /users HTTP/1.1" 400 None
urllib3.util.retry: DEBUG: Incremented Retry for (url='/users'): Retry(total=1, connect=None, read=None, redirect=None, status=None)
urllib3.connectionpool: DEBUG: Retry: /users
urllib3.connectionpool: DEBUG: Resetting dropped connection: localhost
urllib3.connectionpool: DEBUG: http://localhost:65011 "POST /users HTTP/1.1" 400 None
urllib3.util.retry: DEBUG: Incremented Retry for (url='/users'): Retry(total=0, connect=None, read=None, redirect=None, status=None)
urllib3.connectionpool: DEBUG: Retry: /users
urllib3.connectionpool: DEBUG: Resetting dropped connection: localhost
urllib3.connectionpool: DEBUG: http://localhost:65011 "POST /users HTTP/1.1" 400 None
--------------------- >> end captured logging << ---------------------

----------------------------------------------------------------------
Ran 1 test in 42.048s

FAILED (errors=1)

Reference:
https://realpython.com/testing-third-party-apis-with-mock-servers/
https://findwork.dev/blog/advanced-usage-python-requests-timeouts-retries-hooks/