Python小ネタ

小ネタ⑤　requests_mockの使い方

サードパーティ製のライブラリにはなりますが、requests_mockを使用するとrequestsにパッチを当てるのがとても簡単になります。

詳細は例によってコードコメントに書いています。

from datetime import datetime, timedelta
import json
import time
import requests
import requests_mock
import urllib.parse
import urllib.request

import ssl
ssl._create_default_https_context = ssl._create_unverified_context

URL1 = "http://localhost:12345/resource1"
URL2 = "https://www.mathkuro.com/"

SAMPLE_HEADER = {'Content-Type': 'application/json'}
SAMPLE_BODY = '{"name": "hoge", "email": "mail@example.com"}'


def send_get():
    # 今回のテスト対象
    return requests.get(URL1)

def send_post():
    # 今回のテスト対象
    return requests.post(URL1, json=json.loads(SAMPLE_BODY))

def test_multiple_get(requests_mock: requests_mock.Mocker):
    # 複数レスポンスがある場合
    requests_mock.register_uri("get", URL1, [{"text": "res1", "status_code": 200}, {"text": "res1", "status_code": 400}])
    requests_mock.register_uri("post", URL1, [{"status_code": 200}, {"status_code": 400}])
    ret = send_get()
    assert ret.status_code == 200

    ret = send_get()
    assert ret.status_code == 400

    ret = send_post()
    assert ret.status_code == 200

    # requestsの呼び出しの検証を行う場合
    assert requests_mock.call_count == 3
    for req in requests_mock.request_history:
        print(req.method, req.url, req.text)

start_dt = datetime.now() + timedelta(seconds=2)
def timered_succsess(request, context):
    context.status_code = 200
    context.headers["Content-Type"] = "application/json"
    if datetime.now() > start_dt:
        return SAMPLE_BODY
    return ''

def test_dynamic_response(requests_mock: requests_mock.Mocker):
    # 動的なレスポンスを設定したい場合
    requests_mock.register_uri("get", URL1, text=timered_succsess)
    while "" == send_get().text:
        time.sleep(1)

    assert send_get().text == SAMPLE_BODY

def test_with_normal_request(requests_mock: requests_mock.Mocker):
    # mockしたいresponseとmockにしたく無いresponseがある場合
    requests_mock.register_uri("get", URL1, [{"text": "res1", "status_code": 200}])
    ret = send_get()

    assert ret.status_code == 200

    normal_send()


def normal_send():
    req = urllib. request.Request(URL2, headers=SAMPLE_HEADER)

    # urlopenでリクエストを投げる。
    with urllib.request.urlopen(req) as f:

        # bodyは.read()で取得可能。(bytesオブジェクト)
        body = f.read()

        # 例えばbodyがjson形式ならjson.loads()でpythonオブジェクトに変換可能。
        print(len(body))
        print(f.info())

        # 自分がどんなURLを投げたのか確認したい時
        print(f.geturl())