(obj: PyObjectRef, vm: &VirtualMachine)
| 2702 | } |
| 2703 | |
| 2704 | fn is_coroutine(obj: PyObjectRef, vm: &VirtualMachine) -> PyResult<bool> { |
| 2705 | if obj.class().is(vm.ctx.types.coroutine_type) { |
| 2706 | return Ok(true); |
| 2707 | } |
| 2708 | |
| 2709 | let asyncio_coroutines = vm.import("asyncio.coroutines", 0)?; |
| 2710 | if let Some(iscoroutine) = |
| 2711 | vm.get_attribute_opt(asyncio_coroutines, vm.ctx.intern_str("iscoroutine"))? |
| 2712 | { |
| 2713 | let result = iscoroutine.call((obj,), vm)?; |
| 2714 | result.try_to_bool(vm) |
| 2715 | } else { |
| 2716 | Ok(false) |
| 2717 | } |
| 2718 | } |
| 2719 | |
| 2720 | fn new_invalid_state_error(vm: &VirtualMachine, msg: &str) -> PyBaseExceptionRef { |
| 2721 | match vm.import("asyncio.exceptions", 0) { |
no test coverage detected