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
|
# -*- coding: utf-8 -*-
# thumbor imaging service
# https://github.com/thumbor/thumbor/wiki
# Licensed under the MIT license:
# http://www.opensource.org/licenses/mit-license
# Copyright (c) 2011 globo.com thumbor@googlegroups.com
# pylint: disable=protected-access
from time import sleep
from unittest import mock
import pytest
from preggy import expect
from thumbor.threadpool import ThreadPool
@pytest.mark.asyncio
async def test_can_get_threadpool_instance():
instance = ThreadPool.instance(0)
expect(instance.pool).to_be_null()
instance = ThreadPool.instance(10)
expect(instance).not_to_be_null()
instance2 = ThreadPool.instance(10)
expect(instance2).to_equal(instance)
instance3 = ThreadPool.instance(11)
expect(instance3).not_to_equal(instance)
@pytest.mark.asyncio
async def test_can_run_task_in_foreground():
instance = ThreadPool.instance(0)
expect(instance).not_to_be_null()
def add():
return 10
result = await instance._execute_in_foreground(add)
expect(result).to_equal(10)
@pytest.mark.asyncio
async def test_can_run_task_in_foreground_and_exception_happens():
instance = ThreadPool.instance(0)
expect(instance).not_to_be_null()
exception = Exception("Boom")
def add():
raise exception
with expect.error_to_happen(Exception, message="Boom"):
await instance._execute_in_foreground(add)
@pytest.mark.asyncio
async def test_queueing_task_when_no_pool_runs_sync():
instance = ThreadPool.instance(0)
expect(instance).not_to_be_null()
def add():
return 10
result = await instance.queue(add)
expect(result).to_equal(10)
@pytest.mark.asyncio
async def test_queueing_task_when_no_pool_runs_sync_and_exception_happens():
instance = ThreadPool.instance(0)
expect(instance).not_to_be_null()
exception = Exception("Boom")
def add():
raise exception
with expect.error_to_happen(Exception, message="Boom"):
await instance.queue(add)
@pytest.mark.asyncio
async def test_can_run_async():
instance = ThreadPool.instance(10)
expect(instance).not_to_be_null()
def add():
sleep(1)
return 10
result = await instance._execute_in_pool(add)
expect(result).to_equal(10)
@pytest.mark.asyncio
async def test_can_run_async_with_queue():
instance = ThreadPool.instance(10)
expect(instance).not_to_be_null()
def add(value):
sleep(1)
return value
result = await instance.queue(add, 10)
expect(result).to_equal(10)
@pytest.mark.asyncio
async def test_can_cleanup_pool():
instance = ThreadPool.instance(0)
instance.pool = mock.Mock()
instance.cleanup()
expect(instance.pool.shutdown.called).to_be_true()
|
