---
# System prepended metadata

title: Flask實作_消息隊列_Testing with Celery
tags: [flask, celery, celery 4.4]

---

# 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()
```



