FilesExpand file tree

master

/

test_threaded.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

52 lines (37 loc) · 1.32 KB

master

/

test_threaded.py

Copy path

Top

File metadata and controls

• Code
• Blame

52 lines (37 loc) · 1.32 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

import http.client

import threading

import time

from collections.abc import Iterable

import pytest

from werkzeug import Request

from werkzeug import Response

from pytest_httpserver import HTTPServer

@pytest.fixture

def threaded() -> Iterable[HTTPServer]:

server = HTTPServer(threaded=True)

server.start()

yield server

server.clear()

if server.is_running():

server.stop()

def test_threaded(threaded: HTTPServer):

sleep_time = 0.5

def handler(_request: Request):

# allow some time to the client to have multiple pending request

# handlers running in parallel

time.sleep(sleep_time)

# send back thread id

return Response(f"{threading.get_ident()}")

threaded.expect_request("/foo").respond_with_handler(handler)

number_of_connections = 5

conns = [http.client.HTTPConnection(threaded.host, threaded.port) for _ in range(number_of_connections)]

for conn in conns:

conn.request("GET", "/foo", headers={"Host": threaded.host})

thread_ids: list[int] = []

for conn in conns:

response = conn.getresponse()

assert response.status == 200

thread_ids.append(int(response.read()))

for conn in conns:

conn.close()

assert len(thread_ids) == len(set(thread_ids)), "thread ids returned should be unique"