(
secrethash: SecretHash, transfer_task: TransferTask
)
| 141 | |
| 142 | |
| 143 | def get_transfer_from_task( |
| 144 | secrethash: SecretHash, transfer_task: TransferTask |
| 145 | ) -> Optional[LockedTransferType]: |
| 146 | if isinstance(transfer_task, InitiatorTask): |
| 147 | # Work around for https://github.com/raiden-network/raiden/issues/5480, |
| 148 | # can be removed when |
| 149 | # https://github.com/raiden-network/raiden/issues/5515 is done. |
| 150 | if secrethash not in transfer_task.manager_state.initiator_transfers: |
| 151 | return None |
| 152 | |
| 153 | return transfer_task.manager_state.initiator_transfers[secrethash].transfer |
| 154 | elif isinstance(transfer_task, MediatorTask): |
| 155 | pairs = transfer_task.mediator_state.transfers_pair |
| 156 | if pairs: |
| 157 | return pairs[-1].payer_transfer |
| 158 | |
| 159 | assert transfer_task.mediator_state.waiting_transfer, "Invalid mediator_state" |
| 160 | return transfer_task.mediator_state.waiting_transfer.transfer |
| 161 | elif isinstance(transfer_task, TargetTask): |
| 162 | return transfer_task.target_state.transfer |
| 163 | |
| 164 | raise ValueError("get_transfer_from_task for a non TransferTask argument") |
| 165 | |
| 166 | |
| 167 | def transfer_tasks_view( |
no outgoing calls
no test coverage detected