sanity: Add tests for NodeStageVolume/NodeUnstageVolume

mock/service/node.go

@@ -16,15 +16,86 @@ func (s *service) NodeStageVolume(
 	req *csi.NodeStageVolumeRequest) (
 	*csi.NodeStageVolumeResponse, error) {
 
-	return nil, status.Error(codes.Unimplemented, "")
+	device, ok := req.PublishInfo["device"]
+	if !ok {
+		return nil, status.Error(
+			codes.InvalidArgument,
+			"stage volume info 'device' key required")
+	}
+
+	if len(req.GetVolumeId()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+	}
+
+	if len(req.GetStagingTargetPath()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Staging Target Path cannot be empty")
+	}
+
+	if req.GetVolumeCapability() == nil {
+		return nil, status.Error(codes.InvalidArgument, "Volume Capability cannot be empty")
+	}
+
+	s.volsRWL.Lock()
+	defer s.volsRWL.Unlock()
+
+	i, v := s.findVolNoLock("id", req.VolumeId)
+	if i < 0 {
+		return nil, status.Error(codes.NotFound, req.VolumeId)
+	}
+
+	// nodeStgPathKey is the key in the volume's attributes that is set to a
+	// mock stage path if the volume has been published by the node
+	nodeStgPathKey := path.Join(s.nodeID, req.StagingTargetPath)
+
+	// Check to see if the volume has already been staged.
+	if v.Attributes[nodeStgPathKey] != "" {
+		// TODO: Check for the capabilities to be equal. Return "ALREADY_EXISTS"
+		// if the capabilities don't match.
+		return &csi.NodeStageVolumeResponse{}, nil
+	}
+
+	// Stage the volume.
+	v.Attributes[nodeStgPathKey] = device
+	s.vols[i] = v
+
+	return &csi.NodeStageVolumeResponse{}, nil
 }
 
 func (s *service) NodeUnstageVolume(
 	ctx context.Context,
 	req *csi.NodeUnstageVolumeRequest) (
 	*csi.NodeUnstageVolumeResponse, error) {
 
-	return nil, status.Error(codes.Unimplemented, "")
+	if len(req.GetVolumeId()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+	}
+
+	if len(req.GetStagingTargetPath()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Staging Target Path cannot be empty")
+	}
+
+	s.volsRWL.Lock()
+	defer s.volsRWL.Unlock()
+
+	i, v := s.findVolNoLock("id", req.VolumeId)
+	if i < 0 {
+		return nil, status.Error(codes.NotFound, req.VolumeId)
+	}
+
+	// nodeStgPathKey is the key in the volume's attributes that is set to a
+	// mock stage path if the volume has been published by the node
+	nodeStgPathKey := path.Join(s.nodeID, req.StagingTargetPath)
+
+	// Check to see if the volume has already been unstaged.
+	if v.Attributes[nodeStgPathKey] == "" {
+		return &csi.NodeUnstageVolumeResponse{}, nil
+	}
+
+	// Unpublish the volume.
+	delete(v.Attributes, nodeStgPathKey)
+	s.vols[i] = v
+
+	return &csi.NodeUnstageVolumeResponse{}, nil
 }
 
 func (s *service) NodePublishVolume(
@@ -39,6 +110,18 @@ func (s *service) NodePublishVolume(
 			"publish volume info 'device' key required")
 	}
 
+	if len(req.GetVolumeId()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
+	}
+
+	if len(req.GetTargetPath()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Target Path cannot be empty")
+	}
+
+	if req.GetVolumeCapability() == nil {
+		return nil, status.Error(codes.InvalidArgument, "Volume Capability cannot be empty")
+	}
+
 	s.volsRWL.Lock()
 	defer s.volsRWL.Unlock()
 
@@ -64,7 +147,11 @@ func (s *service) NodePublishVolume(
 	}
 
 	// Publish the volume.
-	v.Attributes[nodeMntPathKey] = device
+	if req.GetStagingTargetPath() != "" {
+		v.Attributes[nodeMntPathKey] = req.GetStagingTargetPath()
+	} else {
+		v.Attributes[nodeMntPathKey] = device
+	}
 	s.vols[i] = v
 
 	return &csi.NodePublishVolumeResponse{}, nil
@@ -130,6 +217,13 @@ func (s *service) NodeGetCapabilities(
 					},
 				},
 			},
+			{
+				Type: &csi.NodeServiceCapability_Rpc{
+					Rpc: &csi.NodeServiceCapability_RPC{
+						Type: csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
+					},
+				},
+			},
 		},
 	}, nil
 }

pkg/sanity/node.go

@@ -364,3 +364,157 @@ func testFullWorkflowSuccess(s csi.ControllerClient, c csi.NodeClient, controlle
 		})
 	Expect(err).NotTo(HaveOccurred())
 }
+
+var _ = Describe("NodeStageVolume [Node Server]", func() {
+	var (
+		s                          csi.ControllerClient
+		c                          csi.NodeClient
+		controllerPublishSupported bool
+		nodeStageSupported         bool
+		device                     string
+	)
+
+	BeforeEach(func() {
+		s = csi.NewControllerClient(conn)
+		c = csi.NewNodeClient(conn)
+		device = "/dev/mock"
+		controllerPublishSupported = isControllerCapabilitySupported(
+			s,
+			csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME)
+		nodeStageSupported = isNodeCapabilitySupported(c, csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME)
+		if nodeStageSupported {
+			err := createMountTargetLocation(config.StagingPath)
+			Expect(err).NotTo(HaveOccurred())
+		} else {
+			Skip("NodeStageVolume not supported")
+		}
+	})
+
+	It("should fail when no volume id is provided", func() {
+
+		_, err := c.NodeStageVolume(
+			context.Background(),
+			&csi.NodeStageVolumeRequest{
+				StagingTargetPath: config.StagingPath,
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+				PublishInfo: map[string]string{
+					"device": device,
+				},
+			})
+		Expect(err).To(HaveOccurred())
+
+		serverError, ok := status.FromError(err)
+		Expect(ok).To(BeTrue())
+		Expect(serverError.Code()).To(Equal(codes.InvalidArgument))
+	})
+
+	It("should fail when no staging target path is provided", func() {
+
+		_, err := c.NodeStageVolume(
+			context.Background(),
+			&csi.NodeStageVolumeRequest{
+				VolumeId: "id",
+				VolumeCapability: &csi.VolumeCapability{
+					AccessType: &csi.VolumeCapability_Mount{
+						Mount: &csi.VolumeCapability_MountVolume{},
+					},
+					AccessMode: &csi.VolumeCapability_AccessMode{
+						Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+					},
+				},
+				PublishInfo: map[string]string{
+					"device": device,
+				},
+			})
+		Expect(err).To(HaveOccurred())
+
+		serverError, ok := status.FromError(err)
+		Expect(ok).To(BeTrue())
+		Expect(serverError.Code()).To(Equal(codes.InvalidArgument))
+	})
+
+	It("should fail when no volume capability is provided", func() {
+
+		_, err := c.NodeStageVolume(
+			context.Background(),
+			&csi.NodeStageVolumeRequest{
+				VolumeId:          "id",
+				StagingTargetPath: config.StagingPath,
+				PublishInfo: map[string]string{
+					"device": device,
+				},
+			})
+		Expect(err).To(HaveOccurred())
+
+		serverError, ok := status.FromError(err)
+		Expect(ok).To(BeTrue())
+		Expect(serverError.Code()).To(Equal(codes.InvalidArgument))
+	})
+
+	It("should return appropriate values (no optional values added)", func() {
+		testFullWorkflowSuccess(s, c, controllerPublishSupported, nodeStageSupported)
+	})
+})
+
+var _ = Describe("NodeUnstageVolume [Node Server]", func() {
+	var (
+		s                          csi.ControllerClient
+		c                          csi.NodeClient
+		controllerPublishSupported bool
+		nodeStageSupported         bool
+	)
+
+	BeforeEach(func() {
+		s = csi.NewControllerClient(conn)
+		c = csi.NewNodeClient(conn)
+		controllerPublishSupported = isControllerCapabilitySupported(
+			s,
+			csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME)
+		nodeStageSupported = isNodeCapabilitySupported(c, csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME)
+		if nodeStageSupported {
+			err := createMountTargetLocation(config.StagingPath)
+			Expect(err).NotTo(HaveOccurred())
+		} else {
+			Skip("NodeUnstageVolume not supported")
+		}
+	})
+
+	It("should fail when no volume id is provided", func() {
+
+		_, err := c.NodeUnstageVolume(
+			context.Background(),
+			&csi.NodeUnstageVolumeRequest{
+				StagingTargetPath: config.StagingPath,
+			})
+		Expect(err).To(HaveOccurred())
+
+		serverError, ok := status.FromError(err)
+		Expect(ok).To(BeTrue())
+		Expect(serverError.Code()).To(Equal(codes.InvalidArgument))
+	})
+
+	It("should fail when no staging target path is provided", func() {
+
+		_, err := c.NodeUnstageVolume(
+			context.Background(),
+			&csi.NodeUnstageVolumeRequest{
+				VolumeId: "id",
+			})
+		Expect(err).To(HaveOccurred())
+
+		serverError, ok := status.FromError(err)
+		Expect(ok).To(BeTrue())
+		Expect(serverError.Code()).To(Equal(codes.InvalidArgument))
+	})
+
+	It("should return appropriate values (no optional values added)", func() {
+		testFullWorkflowSuccess(s, c, controllerPublishSupported, nodeStageSupported)
+	})
+})