test_http.py

import json
import os
import socket
import tempfile
import time
from unittest import TestCase, mock
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import urlopen

import pytest
import requests

from mocket import Mocket, Mocketizer, mocketize
from mocket.mocks.mockhttp import Entry, Response


class HttpTestCase(TestCase):
    def assertEqualHeaders(self, first, second, msg=None):
        first = {k.lower(): v for k, v in first.items()}
        second = {k.lower(): v for k, v in second.items()}
        self.assertDictEqual(first, second, msg)


@pytest.mark.skipif('os.getenv("SKIP_TRUE_HTTP", False)')
class TrueHttpEntryTestCase(HttpTestCase):
    @mocketize
    def test_truesendall(self):
        url = "http://httpbin.local/ip"
        resp = urlopen(url)
        self.assertEqual(resp.code, 200)
        resp = requests.get(url)
        self.assertEqual(resp.status_code, 200)

    def test_truesendall_with_recording(self):
        with tempfile.TemporaryDirectory() as temp_dir, Mocketizer(
            truesocket_recording_dir=temp_dir
        ):
            url = "http://httpbin.local/ip"

            urlopen(url)
            requests.get(url)
            resp = urlopen(url)
            self.assertEqual(resp.code, 200)
            resp = requests.get(url)
            self.assertEqual(resp.status_code, 200)
            assert "origin" in resp.json()

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

        self.assertEqual(len(responses["httpbin.local"]["80"].keys()), 2)

    def test_truesendall_with_gzip_recording(self):
        with tempfile.TemporaryDirectory() as temp_dir, Mocketizer(
            truesocket_recording_dir=temp_dir
        ):
            url = "http://httpbin.local/gzip"

            requests.get(url)
            resp = requests.get(url)
            self.assertEqual(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.local"]["80"].keys()) == 1

    def test_truesendall_with_chunk_recording(self):
        with tempfile.TemporaryDirectory() as temp_dir, Mocketizer(
            truesocket_recording_dir=temp_dir
        ):
            url = "http://httpbin.local/range/70000?chunk_size=65536"

            requests.get(url)
            resp = requests.get(url)
            self.assertEqual(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.local"]["80"].keys()) == 1

    @mocketize
    def test_wrongpath_truesendall(self):
        Entry.register(
            Entry.GET, "http://httpbin.local/user.agent", Response(status=404)
        )
        response = urlopen("http://httpbin.local/ip")
        self.assertEqual(response.code, 200)


class HttpEntryTestCase(HttpTestCase):
    def test_register(self):
        with mock.patch("time.gmtime") as tt:
            tt.return_value = time.struct_time((2013, 4, 30, 10, 39, 21, 1, 120, 0))
            Entry.register(
                Entry.GET,
                "http://testme.org/get?a=1&b=2#test",
                Response('{"a": "€"}', headers={"content-type": "application/json"}),
            )
        entries = Mocket._entries[("testme.org", 80)]
        self.assertEqual(len(entries), 1)
        entry = entries[0]
        self.assertEqual(entry.method, "GET")
        self.assertEqual(entry.schema, "http")
        self.assertEqual(entry.path, "/get")
        self.assertEqual(entry.query, "a=1&b=2")
        self.assertEqual(len(entry.responses), 1)
        response = entry.responses[0]
        self.assertEqual(response.body, b'{"a": "\xe2\x82\xac"}')
        self.assertEqual(response.status, 200)
        self.assertEqual(
            response.headers,
            {
                "Status": "200",
                "Date": "Tue, 30 Apr 2013 10:39:21 GMT",
                "Connection": "close",
                "Server": "Python/Mocket",
                "Content-Length": "12",
                "Content-Type": "application/json",
            },
        )

    @mocketize
    def test_sendall(self):
        with mock.patch("time.gmtime") as tt:
            tt.return_value = time.struct_time((2013, 4, 30, 10, 39, 21, 1, 120, 0))
            Entry.single_register(
                Entry.GET, "http://testme.org/get/p/?a=1&b=2", body="test_body"
            )
        resp = urlopen("http://testme.org/get/p/?b=2&a=1", timeout=10)
        self.assertEqual(resp.code, 200)
        self.assertEqual(resp.read(), b"test_body")
        self.assertEqualHeaders(
            dict(resp.headers),
            {
                "Status": "200",
                "Content-length": "9",
                "Server": "Python/Mocket",
                "Connection": "close",
                "Date": "Tue, 30 Apr 2013 10:39:21 GMT",
                "Content-type": "text/plain; charset=utf-8",
            },
        )
        self.assertEqual(len(Mocket.request_list()), 1)

    @mocketize
    def test_sendall_json(self):
        with mock.patch("time.gmtime") as tt:
            tt.return_value = time.struct_time((2013, 4, 30, 10, 39, 21, 1, 120, 0))
            Entry.single_register(
                Entry.GET,
                "http://testme.org/get?a=1&b=2#test",
                body='{"a": "€"}',
                headers={"content-type": "application/json"},
            )

        response = urlopen("http://testme.org/get?b=2&a=1#test", timeout=10)
        self.assertEqual(response.code, 200)
        self.assertEqual(response.read(), b'{"a": "\xe2\x82\xac"}')
        self.assertEqualHeaders(
            dict(response.headers),
            {
                "status": "200",
                "content-length": "12",
                "server": "Python/Mocket",
                "connection": "close",
                "date": "Tue, 30 Apr 2013 10:39:21 GMT",
                "content-type": "application/json",
            },
        )
        self.assertEqual(len(Mocket.request_list()), 1)

    @mocketize
    def test_sendall_double(self):
        Entry.register(
            Entry.GET, "http://testme.org/", Response(status=404), Response()
        )
        self.assertRaises(HTTPError, urlopen, "http://testme.org/")
        response = urlopen("http://testme.org/")
        self.assertEqual(response.code, 200)
        response = urlopen("http://testme.org/")
        self.assertEqual(response.code, 200)
        self.assertEqual(len(Mocket.request_list()), 3)

    @mocketize
    def test_mockhttp_entry_collect_duplicates(self):
        Entry.single_register(
            Entry.POST, "http://testme.org/", status=200, match_querystring=False
        )
        requests.post(
            "http://testme.org/?foo=bar",
            data="{'foo': 'bar'}",
            headers={"content-type": "application/json"},
        )
        requests.post("http://testme.org/")
        self.assertEqual(len(Mocket.request_list()), 2)
        self.assertEqual(Mocket.last_request().path, "/")

    @mocketize
    def test_multipart(self):
        url = "http://httpbin.local/post"
        data = '--xXXxXXyYYzzz\r\nContent-Disposition: form-data; name="content"\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: 68\r\n\r\nAction: comment\nText: Comment with attach\nAttachment: x1.txt, x2.txt\r\n--xXXxXXyYYzzz\r\nContent-Disposition: form-data; name="attachment_2"; filename="x.txt"\r\nContent-Type: text/plain\r\nContent-Length: 4\r\n\r\nbye\n\r\n--xXXxXXyYYzzz\r\nContent-Disposition: form-data; name="attachment_1"; filename="x.txt"\r\nContent-Type: text/plain\r\nContent-Length: 4\r\n\r\nbye\n\r\n--xXXxXXyYYzzz--\r\n'
        headers = {
            "Content-Length": "495",
            "Content-Type": "multipart/form-data; boundary=xXXxXXyYYzzz",
            "Accept": "text/plain",
            "User-Agent": "Mocket",
            "Accept-encoding": "identity",
        }
        Entry.register(Entry.POST, url)
        response = requests.post(url, data=data, headers=headers)
        self.assertEqual(response.status_code, 200)
        last_request = Mocket.last_request()
        self.assertEqual(last_request.method, "POST")
        self.assertEqual(last_request.path, "/post")
        self.assertEqual(last_request.body, data)
        sent_headers = dict(last_request.headers)
        self.assertEqualHeaders(
            sent_headers,
            {
                "accept": "text/plain",
                "accept-encoding": "identity",
                "content-length": "495",
                "content-type": "multipart/form-data; boundary=xXXxXXyYYzzz",
                "host": "httpbin.local",
                "user-agent": "Mocket",
                "connection": "keep-alive",
            },
        )
        self.assertEqual(len(Mocket.request_list()), 1)

    @mocketize
    def test_file_object(self):
        url = "http://github.com/fluidicon.png"
        filename = "tests/fluidicon.png"
        with open(filename, "rb") as file_obj:
            Entry.single_register(Entry.GET, url, body=file_obj)
        r = requests.get(url)
        remote_content = r.content
        with open(filename, "rb") as local_file_obj:
            local_content = local_file_obj.read()
        self.assertEqual(remote_content, local_content)
        self.assertEqual(len(remote_content), len(local_content))
        self.assertEqual(int(r.headers["Content-Length"]), len(local_content))
        self.assertEqual(r.headers["Content-Type"], "image/png")

    @mocketize
    def test_same_url_different_methods(self):
        url = "http://bit.ly/fakeurl"
        response_to_mock = {"content": 0, "method": None}
        responses = []
        methods = [Entry.PUT, Entry.GET, Entry.POST]

        for m in methods:
            response_to_mock["method"] = m
            Entry.single_register(m, url, body=json.dumps(response_to_mock))
            response_to_mock["content"] += 1
        for m in methods:
            responses.append(requests.request(m, url).json())

        methods_from_responses = [r["method"] for r in responses]
        contents_from_responses = [r["content"] for r in responses]
        self.assertEqual(methods, methods_from_responses)
        self.assertEqual(list(range(len(methods))), contents_from_responses)

    @mocketize
    def test_request_bodies(self):
        url = "http://bit.ly/fakeurl/{0}"

        for e in range(5):
            u = url.format(e)
            Entry.single_register(Entry.POST, u, body=str(e))
            request_body = urlencode({f"key-{e}": f"value={e}"})
            urlopen(u, request_body.encode("utf-8"))
            last_request = Mocket.last_request()
            assert last_request.body == request_body

    @mocketize(truesocket_recording_dir=os.path.dirname(__file__))
    def test_truesendall_with_dump_from_recording(self):
        requests.get(
            "http://httpbin.local/ip",
            headers={
                "user-agent": "Fake-User-Agent",
                "Accept-Encoding": "gzip, deflate, zstd",
            },
        )
        requests.get(
            "http://httpbin.local/gzip",
            headers={
                "user-agent": "Fake-User-Agent",
                "Accept-Encoding": "gzip, deflate, zstd",
            },
        )

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

        self.assertEqual(len(responses["httpbin.local"]["80"].keys()), 2)

    @mocketize
    def test_post_file_object(self):
        url = "http://github.com/fluidicon.png"
        Entry.single_register(Entry.POST, url, status=201)
        with open("tests/fluidicon.png", "rb") as file_obj:
            files = {"content": file_obj}
            r = requests.post(url, files=files, data={})
        self.assertEqual(r.status_code, 201)

    @mocketize
    def test_raise_exception_from_register(self):
        url = "http://github.com/fluidicon.png"
        Entry.register(Entry.GET, url, OSError())
        with self.assertRaises(requests.exceptions.ConnectionError):
            requests.get(url)

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

    @mocketize
    def test_sockets(self):
        """
        https://github.com/mindflayer/python-mocket/issues/111
        https://gist.github.com/amotl/015ef6b336db55128798d7f1a9a67dea
        """
        # Define HTTP conversation.
        url = "http://127.0.0.1:8080/api/data"
        Entry.single_register(Entry.POST, url)

        # Define HTTP url segments and data.
        host = "127.0.0.1"
        port = 8080
        method = "POST"
        path = "/api/data"
        data = json.dumps({"hello": "world"})

        # Invoke HTTP request.
        address = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)[0]
        sock = socket.socket(address[0], address[1], address[2])

        sock.connect(address[-1])
        sock.send(f"{method} {path} HTTP/1.0\r\n".encode())
        sock.send(f"Host: {host}\r\n".encode())
        sock.send(b"Content-Type: application/json\r\n")
        sock.send(b"Content-Length: %d\r\n" % len(data))
        sock.send(b"Connection: close\r\n\r\n")
        sock.send(data.encode())
        sock.close()

        # Proof that worked.
        self.assertEqual(Mocket.last_request().body, '{"hello": "world"}')

    @mocketize
    def test_fail_because_entry_not_served(self):
        url = "http://github.com/fluidicon.png"
        Entry.single_register(Entry.GET, url)
        Entry.single_register(Entry.GET, "http://github.com/fluidicon.jpg")
        requests.get(url)
        with self.assertRaises(AssertionError):
            Mocket.assert_fail_if_entries_not_served()

    @mocketize
    def test_does_not_fail_because_all_entries_are_served(self):
        url = "http://github.com/fluidicon.png"
        second_url = "http://github.com/fluidicon.jpg"
        Entry.single_register(Entry.GET, url)
        Entry.single_register(Entry.GET, second_url)
        requests.get(url)
        requests.get(second_url)
        Mocket.assert_fail_if_entries_not_served()

    @mocketize
    def test_multi_register(self):
        url = "http://foobar.com/path"
        Entry.register(
            Entry.POST,
            url,
            Response(body='{"foo":"bar0"}', status=200),
            Response(body='{"foo":"bar1"}', status=201),
            Response(body='{"foo":"bar2"}', status=202),
        )

        response = requests.post(url, json={"test": 0})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json(), {"foo": "bar0"})

        response = requests.post(url, json={"test": 1})
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.json(), {"foo": "bar1"})

        response = requests.post(url, json={"test": 2})
        self.assertEqual(response.status_code, 202)
        self.assertEqual(response.json(), {"foo": "bar2"})

        response = requests.post(url, json={"test": 22})
        self.assertEqual(response.status_code, 202)
        self.assertEqual(response.json(), {"foo": "bar2"})

    def test_suggestion_for_register_and_body(self):
        url = "http://foobar.com/path"
        with self.assertRaises(AttributeError):
            Entry.register(
                Entry.POST,
                url,
                body='{"foo":"bar0"}',
            )

    def test_suggestion_for_register_and_status(self):
        url = "http://foobar.com/path"
        with self.assertRaises(AttributeError):
            Entry.register(
                Entry.POST,
                url,
                status=201,
            )

    def test_invalid_config_key(self):
        url = "http://foobar.com/path"
        with self.assertRaises(KeyError):
            Entry.register(
                Entry.POST,
                url,
                Response(body='{"foo":"bar0"}', status=200),
                invalid_key=True,
            )

    def test_add_trailing_slash(self):
        url = "http://testme.org"
        entry = Entry(url, "GET", [Response(body='{"foo":"bar0"}', status=200)])
        self.assertEqual(entry.path, "/")

    @mocketize
    def test_mocket_with_no_path(self):
        Entry.register(Entry.GET, "http://httpbin.local", Response(status=202))
        response = urlopen("http://httpbin.local/")
        self.assertEqual(response.code, 202)
        self.assertEqual(Mocket._entries[("httpbin.local", 80)][0].path, "/")