tests: usbredir: use virtctl/usbredir Client

This ensures that the same code used in virtctl to run and manage the
data from usbredirect binary are also used in the tests.

This also adds a test to expect failure when limit of
v1.UsbClientPassthroughMaxNumberOf devices is reached.

Signed-off-by: Victor Toso <victortoso@redhat.com>

tests/BUILD.bazel

@@ -168,6 +168,7 @@ go_test(
         "//pkg/virt-operator/resource/generate/components:go_default_library",
         "//pkg/virtctl/pause:go_default_library",
         "//pkg/virtctl/softreboot:go_default_library",
+        "//pkg/virtctl/usbredir:go_default_library",
         "//pkg/virtctl/vm:go_default_library",
         "//staging/src/kubevirt.io/api/clone/v1alpha1:go_default_library",
         "//staging/src/kubevirt.io/api/core:go_default_library",

tests/usbredir_test.go

@@ -20,9 +20,11 @@
 package tests_test
 
 import (
-	"io"
+	"net"
 	"time"
 
+	"kubevirt.io/kubevirt/pkg/virtctl/usbredir"
+
 	"kubevirt.io/kubevirt/tests/decorators"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -57,7 +59,6 @@ var helloMessageRemote = []byte{
 
 var _ = Describe("[crit:medium][vendor:cnv-qe@redhat.com][level:component][sig-compute] USB Redirection", decorators.SigCompute, func() {
 
-	var err error
 	var virtClient kubecli.KubevirtClient
 	const enoughMemForSafeBiosEmulation = "32Mi"
 	BeforeEach(func() {
@@ -82,100 +83,139 @@ var _ = Describe("[crit:medium][vendor:cnv-qe@redhat.com][level:component][sig-c
 	Describe("[crit:medium][vendor:cnv-qe@redhat.com][level:component] A VirtualMachineInstance with usbredir support", func() {
 
 		var vmi *v1.VirtualMachineInstance
+
 		BeforeEach(func() {
+			// A VMI for each test to have fresh stack on server side
 			vmi = libvmi.New(libvmi.WithResourceMemory(enoughMemForSafeBiosEmulation), withClientPassthrough())
 			vmi = tests.RunVMIAndExpectLaunch(vmi, 90)
 		})
 
-		Context("with an usbredir connection", func() {
-
-			usbredirConnect := func(connStop chan struct{}) {
-				pipeInReader, pipeInWriter := io.Pipe()
-				pipeOutReader, pipeOutWriter := io.Pipe()
-				defer pipeInReader.Close()
-				defer pipeOutReader.Close()
-
-				k8ResChan := make(chan error)
-				readStop := make(chan []byte)
-
-				By("Stablishing communication with usbredir socket from VMI")
-				go func() {
-					defer GinkgoRecover()
-					usbredirVMI, err := virtClient.VirtualMachineInstance(vmi.ObjectMeta.Namespace).USBRedir(vmi.ObjectMeta.Name)
-					if err != nil {
-						k8ResChan <- err
-						return
-					}
-
-					k8ResChan <- usbredirVMI.Stream(kubecli.StreamOptions{
-						In:  pipeInReader,
-						Out: pipeOutWriter,
-					})
-				}()
-
-				By("Exchanging hello message between client and QEMU's usbredir")
-				go func() {
-					defer GinkgoRecover()
-					buf := make([]byte, 1024, 1024)
-
-					// write hello message to remote (VMI)
-					nw, err := pipeInWriter.Write(helloMessageLocal)
-					Expect(err).ToNot(HaveOccurred())
-					Expect(nw).To(Equal(len(helloMessageLocal)))
-
-					// reading hello message from remote (VMI)
-					nr, err := pipeOutReader.Read(buf)
-					if err != nil && err != io.EOF {
-						Expect(err).ToNot(HaveOccurred())
-						return
-					}
-					if nr == 0 && err == io.EOF {
-						readStop <- []byte("")
-						return
-					}
-					readStop <- buf[0:nr]
-				}()
+		It("Should fail when limit is reached", func() {
+			stops := make([]chan struct{}, v1.UsbClientPassthroughMaxNumberOf+1)
+			errors := make([]chan error, v1.UsbClientPassthroughMaxNumberOf+1)
+			for i := 0; i <= v1.UsbClientPassthroughMaxNumberOf; i++ {
+				stops[i] = make(chan struct{})
+				errors[i] = make(chan error)
+				go runConnectGoroutine(virtClient, vmi, stops[i], errors[i])
+				// avoid too fast requests which might get denied by server (to avoid flakyness)
+				time.Sleep(100 * time.Millisecond)
+			}
 
+			for i := 0; i <= v1.UsbClientPassthroughMaxNumberOf; i++ {
 				select {
-				case response := <-readStop:
-					By("Checking the response from QEMU's usbredir")
-					// Comparing the actual messages could be error prone due the fact that:
-					// 1. Part of the return value is a qemu release version, e.g: 5.2.0 (test would break with different release!)
-					// 2. Capabilities can change over time which means the message would be different then the one hardcoded, correct nonetheless.
-					// I'm keeping the helloMessageRemote to have a proof of working example that could also be used if needed.
-					Expect(response).ToNot(BeEmpty(), "response should not be empty")
-					Expect(response).To(HaveLen(len(helloMessageRemote)))
-				case err = <-k8ResChan:
-					Expect(err).ToNot(HaveOccurred())
-				case <-time.After(45 * time.Second):
-					Fail("Timout reached while waiting for valid response")
-				case <-connStop:
-					return
+				case err := <-errors[i]:
+					Expect(err).To(MatchError(ContainSubstring("websocket: bad handshake")))
+					Expect(i).To(Equal(v1.UsbClientPassthroughMaxNumberOf))
+				case <-time.After(time.Second):
+					stops[i] <- struct{}{}
+					Expect(i).ToNot(Equal(v1.UsbClientPassthroughMaxNumberOf))
 				}
 			}
+		})
+
+		It("Should work in parallel", func() {
+			stops := make([]chan struct{}, v1.UsbClientPassthroughMaxNumberOf)
+			errors := make([]chan error, v1.UsbClientPassthroughMaxNumberOf)
+			for i := 0; i < v1.UsbClientPassthroughMaxNumberOf; i++ {
+				stops[i] = make(chan struct{})
+				errors[i] = make(chan error)
+				go runConnectGoroutine(virtClient, vmi, stops[i], errors[i])
+				// avoid too fast requests which might get denied by server (to avoid flakyness)
+				time.Sleep(100 * time.Millisecond)
+			}
 
-			It("Should work several times", func() {
-				for i := 0; i < 10; i++ {
-					usbredirConnect(nil)
+			for i := 0; i < v1.UsbClientPassthroughMaxNumberOf; i++ {
+				select {
+				case err := <-errors[i]:
+					Expect(err).ToNot(HaveOccurred())
+				case <-time.After(time.Second):
+					stops[i] <- struct{}{}
 				}
-			})
-
-			It("Should work in parallel", func() {
-				connStop := make(chan struct{})
-				defer close(connStop)
-				for i := 0; i < v1.UsbClientPassthroughMaxNumberOf; i++ {
-					go func() {
-						defer GinkgoRecover()
-						usbredirConnect(connStop)
-					}()
+			}
+		})
+
+		It("Should work several times", func() {
+			for i := 0; i < 4*v1.UsbClientPassthroughMaxNumberOf; i++ {
+				stop := make(chan struct{})
+				errch := make(chan error)
+				go runConnectGoroutine(virtClient, vmi, stop, errch)
+
+				select {
+				case err := <-errch:
+					Expect(err).ToNot(HaveOccurred())
+				case <-time.After(time.Second):
+					stop <- struct{}{}
 				}
-				// Wait for connection, write and read of all above.
-				time.Sleep(5 * time.Second)
-			})
+			}
 		})
 	})
 })
 
+func runConnectGoroutine(
+	virtClient kubecli.KubevirtClient,
+	vmi *v1.VirtualMachineInstance,
+	stop chan struct{},
+	errch chan error,
+) {
+	defer GinkgoRecover()
+	usbredirStream, err := virtClient.VirtualMachineInstance(vmi.ObjectMeta.Namespace).USBRedir(vmi.ObjectMeta.Name)
+	if err != nil {
+		errch <- err
+		return
+	}
+	usbredirConnect(usbredirStream, stop)
+}
+
+func usbredirConnect(
+	usbredirStream kubecli.StreamInterface,
+	stop chan struct{},
+) {
+
+	usbredirClient := usbredir.NewUSBRedirClient().
+		WithRemoteVMIStream(usbredirStream).
+		WithLocalTCPClient("localhost:0")
+
+	usbredirClient.ClientConnect = func(device, address string, stop chan struct{}) error {
+		defer GinkgoRecover()
+
+		conn, err := net.Dial("tcp", address)
+		Expect(err).ToNot(HaveOccurred())
+		defer conn.Close()
+
+		buf := make([]byte, 1024, 1024)
+
+		// write hello message to remote (VMI)
+		nw, err := conn.Write([]byte(helloMessageLocal))
+		Expect(err).ToNot(HaveOccurred())
+		Expect(nw).To(Equal(len(helloMessageLocal)))
+
+		// reading hello message from remote (VMI)
+		nr, err := conn.Read(buf)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(buf[0:nr]).ToNot(BeEmpty(), "response should not be empty")
+		Expect(buf[0:nr]).To(HaveLen(len(helloMessageRemote)))
+		<-stop
+		return err
+	}
+
+	usbredirClient.ConnectRemote()
+	usbredirClient.ConnectLocal("dead:beef")
+	run := make(chan error)
+	go func() {
+		run <- usbredirClient.Run()
+	}()
+	var err error
+	select {
+	case err = <-run:
+	case <-stop:
+		// stop requested, make the call
+		usbredirClient.Stop()
+		// now wait till Run() finishes
+		err = <-run
+	}
+	Expect(err).ToNot(HaveOccurred())
+}
+
 func withClientPassthrough() libvmi.Option {
 	return func(vmi *v1.VirtualMachineInstance) {
 		vmi.Spec.Domain.Devices.ClientPassthrough = &v1.ClientPassthroughDevices{}