(self: MonitoredTask, x)
| 63 | |
| 64 | @app.task(bind=True, base=MonitoredTask) |
| 65 | def replaced_identity(self: MonitoredTask, x): |
| 66 | log_demo(self) |
| 67 | logger.warning("Stamping identity_task with MonitoringIdStampingVisitor() before replace") |
| 68 | replaced_task = identity_task.s(x) |
| 69 | # These stamps should be overridden by the stamps from MonitoredTask.on_replace() |
| 70 | replaced_task.stamp(MonitoringIdStampingVisitor()) |
| 71 | return self.replace(replaced_task) |
| 72 | |
| 73 | |
| 74 | @app.task(bind=True, base=StampOnReplace) |
nothing calls this directly
no test coverage detected