<commit_before>package watch

import (
	"encoding/json"
	"strings"
	"time"

	"github.com/golang/mock/gomock"
	. "github.com/onsi/ginkgo"
	"github.com/onsi/ginkgo/extensions/table"
	. "github.com/onsi/gomega"
	k8sv1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/fake"
	"k8s.io/client-go/testing"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/cache/testing"
	"k8s.io/client-go/tools/record"

	virtv1 "kubevirt.io/kubevirt/pkg/api/v1"
	"kubevirt.io/kubevirt/pkg/kubecli"
	"kubevirt.io/kubevirt/pkg/log"
	"kubevirt.io/kubevirt/pkg/testutils"
)

var _ = Describe("Node controller with", func() {
	log.Log.SetIOWriter(GinkgoWriter)

	var ctrl *gomock.Controller
	var vmInterface *kubecli.MockVMInterface
	var nodeSource *framework.FakeControllerSource
	var nodeInformer cache.SharedIndexInformer
	var vmSource *framework.FakeControllerSource
	var vmInformer cache.SharedIndexInformer
	var stop chan struct{}
	var controller *NodeController
	var recorder *record.FakeRecorder
	var mockQueue *testutils.MockWorkQueue
	var virtClient *kubecli.MockKubevirtClient
	var kubeClient *fake.Clientset
	var vmFeeder *testutils.VirtualMachineFeeder

	syncCaches := func(stop chan struct{}) {
		go nodeInformer.Run(stop)
		go vmInformer.Run(stop)
		Expect(cache.WaitForCacheSync(stop, nodeInformer.HasSynced, vmInformer.HasSynced)).To(BeTrue())
	}

	BeforeEach(func() {
		stop = make(chan struct{})
		ctrl = gomock.NewController(GinkgoT())
		virtClient = kubecli.NewMockKubevirtClient(ctrl)
		vmInterface = kubecli.NewMockVMInterface(ctrl)

		nodeInformer, nodeSource = testutils.NewFakeInformerFor(&k8sv1.Node{})
		vmInformer, vmSource = testutils.NewFakeInformerFor(&virtv1.VirtualMachine{})
		recorder = record.NewFakeRecorder(100)

		controller = NewNodeController(virtClient, nodeInformer, vmInformer, recorder)
		// Wrap our workqueue to have a way to detect when we are done processing updates
		mockQueue = testutils.NewMockWorkQueue(controller.Queue)
		controller.Queue = mockQueue
		vmFeeder = testutils.NewVirtualMachineFeeder(mockQueue, vmSource)

		// Set up mock client
		virtClient.EXPECT().VM(v1.NamespaceAll).Return(vmInterface).AnyTimes()
		virtClient.EXPECT().VM(v1.NamespaceDefault).Return(vmInterface).AnyTimes()
		kubeClient = fake.NewSimpleClientset()
		virtClient.EXPECT().CoreV1().Return(kubeClient.CoreV1()).AnyTimes()

		// Make sure that all unexpected calls to kubeClient will fail
		kubeClient.Fake.PrependReactor("*", "*", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
			Expect(action).To(BeNil())
			return true, nil, nil
		})
		syncCaches(stop)
	})

	addNode := func(node *k8sv1.Node) {
		mockQueue.ExpectAdds(1)
		nodeSource.Add(node)
		mockQueue.Wait()
	}

	modifyNode := func(node *k8sv1.Node) {
		mockQueue.ExpectAdds(1)
		nodeSource.Modify(node)
		mockQueue.Wait()
	}

	deleteNode := func(node *k8sv1.Node) {
		mockQueue.ExpectAdds(1)
		nodeSource.Delete(node)
		mockQueue.Wait()
	}

	Context("pods and vms given", func() {
		It("should only select stuck vms", func() {
			node := NewHealthyNode("test")
			finalVM := NewRunningVirtualMachine("finalVM", node)
			finalVM.Status.Phase = virtv1.Succeeded
			vmWithPod := NewRunningVirtualMachine("vmWithPod", node)
			podForVM := NewHealthyPodForVirtualMachine("podForVM", vmWithPod)
			vmWithPodInDifferentNamespace := NewRunningVirtualMachine("vmWithPodInDifferentNamespace", node)
			podInDifferentNamespace := NewHealthyPodForVirtualMachine("podInDifferentnamespace", vmWithPodInDifferentNamespace)
			podInDifferentNamespace.Namespace = "wrong"
			vmWithoutPod := NewRunningVirtualMachine("vmWithoutPod", node)

			vms := filterStuckVirtualMachinesWithoutPods([]*virtv1.VirtualMachine{
				finalVM,
				vmWithPod,
				vmWithPodInDifferentNamespace,
				vmWithoutPod,
			}, []*k8sv1.Pod{
				podForVM,
				podInDifferentNamespace,
			})

			By("filtering out vms in final state")
			Expect(vms).ToNot(ContainElement(finalVM))

			By("filtering out vms which have a pod")
			Expect(vms).ToNot(ContainElement(vmWithPod))

			By("keeping vms which are running und have no pod in their namespace")
			Expect(vms).To(ContainElement(vmWithoutPod))
			Expect(vms).To(ContainElement(vmWithPodInDifferentNamespace))
		})
	})

	Context("responsive virt-handler given", func() {
		It("should do nothing", func() {
			node := NewHealthyNode("testnode")

			addNode(node)

			controller.Execute()
		})
	})

	Context("unresponsive virt-handler given", func() {
		It("should set the node to unschedulable", func() {
			node := NewHealthyNode("testnode")
			node.Annotations[virtv1.VirtHandlerHeartbeat] = nowAsJSONWithOffset(-10 * time.Minute)

			addNode(node)

			kubeClient.Fake.PrependReactor("patch", "nodes", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, nil, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{}, nil)

			controller.Execute()
		})
		table.DescribeTable("should set a vm without a pod to failed state if the vm is in ", func(phase virtv1.VMPhase) {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm1", node)
			vm.Status.Phase = phase

			addNode(node)
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any())

			controller.Execute()
		},
			table.Entry("running state", virtv1.Running),
			table.Entry("scheduled state", virtv1.Scheduled),
		)
		It("should set a vm without a pod to failed state, triggered by vm add event", func() {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm1", node)

			vmFeeder.Add(vm)
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any())

			controller.Execute()
		})
		It("should set a vm without a pod to failed state, triggered by node update", func() {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm1", node)

			nodeInformer.GetStore().Add(node)
			modifyNode(node.DeepCopy())
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any())<commit_msg>Test if node controller can handle vm patch issues<commit_after>package watch

import (
	"encoding/json"
	"strings"
	"time"

	"github.com/golang/mock/gomock"
	. "github.com/onsi/ginkgo"
	"github.com/onsi/ginkgo/extensions/table"
	. "github.com/onsi/gomega"
	k8sv1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/fake"
	"k8s.io/client-go/testing"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/cache/testing"
	"k8s.io/client-go/tools/record"

	"fmt"

	virtv1 "kubevirt.io/kubevirt/pkg/api/v1"
	"kubevirt.io/kubevirt/pkg/kubecli"
	"kubevirt.io/kubevirt/pkg/log"
	"kubevirt.io/kubevirt/pkg/testutils"
)

var _ = Describe("Node controller with", func() {
	log.Log.SetIOWriter(GinkgoWriter)

	var ctrl *gomock.Controller
	var vmInterface *kubecli.MockVMInterface
	var nodeSource *framework.FakeControllerSource
	var nodeInformer cache.SharedIndexInformer
	var vmSource *framework.FakeControllerSource
	var vmInformer cache.SharedIndexInformer
	var stop chan struct{}
	var controller *NodeController
	var recorder *record.FakeRecorder
	var mockQueue *testutils.MockWorkQueue
	var virtClient *kubecli.MockKubevirtClient
	var kubeClient *fake.Clientset
	var vmFeeder *testutils.VirtualMachineFeeder

	syncCaches := func(stop chan struct{}) {
		go nodeInformer.Run(stop)
		go vmInformer.Run(stop)
		Expect(cache.WaitForCacheSync(stop, nodeInformer.HasSynced, vmInformer.HasSynced)).To(BeTrue())
	}

	BeforeEach(func() {
		stop = make(chan struct{})
		ctrl = gomock.NewController(GinkgoT())
		virtClient = kubecli.NewMockKubevirtClient(ctrl)
		vmInterface = kubecli.NewMockVMInterface(ctrl)

		nodeInformer, nodeSource = testutils.NewFakeInformerFor(&k8sv1.Node{})
		vmInformer, vmSource = testutils.NewFakeInformerFor(&virtv1.VirtualMachine{})
		recorder = record.NewFakeRecorder(100)

		controller = NewNodeController(virtClient, nodeInformer, vmInformer, recorder)
		// Wrap our workqueue to have a way to detect when we are done processing updates
		mockQueue = testutils.NewMockWorkQueue(controller.Queue)
		controller.Queue = mockQueue
		vmFeeder = testutils.NewVirtualMachineFeeder(mockQueue, vmSource)

		// Set up mock client
		virtClient.EXPECT().VM(v1.NamespaceAll).Return(vmInterface).AnyTimes()
		virtClient.EXPECT().VM(v1.NamespaceDefault).Return(vmInterface).AnyTimes()
		kubeClient = fake.NewSimpleClientset()
		virtClient.EXPECT().CoreV1().Return(kubeClient.CoreV1()).AnyTimes()

		// Make sure that all unexpected calls to kubeClient will fail
		kubeClient.Fake.PrependReactor("*", "*", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
			Expect(action).To(BeNil())
			return true, nil, nil
		})
		syncCaches(stop)
	})

	addNode := func(node *k8sv1.Node) {
		mockQueue.ExpectAdds(1)
		nodeSource.Add(node)
		mockQueue.Wait()
	}

	modifyNode := func(node *k8sv1.Node) {
		mockQueue.ExpectAdds(1)
		nodeSource.Modify(node)
		mockQueue.Wait()
	}

	deleteNode := func(node *k8sv1.Node) {
		mockQueue.ExpectAdds(1)
		nodeSource.Delete(node)
		mockQueue.Wait()
	}

	Context("pods and vms given", func() {
		It("should only select stuck vms", func() {
			node := NewHealthyNode("test")
			finalVM := NewRunningVirtualMachine("finalVM", node)
			finalVM.Status.Phase = virtv1.Succeeded
			vmWithPod := NewRunningVirtualMachine("vmWithPod", node)
			podForVM := NewHealthyPodForVirtualMachine("podForVM", vmWithPod)
			vmWithPodInDifferentNamespace := NewRunningVirtualMachine("vmWithPodInDifferentNamespace", node)
			podInDifferentNamespace := NewHealthyPodForVirtualMachine("podInDifferentnamespace", vmWithPodInDifferentNamespace)
			podInDifferentNamespace.Namespace = "wrong"
			vmWithoutPod := NewRunningVirtualMachine("vmWithoutPod", node)

			vms := filterStuckVirtualMachinesWithoutPods([]*virtv1.VirtualMachine{
				finalVM,
				vmWithPod,
				vmWithPodInDifferentNamespace,
				vmWithoutPod,
			}, []*k8sv1.Pod{
				podForVM,
				podInDifferentNamespace,
			})

			By("filtering out vms in final state")
			Expect(vms).ToNot(ContainElement(finalVM))

			By("filtering out vms which have a pod")
			Expect(vms).ToNot(ContainElement(vmWithPod))

			By("keeping vms which are running und have no pod in their namespace")
			Expect(vms).To(ContainElement(vmWithoutPod))
			Expect(vms).To(ContainElement(vmWithPodInDifferentNamespace))
		})
	})

	Context("responsive virt-handler given", func() {
		It("should do nothing", func() {
			node := NewHealthyNode("testnode")

			addNode(node)

			controller.Execute()
		})
	})

	Context("unresponsive virt-handler given", func() {
		It("should set the node to unschedulable", func() {
			node := NewHealthyNode("testnode")
			node.Annotations[virtv1.VirtHandlerHeartbeat] = nowAsJSONWithOffset(-10 * time.Minute)

			addNode(node)

			kubeClient.Fake.PrependReactor("patch", "nodes", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, nil, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{}, nil)

			controller.Execute()
		})
		table.DescribeTable("should set a vm without a pod to failed state if the vm is in ", func(phase virtv1.VMPhase) {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm1", node)
			vm.Status.Phase = phase

			addNode(node)
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any())

			controller.Execute()
		},
			table.Entry("running state", virtv1.Running),
			table.Entry("scheduled state", virtv1.Scheduled),
		)
		It("should set multiple vms to failed in one go, even if some updates fail", func() {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm", node)
			vm1 := NewRunningVirtualMachine("vm1", node)
			vm2 := NewRunningVirtualMachine("vm2", node)

			addNode(node)
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm, *vm1, *vm2}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any()).Times(1)
			vmInterface.EXPECT().Patch(vm1.Name, types.JSONPatchType, gomock.Any()).Return(nil, fmt.Errorf("some error")).Times(1)
			vmInterface.EXPECT().Patch(vm2.Name, types.JSONPatchType, gomock.Any()).Times(1)

			controller.Execute()
		})
		It("should set a vm without a pod to failed state, triggered by vm add event", func() {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm1", node)

			vmFeeder.Add(vm)
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any())

			controller.Execute()
		})
		It("should set a vm without a pod to failed state, triggered by node update", func() {
			node := NewUnhealthyNode("testnode")
			vm := NewRunningVirtualMachine("vm1", node)

			nodeInformer.GetStore().Add(node)
			modifyNode(node.DeepCopy())
			kubeClient.Fake.PrependReactor("list", "pods", func(action testing.Action) (handled bool, obj runtime.Object, err error) {
				return true, &k8sv1.PodList{}, nil
			})

			vmInterface.EXPECT().List(gomock.Any()).Return(&virtv1.VirtualMachineList{Items: []virtv1.VirtualMachine{*vm}}, nil)
			vmInterface.EXPECT().Patch(vm.Name, types.JSONPatchType, gomock.Any())
{'fraction_non_alphanumeric': 0.1342895263509006, 'fraction_numerical': 0.009006004002668445, 'mean_word_length': 5.116279069767442, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3297381', 'n_tokens_mistral': 5587, 'n_tokens_neox': 5076, 'n_words': 936}