(self)
| 101 | assert prepare_accept_content(['yaml']) == b4.accept |
| 102 | |
| 103 | def test_get_result_meta(self): |
| 104 | b1 = BaseBackend(self.app) |
| 105 | meta = b1._get_result_meta(result={'fizz': 'buzz'}, |
| 106 | state=states.SUCCESS, traceback=None, |
| 107 | request=None) |
| 108 | assert meta['status'] == states.SUCCESS |
| 109 | assert meta['result'] == {'fizz': 'buzz'} |
| 110 | assert meta['traceback'] is None |
| 111 | |
| 112 | self.app.conf.result_extended = True |
| 113 | args = ['a', 'b'] |
| 114 | kwargs = {'foo': 'bar'} |
| 115 | task_name = 'mytask' |
| 116 | |
| 117 | b2 = BaseBackend(self.app) |
| 118 | request = Context(args=args, kwargs=kwargs, |
| 119 | task=task_name, |
| 120 | delivery_info={'routing_key': 'celery'}) |
| 121 | meta = b2._get_result_meta(result={'fizz': 'buzz'}, |
| 122 | state=states.SUCCESS, traceback=None, |
| 123 | request=request, encode=False) |
| 124 | assert meta['name'] == task_name |
| 125 | assert meta['args'] == args |
| 126 | assert meta['kwargs'] == kwargs |
| 127 | assert meta['queue'] == 'celery' |
| 128 | |
| 129 | def test_get_result_meta_stamps_attribute_error(self): |
| 130 | class Request: |
nothing calls this directly
no test coverage detected