def METHOD_NAME(test_microvm_with_api, bin_vsock_path, test_fc_session_root_path):
    """
    Vsock negative test to validate SIGPIPE/EPIPE handling.
    """
    vm = test_microvm_with_api
    vm.spawn()
    vm.basic_config()
    vm.add_net_iface()
    vm.api.vsock.put(vsock_id="vsock0", guest_cid=3, uds_path=f"/{VSOCK_UDS_PATH}")
    vm.start()
    # Generate the random data blob file, 20MB
    blob_path, blob_hash = make_blob(test_fc_session_root_path, 20 * 2**20)
    vm_blob_path = "/tmp/vsock/test.blob"
    # Set up a tmpfs drive on the guest, so we can copy the blob there.
    # Guest-initiated connections (echo workers) will use this blob.
    _copy_vsock_data_to_guest(vm.ssh, blob_path, vm_blob_path, bin_vsock_path)
    path = os.path.join(vm.jailer.chroot_path(), VSOCK_UDS_PATH)
    # Negative test for host-initiated connections that
    # are closed with in flight data.
    negative_test_host_connections(vm, path, blob_path, blob_hash)
    metrics = vm.flush_metrics()
    # Validate that at least 1 `SIGPIPE` signal was received.
    # Since we are reusing the existing echo server which triggers
    # reads/writes on the UDS backend connections, these might be closed
    # before a read() or a write() is about to be performed by the emulation.
    # The test uses 100 connections it is enough to close at least one
    # before write().
    #
    # If this ever fails due to 100 closes before read() we must
    # add extra tooling that will trigger only writes().
    assert metrics["signals"]["sigpipe"] > 0