# Flask實作_消息隊列_Testing with Celery
###### tags: `flask` `celery` `celery 4.4`
## History
20191119_依據4.4版本說明翻譯
## 目錄
[TOC]
## 參考
[官方連結_Testing with Celery](http://docs.celeryproject.org/en/master/userguide/testing.html)
## Testing with Celery
### Tasks and unit tests
要在單元測試中測試任務行為，首選的方法是模擬。
#### Eager mode:
根據定義，由[`task_always_eager`](http://docs.celeryproject.org/en/master/userguide/configuration.html#std:setting-task_always_eager)設置啟用的eager mode並不適用於單元測試。
當使用eager model測試的時候，你只是測試Worker中發生的事情的模擬，而且模擬與實際發生的事情之間有在許多差異。
Celery任務非常類似web view，因為它只定義如何在被稱為任務的上下文中執行操作。
這意味著最佳任務僅處理序列化、訊息標頭、重試等內容，而實際邏輯則在其它地方實現。
假設我們有一個任務像這個：
```python
from .models import Product
@app.task(bind=True)
def send_order(self, product_pk, quantity, price):
price = Decimal(price) # json serializes this to string.
# models are passed by id, not serialized.
product = Product.objects.get(product_pk)
try:
product.order(quantity, price)
except OperationalError as exc:
raise self.retry(exc=exc)
```
注意：任務的[綁定](http://docs.celeryproject.org/en/latest/userguide/tasks.html#bound-tasks)意味著任務的第一個參數將總是任務實例(`self`)。這意味著你
確實將`self`做為第一個參數，並且可以使用`Task`類的方法與屬性。
你可以使用模擬來寫為這個任務寫單元測試，就像下面範例：
```python
from pytest import raises
from celery.exceptions import Retry
# for python 2: use mock.patch from `pip install mock`.
from unittest.mock import patch
from proj.models import Product
from proj.tasks import send_order
class test_send_order:
@patch('proj.tasks.Product.order') # < patching Product in module above
def test_success(self, product_order):
product = Product.objects.create(
name='Foo',
)
send_order(product.pk, 3, Decimal(30.3))
product_order.assert_called_with(3, Decimal(30.3))
@patch('proj.tasks.Product.order')
@patch('proj.tasks.send_order.retry')
def test_failure(self, send_order_retry, product_order):
product = Product.objects.create(
name='Foo',
)
# Set a side effect on the patched methods
# so that they raise the errors we want.
send_order_retry.side_effect = Retry()
product_order.side_effect = OperationalError()
with raises(Retry):
send_order(product.pk, 3, Decimal(30.6))
```
### Py.test
*New in version 4.0.*
Celery也是[pytest](https://pypi.org/project/pytest/)插件，它添加了可以在集成（或單元）測試套件中使用的設置。
#### Marks
##### celery - Set test app configuration.
celery的標記讓你可以覆寫用於單個測試案例的配置：
```python
@pytest.mark.celery(result_backend='redis://')
def test_something():
...
```
或是在一個類別中的所有測試案例：
```python
@pytest.mark.celery(result_backend='redis://')
class test_something:
def test_one(self):
...
def test_two(self):
...
```
#### Fixtures
##### Function scope
celery_app **- Celery app used for testing**.
這個設置會回傳可用於測試的Celery應用程式。
範例：
```python
def test_create_task(celery_app, celery_worker):
@celery_app.task
def mul(x, y):
return x * y
assert mul.delay(4, 4).get(timeout=10) == 16
```
celery_worker **- Embed live worker**.
這個設置啟動一個Celery worker實例，你可以用它來集成測試。這個worker會在一個隔離的[線程](http://terms.naer.edu.tw/detail/2430706/)中被啟動，並在測試回傳之後立即關閉。
範例：
```python
@pytest.fixture(scope='session')
def celery_config():
return {
'broker_url': 'amqp://',
'result_backend': 'redis://'
}
def test_add(celery_worker):
mytask.delay()
# If you wish to override some setting in one test cases
# only - you can use the ``celery`` mark:
@pytest.mark.celery(result_backend='rpc')
def test_other(celery_worker):
...
```
##### Session scope
celery_config **- Override to setup Celery test app configuration**.
你可以重新定義這個設置來配置測試Celery應用程式。
然後，透過你的設置回傳的配置會用來配置**celery_app()**，與**celery_session_app()**。
範例：
```python
def celery_config():
return {
'broker_url': 'amqp://',
'result_backend': 'rpc',
}
```
celery_parameters **- Override to setup Celery test app parameters**.
你可以重新定義這個設置來改變測試Celery應用程式的__init__參數。與`celery_config()`相比，這些設置會在實例化Celery的時候直接傳遞。
然後，透過你的設置回傳的配置會用來配置**celery_app()**，與**celery_session_app()**。
範例：
```python
@pytest.fixture(scope='session')
def celery_parameters():
return {
'task_cls': my.package.MyCustomTaskClass,
'strict_typing': False,
}
```
celery_worker_parameters **- Override to setup Celery worker parameters**.
你可以重新定義這個設置來改變測試Celery worker的__init__參數。這些設置會在初始化的時候直接傳遞給[WorkController](http://docs.celeryproject.org/en/master/reference/celery.worker.html#celery.worker.WorkController)
然後，透過你的設置回傳的配置會用來配置**celery_worker()**，與**celery_session_worker()**。
範例：
```python
@pytest.fixture(scope='session')
def celery_worker_parameters():
return {
'queues': ('high-prio', 'low-prio'),
'exclude_queues': ('celery'),
}
```
celery_enable_logging **- Override to enable logging in embedded workers**.
你可以在嵌入式worker啟動的時候覆寫設置來包括模組。
你可以讓它回傳要導入的模組名稱清單，清單可以是任務模組，註冊信號的模組，等等。
範例：
```python
@pytest.fixture(scope='session')
def celery_includes():
return [
'proj.tests.tasks',
'proj.tests.celery_signal_handlers',
]
```
celery_worker_pool **- Override the pool used for embedded workers**.
你可以覆寫設置來配置用於嵌入式workers的執行池。
範例：
```python
@pytest.fixture(scope='session')
def celery_worker_pool():
return 'prefork'
```
:::warning
**Warning:**
你不能使用gevent/eventlet pools，除非你整個測試套件都在啟用monkeypatches的情況下執行。
:::
celery_session_worker **- Embedded worker that lives throughout the session**.
這個設置啟動一個在整個測試過程中都存活的worker(不會在每次測試中被啟動/停止)。
```python
# Add this to your conftest.py
@pytest.fixture(scope='session')
def celery_config():
return {
'broker_url': 'amqp://',
'result_backend': 'rpc',
}
# Do this in your tests.
def test_add_task(celery_session_worker):
assert add.delay(2, 2) == 4
```
:::warning
**Warning:**
將session與臨時的worker混在一起不是一個好主意…
:::
celery_session_app **- Celery app used for testing (session scope)**.
當其它會話範圍的設置需要引用Celery應用程式實例的時候，可以使用此功能。
use_celery_app_trap **- Raise exception on falling back to default app**.
你可以在你的`conftest.py`覆寫這個設置，以啟用"app trap"：如果有東西試著訪問預設或當前應用程式(`current_app`)，則會拋出異常。
範例：
```python
@pytest.fixture(scope='session')
def use_celery_app_trap():
return True
```
如果有一個測試想訪問預設的應用程式，你應該使用depends_on_current_app設置來標記它：
```python
@pytest.mark.usefixtures('depends_on_current_app')
def test_something():
something()
```
Sign in with Wallet
Connect another wallet