(self)
| 1064 | state.should_terminate = original_should_terminate |
| 1065 | |
| 1066 | def test_fast_trace_task(self): |
| 1067 | assert self.app.use_fast_trace_task is False |
| 1068 | setup_worker_optimizations(self.app) |
| 1069 | assert self.app.use_fast_trace_task is True |
| 1070 | tid = uuid() |
| 1071 | message = self.TaskMessage(self.mytask.name, tid, args=[4]) |
| 1072 | assert len(message.payload) == 3 |
| 1073 | try: |
| 1074 | self.mytask.__trace__ = build_tracer( |
| 1075 | self.mytask.name, self.mytask, self.app.loader, 'test', |
| 1076 | app=self.app, |
| 1077 | ) |
| 1078 | failed, res, runtime = fast_trace_task( |
| 1079 | self.mytask.name, tid, message.headers, message.body, |
| 1080 | message.content_type, message.content_encoding) |
| 1081 | assert not failed |
| 1082 | assert res == repr(4 ** 4) |
| 1083 | assert runtime is not None |
| 1084 | assert isinstance(runtime, numbers.Real) |
| 1085 | finally: |
| 1086 | reset_worker_optimizations(self.app) |
| 1087 | assert self.app.use_fast_trace_task is False |
| 1088 | delattr(self.mytask, '__trace__') |
| 1089 | failed, res, runtime = trace_task_ret( |
| 1090 | self.mytask.name, tid, message.headers, message.body, |
| 1091 | message.content_type, message.content_encoding, app=self.app, |
| 1092 | ) |
| 1093 | assert not failed |
| 1094 | assert res == repr(4 ** 4) |
| 1095 | assert runtime is not None |
| 1096 | assert isinstance(runtime, numbers.Real) |
| 1097 | |
| 1098 | def test_trace_task_ret(self): |
| 1099 | self.mytask.__trace__ = build_tracer( |
nothing calls this directly
no test coverage detected