test: cover invalid borrowed receive view lifecycle operations

Author: yuweih205

tests/ep/test_recv_view.py

@@ -77,6 +77,13 @@ def check(payload, recv_ids, recv_weights, handle):
     def dispatch(borrow, value=x):
         result = buffer.dispatch(value, borrow_recv=borrow, **kwargs)
         if asynchronous:
+            if borrow:
+                # A failed early release must leave the lease usable after waiting.
+                pending_view = result[0]
+                rejected(pending_view.wait, RuntimeError)
+                rejected(lambda: pending_view.slab, RuntimeError)
+                rejected(lambda: pending_view.row_indices, RuntimeError)
+                rejected(pending_view.release, RuntimeError)
             result[-1].current_stream_wait()
         return result[:4]
 
@@ -89,6 +96,9 @@ def dispatch(borrow, value=x):
     rejected(buffer.destroy, RuntimeError)
     view.release()
     rejected(lambda: view.slab, RuntimeError)
+    rejected(lambda: view.row_indices, RuntimeError)
+    rejected(view.wait, RuntimeError)
+    rejected(view.release, RuntimeError)
 
     replay, _, _, _, event = buffer.dispatch(x, handle=handle, num_sms=8, async_with_compute_stream=True)
     event.current_stream_wait()
@@ -124,6 +134,8 @@ def dispatch(borrow, value=x):
     for extra in [dict(do_expand=True), dict(defer_epilogue=True), dict(do_cpu_sync=False), dict(handle=handle)]:
         rejected(lambda extra=extra: buffer.dispatch(x, borrow_recv=True, **(kwargs | extra)), ValueError)
     rejected(lambda: buffer.dispatch(x.float(), borrow_recv=True, **kwargs), ValueError)
+    fp8 = (x.to(torch.float8_e4m3fn), torch.ones((tokens, hidden // 128), device=x.device))
+    rejected(lambda: buffer.dispatch(fp8, borrow_recv=True, **kwargs), ValueError)
     dist.barrier()
     buffer.destroy()