python-mocket/tests/test_https.py at 26e596b0dc136c31c7b25778317c2b94e3fd7183 · mindflayer/python-mocket · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
import json
import os
import tempfile
from urllib.request import urlopen

import pytest
import requests

from mocket import Mocket, Mocketizer, mocketize
from mocket.mockhttp import Entry  # noqa - test retrocompatibility


@pytest.fixture
def url_to_mock():
    return "https://httpbin.org/ip"


@pytest.fixture
def response():
    return {
        "integer": 1,
        "string": "asd",
        "boolean": False,
    }


@mocketize
def test_json(response):
    url_to_mock = "https://testme.org/json"

    Entry.single_register(
        Entry.GET,
        url_to_mock,
        body=json.dumps(response),
        headers={"content-type": "application/json"},
    )

    mocked_response = requests.get(url_to_mock).json()
    assert response == mocked_response

    mocked_response = json.loads(urlopen(url_to_mock).read().decode("utf-8"))
    assert response == mocked_response


@pytest.mark.skipif('os.getenv("SKIP_TRUE_HTTP", False)')
@pytest.mark.xfail(reason="Service down or blocking GitHub actions IPs")
def test_truesendall_with_recording_https(url_to_mock):
    with tempfile.TemporaryDirectory() as temp_dir, Mocketizer(
        truesocket_recording_dir=temp_dir
    ):
        requests.get(url_to_mock, headers={"Accept": "application/json"})
        resp = requests.get(url_to_mock, headers={"Accept": "application/json"})
        assert resp.status_code == 200

        dump_filename = os.path.join(
            Mocket.get_truesocket_recording_dir(),
            Mocket.get_namespace() + ".json",
        )
        with open(dump_filename) as f:
            responses = json.load(f)

    assert len(responses["httpbin.org"]["443"].keys()) == 1


@pytest.mark.skipif('os.getenv("SKIP_TRUE_HTTP", False)')
@pytest.mark.xfail(reason="Service down or blocking GitHub actions IPs")
def test_truesendall_after_mocket_session(url_to_mock):
    Mocket.enable()
    Mocket.disable()

    resp = requests.get(url_to_mock)
    assert resp.status_code == 200


@pytest.mark.skipif('os.getenv("SKIP_TRUE_HTTP", False)')
@pytest.mark.xfail(reason="Service down or blocking GitHub actions IPs")
def test_real_request_session(url_to_mock):
    session = requests.Session()

    url_to_compare = "http://httpbin.org/headers"

    with Mocketizer():
        assert len(session.get(url_to_mock).content) < len(
            session.get(url_to_compare).content
        )


@mocketize
def test_raise_exception_from_single_register():
    url = "https://github.com/fluidicon.png"
    Entry.single_register(Entry.GET, url, exception=OSError())
    with pytest.raises(requests.exceptions.ConnectionError):
        requests.get(url)


@mocketize
def test_can_handle():
    Entry.single_register(
        Entry.GET,
        "https://httpbin.org",
        body=json.dumps({"message": "Nope... not this time!"}),
        headers={"content-type": "application/json"},
        can_handle_fun=lambda path, qs_dict: path == "/ip" and qs_dict,
    )
    Entry.single_register(
        Entry.GET,
        "https://httpbin.org",
        body=json.dumps({"message": "There you go!"}),
        headers={"content-type": "application/json"},
        can_handle_fun=lambda path, qs_dict: path == "/ip" and not qs_dict,
    )
    resp = requests.get("https://httpbin.org/ip")
    assert resp.status_code == 200
    assert resp.json() == {"message": "There you go!"}