See the License for the specific language governing permissions and
limitations under the License.
*/

package csi

import (
	"context"
	"errors"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"testing"

	csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
	"github.com/stretchr/testify/assert"

	api "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"

	"k8s.io/kubernetes/pkg/volume"
	"k8s.io/kubernetes/pkg/volume/csi"
	"k8s.io/kubernetes/pkg/volume/util"
	"k8s.io/kubernetes/test/utils/ktesting"
)

const (
	intreePDUniqueNamePrefix = "kubernetes.io/gce-pd/"
	csiPDUniqueNamePrefix    = "kubernetes.io/csi/pd.csi.storage.gke.io^projects/UNSPECIFIED/zones/UNSPECIFIED/disks/"
)

![ReDoc](https://fastapi.tiangolo.com/img/index/index-02-redoc-simple.png)

## Àpẹẹrẹ ìgbésókè mìíràn

Ní báyìí ṣe àtúnṣe fáìlì `main.py` láti gba kókó èsì láti inú ìbéèrè `PUT`.

Ní báyìí, ṣe ìkéde kókó èsì API nínú kóòdù rẹ nipa lílo àwọn ìtọ́kasí àmì irúfẹ́ Python, ọpẹ́ pàtàkìsi sí Pydantic.

```Python hl_lines="4  9-12  25-27"
from typing import Union

from fastapi import FastAPI

	return nil
}

// Return the volume plugin name, together with the CSI driver name if it's a CSI volume.
func getVolumePluginNameWithDriver(plugin volume.VolumePlugin, spec *volume.Spec) string {
	pluginName := plugin.GetPluginName()
	if pluginName != csi.CSIPluginName {
		return pluginName
	}

	// It's a CSI volume
	driverName, err := csi.GetCSIDriverName(spec)
	if err != nil {

	pvcQueue workqueue.TypedRateLimitingInterface[string]

	// csiMigratedPluginManager detects in-tree plugins that have been migrated to CSI
	csiMigratedPluginManager csimigration.PluginManager

	// intreeToCSITranslator translates from in-tree volume specs to CSI
	intreeToCSITranslator csimigration.InTreeToCSITranslator
}

func (adc *attachDetachController) Run(ctx context.Context) {
	defer runtime.HandleCrash()

	migratable, err := expc.csiMigratedPluginManager.IsMigratable(volumeSpec)
	if err != nil {
		logger.V(4).Info("Failed to check CSI migration status for PVC with error", "pvcKey", key, "err", err)
		return nil
	}
	// handle CSI migration scenarios before invoking FindExpandablePluginBySpec for in-tree
	if migratable {

			errors:          noerrors,
			test:            testSyncClaim,
		},
		{
			// Provision a volume with a data source will proceed
			// for CSI plugins
			"11-26 - csi with data source",
			novolumes,
			novolumes,
			claimWithAnnotation(volume.AnnStorageProvisioner, "mydriver.csi.k8s.io",

		}
		unescapePluginName := utilstrings.UnescapeQualifiedName(volumePluginName)

		if unescapePluginName != csi.CSIPluginName {
			for _, volumeDir := range volumeDirs {
				volumes = append(volumes, filepath.Join(volumePluginPath, volumeDir))
			}
		} else {
			// For CSI volumes, the mounted volume path has an extra sub path "/mount", so also add it
			// to the list if the mounted path exists.

			logger.V(5).Info("Could not get a CSINode object for the node", "node", klog.KObj(node), "err", err)
		}
	}

	// if a plugin has been migrated to a CSI driver, defer to the CSI predicate
	if pl.filter.IsMigrated(csiNode) {
		return nil
	}

	// count unique volumes
	existingVolumes := sets.New[string]()
	for _, existingPod := range nodeInfo.Pods {

	R10W: "R10W",
	R11W: "R11W",
	R12W: "R12W",
	R13W: "R13W",
	R14W: "R14W",
	R15W: "R15W",
	EAX:  "EAX",
	ECX:  "ECX",
	EDX:  "EDX",
	EBX:  "EBX",
	ESP:  "ESP",
	EBP:  "EBP",
	ESI:  "ESI",
	EDI:  "EDI",
	R8L:  "R8L",
	R9L:  "R9L",
	R10L: "R10L",
	R11L: "R11L",
	R12L: "R12L",
	R13L: "R13L",
	R14L: "R14L",
	R15L: "R15L",
	RAX:  "RAX",
	RCX:  "RCX",
	RDX:  "RDX",