uAjb@s:dZddlZddlmZddlmZmZmZmZm Z ddl Z erddl m Z ddl mZn eZ ZejjdZd d Zed d dd d Ze jddddZe jddddZe jddddZe jddddZe jddddZe jddddZe jddddZe jddddZe jddd d!Ze jd"d#Ze jddd$d%Z e jd&d'Z!e jd(d)Z"dS)*z;Fixtures and testing utilities for :pypi:`pytest `.N)contextmanager) TYPE_CHECKINGAnyMappingSequenceUnion)Celery)WorkController NO_WORKERcCs|jdddS)z)Register additional pytest configuration.markerszBcelery(**overrides): override celery configuration for a test caseN)addinivalue_line)configr_/home/www/goberdata.com/tradingbinance/backend/.venv/Lib/site-packages/celery/contrib/pytest.pypytest_configuresrFc ksiddlm}m}|s"in|}|ddd|d||}||d| |VWdQRXdS) z=Utility context used to setup Celery app for pytest fixtures.)TestAppsetup_default_appset_as_currentFenable_loggingruse_trapN) testing.apprr)rr parametersrrrtest_apprrr _create_app s rscopesessioncCsdS)zYou can override this fixture to enable the app trap. The app trap raises an exception whenever something attempts to use the current or default apps. Frrrrruse_celery_app_trap5src cs{|jjd}t||r'|jni}td|d|d||&}|sk|j|j|VWdQRXdS)z1Session Fixture: Return app for session fixtures.celeryrrrN)nodeget_closest_markerdictkwargsr set_default set_current)request celery_configcelery_parameterscelery_enable_loggingrmarkrapprrrcelery_session_app@s    r,c csddlm}ts|x|D]}|jj|qWx|D]}|j|q>W|j|d|| } | VWdQRXdS)z?Session Fixture: Start worker that lives throughout test suite.r)workerpoolN)testingr-r loaderimport_task_module register_task start_worker) r&r,celery_includescelery_class_taskscelery_worker_poolcelery_worker_parametersr-module class_taskwrrrcelery_session_workerTs     r;cCsdS)z0You can override this fixture to enable logging.Frrrrrr)lsr)cCsfS)zYou can override this include modules when a worker start. You can have this return a list of module names to import, these can be task modules, modules registering signals, and so on. rrrrrr4ssr4cCsdS)zYou can override this fixture to set the worker pool. The "solo" pool is used by default, but you can set this to return e.g. "prefork". solorrrrrr6~sr6cCsiS)zRedefine this fixture to configure the test Celery app. The config returned by your fixture will then be used to configure the :func:`celery_app` fixture. rrrrrr'sr'cCsiS)zRedefine this fixture to change the init parameters of test Celery app. The dict returned by your fixture will then be used as parameters when instantiating :class:`~celery.Celery`. rrrrrr(sr(cCsiS)a"Redefine this fixture to change the init parameters of Celery workers. This can be used e. g. to define queues the worker will consume tasks from. The dict returned by your fixture will then be used as parameters when instantiating :class:`~celery.worker.WorkController`. rrrrrr7s r7c csa|jjd}t||r'|jni}td|d|d|| }|VWdQRXdS)z/Fixture creating a Celery application instance.rrrrN)r r!r"r#r)r&r'r(r)rr*rr+rrr celery_apps  r=cCsgS)zARedefine this fixture to register tasks with the test Celery app.rrrrrr5sr5c csbddlm}ts^x|D]}|jj|qW|j|d|| }|VWdQRXdS)zAFixture: Start worker in a thread, stop it when the test returns.r)r-r.N)r/r-r r0r1r3)r&r=r4r6r7r-r8r:rrr celery_workers   r>cCs|jdS)z!Fixture that sets app as current.N)r%)r=rrrdepends_on_current_appsr?)#__doc__os contextlibrtypingrrrrrpytestrrr-r objectenvirongetr rrfixturerr,r;r)r4r6r'r(r7r=r5r>r?rrrrs6 (