FilesExpand file tree

flake

/

test_blocking_httpserver.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

122 lines (79 loc) · 3.4 KB

flake

/

test_blocking_httpserver.py

Copy path

Top

File metadata and controls

• Code
• Blame

122 lines (79 loc) · 3.4 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

115

116

117

118

119

120

121

122

from contextlib import contextmanager

from copy import deepcopy

from multiprocessing.pool import ThreadPool

from urllib.parse import urlparse

import pytest

import requests

from pytest_httpserver import BlockingHTTPServer

@contextmanager

def when_a_request_is_being_sent_to_the_server(request):

with ThreadPool(1) as pool:

yield pool.apply_async(requests.request, kwds=request)

def then_the_server_gets_the_request(server, request):

request = deepcopy(request)

replace_url_with_uri(request)

return server.assert_request(**request)

def replace_url_with_uri(request):

request["uri"] = get_uri(request["url"])

del request["url"]

def get_uri(url):

url = urlparse(url)

return "?".join(item for item in [url.path, url.query] if item)

def when_the_server_responds_to(client_connection, response):

client_connection.respond_with_json(response)

def then_the_response_is_got_from(server_connection, response):

assert server_connection.get(timeout=9).json() == response

@pytest.fixture

def httpserver():

server = BlockingHTTPServer(timeout=1)

server.start()

yield server

server.clear()

if server.is_running():

server.stop()

def test_behave_workflow(httpserver: BlockingHTTPServer):

request = dict(

method="GET",

url=httpserver.url_for("/my/path"),

)

with when_a_request_is_being_sent_to_the_server(request) as server_connection:

client_connection = then_the_server_gets_the_request(httpserver, request)

response = {"foo": "bar"}

when_the_server_responds_to(client_connection, response)

then_the_response_is_got_from(server_connection, response)

def test_raises_assertion_error_when_request_does_not_match(httpserver: BlockingHTTPServer):

request = dict(

method="GET",

url=httpserver.url_for("/my/path"),

)

with when_a_request_is_being_sent_to_the_server(request):

with pytest.raises(AssertionError) as exc:

httpserver.assert_request(uri="/not/my/path/")

assert "/not/my/path/" in str(exc)

assert "does not match" in str(exc)

def test_raises_assertion_error_when_request_was_not_sent(httpserver: BlockingHTTPServer):

with pytest.raises(AssertionError) as exc:

httpserver.assert_request(uri="/my/path/", timeout=1)

assert "/my/path/" in str(exc)

assert "timed out" in str(exc)

def test_ignores_when_request_is_not_asserted(httpserver: BlockingHTTPServer):

request = dict(

method="GET",

url=httpserver.url_for("/my/path"),

)

with when_a_request_is_being_sent_to_the_server(request) as server_connection:

assert server_connection.get(timeout=9).text == "No handler found for this request"

def test_raises_assertion_error_when_request_was_not_responded(httpserver: BlockingHTTPServer):

request = dict(

method="GET",

url=httpserver.url_for("/my/path"),

)

with when_a_request_is_being_sent_to_the_server(request):

then_the_server_gets_the_request(httpserver, request)

httpserver.stop() # waiting for timeout of waiting for the response

with pytest.raises(AssertionError) as exc:

httpserver.check_assertions()

assert "/my/path" in str(exc)

assert "no response" in str(exc).lower()

def test_repr(httpserver: BlockingHTTPServer):

assert repr(httpserver) == f"<BlockingHTTPServer host=localhost port={httpserver.port}>"