我正在遵循芹菜测试的文档，以测试我的芹菜任务。
我还将PyTest-bdd用于BDD测试。
我的项目结构如下
myprj/
├── __init__.py
├── app
│ └── __init__.py
├── requirements.txt
├── tests
│ ├── __init__.py
│ ├── features
│ │ └── automation
│ │ └── core
│ │ └── celerytasks.feature
│ └── functional
│ ├── __init__.py
│ └── automation
│ ├── __init__.py
│ ├── conftest.py
│ └── core
│ ├── __init__.py
│ └── test_celerytasks.py
├── tox-requirements.txt
└── tox.ini
myprj/insuert.txt
amqp==2.4.2
billiard==3.6.0.0
celery==4.3.0
kombu==4.5.0
vine==1.3.0
myprj/tox-requirements.txt
coverage>=4.1
flake8
pylint
pytest-cov==2.6.1
pytest==4.4.0
pytest-bdd
pytest-flask
pytest-mock
pytest-xdist
myprj/tox.ini
[tox]
envlist = functional,lint
skipsdist = True
sitepackages = False
skip_missing_interpreters = True
[testenv]
basepython=python3.5
deps=
-rtox-requirements.txt
-rrequirements.txt
install_command = pip install -i https://pypi.mydomain.com --extra-index-url https://pypi.mydomain.com --trusted-host pypi.mydomain.com {opts} {packages} --exists-action w
setenv = FLASK_CONFIG=testing
[testenv:functional]
commands=
py.test -n auto -l --cov=app --cov-report term-missing --no-cov-on-fail --max-slave-restart=0 tests/functional/
[testenv:lint]
commands=pylint app
myprj/tests/features/automation/core/celerytasks.feature
@automation @celerytasks
Feature: Get core device details
Get the core details for device
Scenario: Valid device number with argument string
Given valid device number as deviceNumbers
When enter the deviceNumbers "123456" with attributes "name"
Then it should return "mydevicename"
myprj/tests/functional/automation/conftest.py
import pytest
@pytest.fixture(scope='session')
def celery_config():
return {
'broker_url': 'amqp://',
'result_backend': 'redis://'
}
@pytest.fixture(scope='session')
def celery_enable_logging():
return True
myprj/tests/functional/automation/core/test_celerytasks.py
import pytest
from pytest_bdd import when
from pytest_bdd import then
from pytest_bdd import given
from pytest_bdd import parsers
from pytest_bdd import scenario
@scenario("../../../features/automation/core/celerytasks.feature",
"Valid device number with argument string")
def test_celerytasks(celery_worker):
pass
@given("valid device number as deviceNumbers")
def context():
return {}
@when(parsers.parse('enter the deviceNumbers "{deviceNumbers:d}"'
' with attributes "{attributes}"'))
def set_data(deviceNumbers, attributes, context):
context['deviceNumbers'] = deviceNumbers
context['attributes'] = attributes
@pytest.mark.celery
@then(parsers.parse('it should return "{result}"'))
def get_results(result, context, celery_worker):
print(context)
当我使用托克斯运行该操作时，它会为身份验证提供错误。
$ tox -e functional
functional create: /Users/myuser/myprj/.tox/functional
functional installdeps: -rtox-requirements.txt, -rrequirements.txt
functional installed: amqp==2.4.2,apipkg==1.4,astroid==1.6.2,atomicwrites==1.3.0,attrs==19.1.0,billiard==3.6.0.0,celery==4.3.0,click==6.7,coverage==4.5.2,execnet==1.5.0,flake8==3.5.0,Flask==1.0.2,glob2==0.6,isort==4.3.4,itsdangerous==0.24,Jinja2==2.10,kombu==4.5.0,lazy-object-proxy==1.3.1,Mako==1.0.9,MarkupSafe==1.0,mccabe==0.6.1,more-itertools==7.0.0,parse==1.12.0,parse-type==0.4.2,pathlib2==2.3.3,pluggy==0.11.0,py==1.8.0,pycodestyle==2.3.1,pyflakes==1.6.0,pylint==1.8.3,pytest==4.4.0,pytest-bdd==3.1.0,pytest-cov==2.6.1,pytest-flask==0.10.0,pytest-forked==0.2,pytest-mock==1.7.1,pytest-xdist==1.22.2,pytz==2019.1,six==1.12.0,vine==1.3.0,Werkzeug==0.14.1,wrapt==1.10.11
functional run-test-pre: PYTHONHASHSEED='4168571602'
functional run-test: commands[0] | py.test -n auto -l --cov=app --cov-report term-missing --no-cov-on-fail --max-slave-restart=0 tests/functional/
=============================================================================================== test session starts ================================================================================================
platform darwin -- Python 3.5.2, pytest-4.4.0, py-1.8.0, pluggy-0.11.0
cachedir: .tox/functional/.pytest_cache
rootdir: /Users/myuser/myprj
plugins: xdist-1.22.2, mock-1.7.1, forked-0.2, flask-0.10.0, cov-2.6.1, bdd-3.1.0, celery-4.3.0
gw0 [1] / gw1 [1] / gw2 [1] / gw3 [1] / gw4 [1] / gw5 [1] / gw6 [1] / gw7 [1]
scheduling tests via LoadScheduling
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
Coverage.py warning: No data was collected. (no-data-collected)
E [100%]Coverage.py warning: No data was collected. (no-data-collected)
====================================================================================================== ERRORS ======================================================================================================
________________________________________________________________________________________ ERROR at setup of test_celerytasks ________________________________________________________________________________________
[gw5] darwin -- Python 3.5.2 /Users/myuser/myprj/.tox/functional/bin/python3.5
request = <SubRequest 'celery_worker' for <Function test_celerytasks>>, celery_app = <Celery celery.tests at 0x10c6a03c8>, celery_includes = (), celery_worker_pool = 'solo', celery_worker_parameters = {}
@pytest.fixture()
def celery_worker(request,
celery_app,
celery_includes,
celery_worker_pool,
celery_worker_parameters):
# type: (Any, Celery, Sequence[str], str) -> WorkController
"""Fixture: Start worker in a thread, stop it when the test returns."""
if not NO_WORKER:
for module in celery_includes:
celery_app.loader.import_task_module(module)
with worker.start_worker(celery_app,
pool=celery_worker_pool,
> **celery_worker_parameters) as w:
celery_app = <Celery celery.tests at 0x10c6a03c8>
celery_includes = ()
celery_worker_parameters = {}
celery_worker_pool = 'solo'
request = <SubRequest 'celery_worker' for <Function test_celerytasks>>
.tox/functional/lib/python3.5/site-packages/celery/contrib/pytest.py:176:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../.pyenv/versions/3.5.2/lib/python3.5/contextlib.py:59: in __enter__
return next(self.gen)
.tox/functional/lib/python3.5/site-packages/celery/contrib/testing/worker.py:78: in start_worker
**kwargs) as worker:
../.pyenv/versions/3.5.2/lib/python3.5/contextlib.py:59: in __enter__
return next(self.gen)
.tox/functional/lib/python3.5/site-packages/celery/contrib/testing/worker.py:106: in _start_worker_thread
conn.default_channel.queue_declare
.tox/functional/lib/python3.5/site-packages/kombu/connection.py:852: in default_channel
self.ensure_connection(**conn_opts)
.tox/functional/lib/python3.5/site-packages/kombu/connection.py:422: in ensure_connection
callback, timeout=timeout)
.tox/functional/lib/python3.5/site-packages/kombu/utils/functional.py:341: in retry_over_time
return fun(*args, **kwargs)
.tox/functional/lib/python3.5/site-packages/kombu/connection.py:275: in connect
return self.connection
.tox/functional/lib/python3.5/site-packages/kombu/connection.py:823: in connection
self._connection = self._establish_connection()
.tox/functional/lib/python3.5/site-packages/kombu/connection.py:778: in _establish_connection
conn = self.transport.establish_connection()
.tox/functional/lib/python3.5/site-packages/kombu/transport/pyamqp.py:130: in establish_connection
conn.connect()
.tox/functional/lib/python3.5/site-packages/amqp/connection.py:313: in connect
self.drain_events(timeout=self.connect_timeout)
.tox/functional/lib/python3.5/site-packages/amqp/connection.py:500: in drain_events
while not self.blocking_read(timeout):
.tox/functional/lib/python3.5/site-packages/amqp/connection.py:506: in blocking_read
return self.on_inbound_frame(frame)
.tox/functional/lib/python3.5/site-packages/amqp/method_framing.py:55: in on_frame
callback(channel, method_sig, buf, None)
.tox/functional/lib/python3.5/site-packages/amqp/connection.py:510: in on_inbound_method
method_sig, payload, content,
.tox/functional/lib/python3.5/site-packages/amqp/abstract_channel.py:126: in dispatch_method
listener(*args)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <kombu.transport.pyamqp.Connection object at 0x10c75ada0>, reply_code = 403, reply_text = 'ACCESS_REFUSED - Login was refused using authentication mechanism AMQPLAIN. For details see the broker logfile.'
class_id = 0, method_id = 0
def _on_close(self, reply_code, reply_text, class_id, method_id):
"""Request a connection close.
This method indicates that the sender wants to close the
connection. This may be due to internal conditions (e.g. a
forced shut-down) or due to an error handling a specific
method, i.e. an exception. When a close is due to an
exception, the sender provides the class and method id of the
method which caused the exception.
RULE:
After sending this method any received method except the
Close-OK method MUST be discarded.
RULE:
The peer sending this method MAY use a counter or timeout
to detect failure of the other peer to respond correctly
with the Close-OK method.
RULE:
When a server receives the Close method from a client it
MUST delete all server-side resources associated with the
client's context. A client CANNOT reconnect to a context
after sending or receiving a Close method.
PARAMETERS:
reply_code: short
The reply code. The AMQ reply codes are defined in AMQ
RFC 011.
reply_text: shortstr
The localised reply text. This text can be logged as an
aid to resolving issues.
class_id: short
failing method class
When the close is provoked by a method exception, this
is the class of the method.
method_id: short
failing method ID
When the close is provoked by a method exception, this
is the ID of the method.
"""
self._x_close_ok()
raise error_for_code(reply_code, reply_text,
> (class_id, method_id), ConnectionError)
E amqp.exceptions.AccessRefused: (0, 0): (403) ACCESS_REFUSED - Login was refused using authentication mechanism AMQPLAIN. For details see the broker logfile.
class_id = 0
method_id = 0
reply_code = 403
reply_text = 'ACCESS_REFUSED - Login was refused using authentication mechanism AMQPLAIN. For details see the broker logfile.'
self = <kombu.transport.pyamqp.Connection object at 0x10c75ada0>
.tox/functional/lib/python3.5/site-packages/amqp/connection.py:639: AccessRefused
============================================================================================= 1 error in 3.26 seconds ==============================================================================================
ERROR: InvocationError for command /Users/myuser/myprj/.tox/functional/bin/py.test -n auto -l --cov=app --cov-report term-missing --no-cov-on-fail --max-slave-restart=0 tests/functional/ (exited with code 1)
_____________________________________________________________________________________________________ summary ______________________________________________________________________________________________________
ERROR: functional: commands failed
当我们模拟芹菜的配置时，我不知道为什么它给出身份验证错误？
看起来像在您的conftest下。这对我有用。
@pytest.fixture(autouse=True, scope='session')
def celery_config(django_db_setup, django_db_blocker,request):
with django_db_blocker.unblock():
return {
'broker_url': 'redis://localhost:6379',
'result_backend': 'redis://localhost:6379'
}