diff --git a/pkg/ddc/base/dataset_setup_volume_test.go b/pkg/ddc/base/dataset_setup_volume_test.go
new file mode 100644
index 00000000000..5c5117add04
--- /dev/null
+++ b/pkg/ddc/base/dataset_setup_volume_test.go
@@ -0,0 +1,408 @@
+/*
+Copyright 2026 The Fluid Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package base
+
+import (
+	"context"
+	"errors"
+
+	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
+	"github.com/fluid-cloudnative/fluid/pkg/dataoperation"
+	cruntime "github.com/fluid-cloudnative/fluid/pkg/runtime"
+	"github.com/fluid-cloudnative/fluid/pkg/utils"
+	"github.com/fluid-cloudnative/fluid/pkg/utils/fake"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+const (
+	testDatasetName      = "spark"
+	testDatasetNamespace = "fluid"
+	testDatasetUID       = "dataset-uid"
+	volumeContextKey     = testContextKey("volume")
+)
+
+type testContextKey string
+
+type stubImplement struct {
+	shouldSetupMasterFn           func() (bool, error)
+	setupMasterFn                 func() error
+	checkMasterReadyFn            func() (bool, error)
+	shouldCheckUFSFn              func() (bool, error)
+	prepareUFSFn                  func() error
+	shouldSetupWorkersFn          func() (bool, error)
+	setupWorkersFn                func() error
+	checkWorkersReadyFn           func() (bool, error)
+	checkAndUpdateRuntimeStatusFn func() (bool, error)
+	bindToDatasetFn               func() error
+	createVolumeFn                func(context.Context) error
+	deleteVolumeFn                func(context.Context) error
+	shouldSetupMasterCalls        int
+	checkMasterReadyCalls         int
+}
+
+func (s *stubImplement) GetDataOperationValueFile(cruntime.ReconcileRequestContext, dataoperation.OperationInterface) (string, error) {
+	return "", nil
+}
+
+func (s *stubImplement) CheckMasterReady() (bool, error) {
+	s.checkMasterReadyCalls++
+	if s.checkMasterReadyFn != nil {
+		return s.checkMasterReadyFn()
+	}
+	return false, nil
+}
+
+func (s *stubImplement) CheckWorkersReady() (bool, error) {
+	if s.checkWorkersReadyFn != nil {
+		return s.checkWorkersReadyFn()
+	}
+	return false, nil
+}
+
+func (s *stubImplement) ShouldSetupMaster() (bool, error) {
+	s.shouldSetupMasterCalls++
+	if s.shouldSetupMasterFn != nil {
+		return s.shouldSetupMasterFn()
+	}
+	return false, nil
+}
+
+func (s *stubImplement) ShouldSetupWorkers() (bool, error) {
+	if s.shouldSetupWorkersFn != nil {
+		return s.shouldSetupWorkersFn()
+	}
+	return false, nil
+}
+
+func (s *stubImplement) ShouldCheckUFS() (bool, error) {
+	if s.shouldCheckUFSFn != nil {
+		return s.shouldCheckUFSFn()
+	}
+	return false, nil
+}
+
+func (s *stubImplement) SetupMaster() error {
+	if s.setupMasterFn != nil {
+		return s.setupMasterFn()
+	}
+	return nil
+}
+
+func (s *stubImplement) SetupWorkers() error {
+	if s.setupWorkersFn != nil {
+		return s.setupWorkersFn()
+	}
+	return nil
+}
+
+func (s *stubImplement) UpdateDatasetStatus(datav1alpha1.DatasetPhase) error { return nil }
+
+func (s *stubImplement) PrepareUFS() error {
+	if s.prepareUFSFn != nil {
+		return s.prepareUFSFn()
+	}
+	return nil
+}
+
+func (s *stubImplement) ShouldSyncDatasetMounts() (bool, error) { return false, nil }
+
+func (s *stubImplement) SyncDatasetMounts() error { return nil }
+
+func (s *stubImplement) ShouldUpdateUFS() *utils.UFSToUpdate { return nil }
+
+func (s *stubImplement) UpdateOnUFSChange(*utils.UFSToUpdate) (bool, error) { return false, nil }
+
+func (s *stubImplement) Shutdown() error { return nil }
+
+func (s *stubImplement) CheckRuntimeHealthy() error { return nil }
+
+func (s *stubImplement) UpdateCacheOfDataset() error { return nil }
+
+func (s *stubImplement) CheckAndUpdateRuntimeStatus() (bool, error) {
+	if s.checkAndUpdateRuntimeStatusFn != nil {
+		return s.checkAndUpdateRuntimeStatusFn()
+	}
+	return false, nil
+}
+
+func (s *stubImplement) CreateVolume(ctx context.Context) error {
+	if s.createVolumeFn != nil {
+		return s.createVolumeFn(ctx)
+	}
+	return nil
+}
+
+func (s *stubImplement) SyncReplicas(cruntime.ReconcileRequestContext) error { return nil }
+
+func (s *stubImplement) SyncMetadata() error { return nil }
+
+func (s *stubImplement) DeleteVolume(ctx context.Context) error {
+	if s.deleteVolumeFn != nil {
+		return s.deleteVolumeFn(ctx)
+	}
+	return nil
+}
+
+func (s *stubImplement) BindToDataset() error {
+	if s.bindToDatasetFn != nil {
+		return s.bindToDatasetFn()
+	}
+	return nil
+}
+
+func (s *stubImplement) CheckRuntimeReady() bool { return false }
+
+func (s *stubImplement) SyncRuntime(cruntime.ReconcileRequestContext) (bool, error) {
+	return false, nil
+}
+
+func (s *stubImplement) SyncScheduleInfoToCacheNodes() error { return nil }
+
+func (s *stubImplement) Validate(cruntime.ReconcileRequestContext) error { return nil }
+
+func (s *stubImplement) UsedStorageBytes() (int64, error) { return 0, nil }
+
+func (s *stubImplement) FreeStorageBytes() (int64, error) { return 0, nil }
+
+func (s *stubImplement) TotalStorageBytes() (int64, error) { return 0, nil }
+
+func (s *stubImplement) TotalFileNums() (int64, error) { return 0, nil }
+
+func (s *stubImplement) Operate(cruntime.ReconcileRequestContext, *datav1alpha1.OperationStatus, dataoperation.OperationInterface) (ctrl.Result, error) {
+	return ctrl.Result{}, nil
+}
+
+func newTemplateEngineForTest(impl Implement) (*TemplateEngine, cruntime.ReconcileRequestContext) {
+	ctx := cruntime.ReconcileRequestContext{
+		Context: context.Background(),
+		NamespacedName: types.NamespacedName{
+			Namespace: testDatasetNamespace,
+			Name:      testDatasetName,
+		},
+		Log:         fake.NullLogger(),
+		RuntimeType: "alluxio",
+		Runtime: &datav1alpha1.AlluxioRuntime{
+			TypeMeta: metav1.TypeMeta{Kind: "AlluxioRuntime"},
+		},
+	}
+
+	return NewTemplateEngine(impl, "test-engine", ctx), ctx
+}
+
+var _ = Describe("TemplateEngine setup", func() {
+	var (
+		impl   *stubImplement
+		engine *TemplateEngine
+		ctx    cruntime.ReconcileRequestContext
+	)
+
+	BeforeEach(func() {
+		impl = &stubImplement{}
+		engine, ctx = newTemplateEngineForTest(impl)
+	})
+
+	It("returns ready when master, workers, and runtime are all ready", func() {
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, nil }
+		impl.checkMasterReadyFn = func() (bool, error) { return true, nil }
+		impl.shouldCheckUFSFn = func() (bool, error) { return false, nil }
+		impl.shouldSetupWorkersFn = func() (bool, error) { return false, nil }
+		impl.checkWorkersReadyFn = func() (bool, error) { return true, nil }
+		impl.checkAndUpdateRuntimeStatusFn = func() (bool, error) { return true, nil }
+		impl.bindToDatasetFn = func() error { return nil }
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(ready).To(BeTrue())
+	})
+
+	It("propagates errors while deciding whether master setup is required", func() {
+		expectedErr := errors.New("should setup master failed")
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, expectedErr }
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).To(MatchError(expectedErr))
+		Expect(ready).To(BeFalse())
+		Expect(impl.shouldSetupMasterCalls).To(Equal(1))
+		Expect(impl.checkMasterReadyCalls).To(BeZero())
+	})
+
+	It("short-circuits when the master is not ready yet", func() {
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, nil }
+		impl.checkMasterReadyFn = func() (bool, error) { return false, nil }
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(ready).To(BeFalse())
+	})
+
+	It("keeps zero-value stub defaults from reporting a misleading ready state", func() {
+		runtimeStatusChecked := false
+		datasetBound := false
+		impl.checkAndUpdateRuntimeStatusFn = func() (bool, error) {
+			runtimeStatusChecked = true
+			return true, nil
+		}
+		impl.bindToDatasetFn = func() error {
+			datasetBound = true
+			return nil
+		}
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(ready).To(BeFalse())
+		Expect(impl.shouldSetupMasterCalls).To(Equal(1))
+		Expect(impl.checkMasterReadyCalls).To(Equal(1))
+		Expect(runtimeStatusChecked).To(BeFalse())
+		Expect(datasetBound).To(BeFalse())
+	})
+
+	It("short-circuits when the workers are not ready yet", func() {
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, nil }
+		impl.checkMasterReadyFn = func() (bool, error) { return true, nil }
+		impl.shouldCheckUFSFn = func() (bool, error) { return false, nil }
+		impl.shouldSetupWorkersFn = func() (bool, error) { return false, nil }
+		impl.checkWorkersReadyFn = func() (bool, error) { return false, nil }
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(ready).To(BeFalse())
+	})
+
+	It("propagates errors while checking whether the master is ready", func() {
+		expectedErr := errors.New("check master ready failed")
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, nil }
+		impl.checkMasterReadyFn = func() (bool, error) { return false, expectedErr }
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).To(MatchError(expectedErr))
+		Expect(ready).To(BeFalse())
+		Expect(impl.shouldSetupMasterCalls).To(Equal(1))
+		Expect(impl.checkMasterReadyCalls).To(Equal(1))
+	})
+
+	It("propagates runtime status update errors without binding the dataset", func() {
+		expectedErr := errors.New("runtime status failed")
+		bindCalled := false
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, nil }
+		impl.checkMasterReadyFn = func() (bool, error) { return true, nil }
+		impl.shouldCheckUFSFn = func() (bool, error) { return false, nil }
+		impl.shouldSetupWorkersFn = func() (bool, error) { return false, nil }
+		impl.checkWorkersReadyFn = func() (bool, error) { return true, nil }
+		impl.checkAndUpdateRuntimeStatusFn = func() (bool, error) { return true, expectedErr }
+		impl.bindToDatasetFn = func() error {
+			bindCalled = true
+			return nil
+		}
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).To(MatchError(expectedErr))
+		Expect(ready).To(BeTrue())
+		Expect(bindCalled).To(BeFalse())
+	})
+
+	It("propagates dataset binding errors after the runtime becomes ready", func() {
+		expectedErr := errors.New("bind failed")
+		bindCalled := false
+		impl.shouldSetupMasterFn = func() (bool, error) { return false, nil }
+		impl.checkMasterReadyFn = func() (bool, error) { return true, nil }
+		impl.shouldCheckUFSFn = func() (bool, error) { return false, nil }
+		impl.shouldSetupWorkersFn = func() (bool, error) { return false, nil }
+		impl.checkWorkersReadyFn = func() (bool, error) { return true, nil }
+		impl.checkAndUpdateRuntimeStatusFn = func() (bool, error) { return true, nil }
+		impl.bindToDatasetFn = func() error {
+			bindCalled = true
+			return expectedErr
+		}
+
+		ready, err := engine.Setup(ctx)
+
+		Expect(err).To(MatchError(expectedErr))
+		Expect(ready).To(BeTrue())
+		Expect(bindCalled).To(BeTrue())
+	})
+})
+
+var _ = Describe("TemplateEngine volume operations", func() {
+	var (
+		impl   *stubImplement
+		engine *TemplateEngine
+	)
+
+	BeforeEach(func() {
+		impl = &stubImplement{}
+		engine, _ = newTemplateEngineForTest(impl)
+	})
+
+	It("delegates CreateVolume to the implementation", func() {
+		volumeCtx := context.WithValue(context.Background(), volumeContextKey, "create")
+		called := false
+		impl.createVolumeFn = func(ctx context.Context) error {
+			called = true
+			Expect(ctx).To(BeIdenticalTo(volumeCtx))
+			return nil
+		}
+
+		Expect(engine.CreateVolume(volumeCtx)).To(Succeed())
+		Expect(called).To(BeTrue())
+	})
+
+	It("propagates CreateVolume errors", func() {
+		volumeCtx := context.WithValue(context.Background(), volumeContextKey, "create-error")
+		expectedErr := errors.New("create failed")
+		impl.createVolumeFn = func(ctx context.Context) error {
+			Expect(ctx).To(BeIdenticalTo(volumeCtx))
+			return expectedErr
+		}
+
+		Expect(engine.CreateVolume(volumeCtx)).To(MatchError(expectedErr))
+	})
+
+	It("delegates DeleteVolume to the implementation", func() {
+		volumeCtx := context.WithValue(context.Background(), volumeContextKey, "delete")
+		called := false
+		impl.deleteVolumeFn = func(ctx context.Context) error {
+			called = true
+			Expect(ctx).To(BeIdenticalTo(volumeCtx))
+			return nil
+		}
+
+		Expect(engine.DeleteVolume(volumeCtx)).To(Succeed())
+		Expect(called).To(BeTrue())
+	})
+
+	It("propagates DeleteVolume errors", func() {
+		volumeCtx := context.WithValue(context.Background(), volumeContextKey, "delete-error")
+		expectedErr := errors.New("delete failed")
+		impl.deleteVolumeFn = func(ctx context.Context) error {
+			Expect(ctx).To(BeIdenticalTo(volumeCtx))
+			return expectedErr
+		}
+
+		Expect(engine.DeleteVolume(volumeCtx)).To(MatchError(expectedErr))
+	})
+})
diff --git a/pkg/ddc/base/dataset_test.go b/pkg/ddc/base/dataset_test.go
index df01da02a17..a866309bea5 100644
--- a/pkg/ddc/base/dataset_test.go
+++ b/pkg/ddc/base/dataset_test.go
@@ -17,252 +17,162 @@
 package base
 
 import (
-	"reflect"
-	"testing"
-
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 )
 
-func TestGetPhysicalDatasetFromMounts(t *testing.T) {
-	tests := []struct {
-		virtualDataset *datav1alpha1.Dataset
-		want           int
-	}{
-		{
-			virtualDataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
-				},
-				Spec: datav1alpha1.DatasetSpec{
-					Mounts: []datav1alpha1.Mount{
-						{
-							MountPoint: "dataset://ns-a/n-a",
-						},
-						{
-							MountPoint: "dataset://ns-b/n-b",
-						},
-					},
-				},
+var _ = Describe("Dataset", func() {
+	Describe("GetPhysicalDatasetFromMounts", func() {
+		DescribeTable("counts only dataset reference mounts",
+			func(mounts []datav1alpha1.Mount, expected []types.NamespacedName) {
+				Expect(GetPhysicalDatasetFromMounts(mounts)).To(Equal(expected))
 			},
-			want: 2,
-		},
-		{
-			virtualDataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
+			Entry("all mounts are dataset references",
+				[]datav1alpha1.Mount{
+					{MountPoint: "dataset://ns-a/n-a"},
+					{MountPoint: "dataset://ns-b/n-b"},
 				},
-				Spec: datav1alpha1.DatasetSpec{
-					Mounts: []datav1alpha1.Mount{
-						{
-							MountPoint: "dataset://ns-a/n-a",
-						},
-						{
-							MountPoint: "http://ns-b/n-b",
-						},
-					},
+				[]types.NamespacedName{{Namespace: "ns-a", Name: "n-a"}, {Namespace: "ns-b", Name: "n-b"}},
+			),
+			Entry("mixed mount types only return dataset references",
+				[]datav1alpha1.Mount{
+					{MountPoint: "dataset://ns-a/n-a"},
+					{MountPoint: "http://ns-b/n-b"},
 				},
-			},
-			want: 1,
-		},
-		{
-			virtualDataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
-				},
-				Spec: datav1alpha1.DatasetSpec{},
-			},
-			want: 0,
-		},
-	}
-	for _, tt := range tests {
-		if got := GetPhysicalDatasetFromMounts(tt.virtualDataset.Spec.Mounts); len(got) != tt.want {
-			t.Errorf("GetPhysicalDatasetFromMounts() len = %v, want %v", got, tt.want)
-		}
-	}
-}
+				[]types.NamespacedName{{Namespace: "ns-a", Name: "n-a"}},
+			),
+			Entry("empty mounts return no dataset references", nil, nil),
+		)
+	})
 
-func TestGetDatasetRefName(t *testing.T) {
-	refNameA := GetDatasetRefName("a-b", "c")
-	refNameB := GetDatasetRefName("a", "bc")
+	Describe("GetDatasetRefName", func() {
+		It("includes both namespace and name in the ref name", func() {
+			refNameA := GetDatasetRefName("a-b", "c")
+			refNameB := GetDatasetRefName("a", "bc")
 
-	if refNameB == refNameA {
-		t.Errorf("RefName is equal for different name and namespace")
-	}
-}
+			Expect(refNameA).To(Equal("c/a-b"))
+			Expect(refNameB).To(Equal("bc/a"))
+			Expect(refNameB).NotTo(Equal(refNameA))
+		})
+	})
 
-func TestCheckReferenceDataset(t *testing.T) {
+	Describe("CheckReferenceDataset", func() {
+		DescribeTable("validates reference dataset mounts",
+			func(dataset *datav1alpha1.Dataset, wantCheck bool, wantErr bool) {
+				gotCheck, err := CheckReferenceDataset(dataset)
 
-	tests := []struct {
-		name      string
-		dataset   *datav1alpha1.Dataset
-		wantCheck bool
-		wantErr   bool
-	}{
-		// TODO: Add test cases.
-		{
-			name: "dataset_with_two_datasetmounts",
-			dataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
-				},
-				Spec: datav1alpha1.DatasetSpec{
-					Mounts: []datav1alpha1.Mount{
-						{
-							MountPoint: "dataset://ns-a/n-a",
-						},
-						{
-							MountPoint: "dataset://ns-b/n-b",
-						},
-					},
-				},
-			},
-			wantCheck: false,
-			wantErr:   true,
-		},
-		{
-			name: "dataset_with_two_datasetmounts",
-			dataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
-				},
-				Spec: datav1alpha1.DatasetSpec{
-					Mounts: []datav1alpha1.Mount{
-						{
-							MountPoint: "dataset://ns-a/n-a",
-						},
-						{
-							MountPoint: "http://ns-b/n-b",
-						},
-					},
-				},
-			},
-			wantCheck: false,
-			wantErr:   true,
-		},
-		{
-			name: "referenced_dataset",
-			dataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
-				},
-				Spec: datav1alpha1.DatasetSpec{
-					Mounts: []datav1alpha1.Mount{
-						{
-							MountPoint: "dataset://ns-a/n-a",
-						},
-					},
-				},
+				Expect(gotCheck).To(Equal(wantCheck))
+				if wantErr {
+					Expect(err).To(HaveOccurred())
+				} else {
+					Expect(err).NotTo(HaveOccurred())
+				}
 			},
-			wantCheck: true,
-			wantErr:   false,
-		},
-		{
-			name: "no_mounts",
-			dataset: &datav1alpha1.Dataset{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "hbase",
-					Namespace: "fluid",
-				},
-				Spec: datav1alpha1.DatasetSpec{},
+			Entry("rejects two dataset mounts",
+				&datav1alpha1.Dataset{Spec: datav1alpha1.DatasetSpec{Mounts: []datav1alpha1.Mount{{MountPoint: "dataset://ns-a/n-a"}, {MountPoint: "dataset://ns-b/n-b"}}}},
+				false,
+				true,
+			),
+			Entry("rejects mixed dataset and non-dataset mounts",
+				&datav1alpha1.Dataset{Spec: datav1alpha1.DatasetSpec{Mounts: []datav1alpha1.Mount{{MountPoint: "dataset://ns-a/n-a"}, {MountPoint: "http://ns-b/n-b"}}}},
+				false,
+				true,
+			),
+			Entry("accepts a single dataset mount",
+				&datav1alpha1.Dataset{Spec: datav1alpha1.DatasetSpec{Mounts: []datav1alpha1.Mount{{MountPoint: "dataset://ns-a/n-a"}}}},
+				true,
+				false,
+			),
+			Entry("accepts datasets with no mounts",
+				&datav1alpha1.Dataset{},
+				false,
+				false,
+			),
+		)
+	})
+
+	Describe("GetPhysicalDatasetSubPath", func() {
+		DescribeTable("extracts dataset subpaths",
+			func(dataset *datav1alpha1.Dataset, expected []string) {
+				Expect(GetPhysicalDatasetSubPath(dataset)).To(Equal(expected))
 			},
-			wantCheck: false,
-			wantErr:   false,
-		},
-	}
+			Entry("returns nested subpaths",
+				&datav1alpha1.Dataset{Spec: datav1alpha1.DatasetSpec{Mounts: []datav1alpha1.Mount{{MountPoint: "dataset://ns-a/ns-b/sub-c/sub-d"}}}},
+				[]string{"sub-c/sub-d"},
+			),
+			Entry("returns an empty subpath when path ends with slash",
+				&datav1alpha1.Dataset{Spec: datav1alpha1.DatasetSpec{Mounts: []datav1alpha1.Mount{{MountPoint: "dataset://ns-a/ns-b/"}}}},
+				[]string{""},
+			),
+			Entry("returns nil when no subpath exists",
+				&datav1alpha1.Dataset{Spec: datav1alpha1.DatasetSpec{Mounts: []datav1alpha1.Mount{{MountPoint: "dataset://ns-a/ns-b"}}}},
+				nil,
+			),
+		)
+	})
 
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			gotCheck, err := CheckReferenceDataset(tt.dataset)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Testcase %v CheckReferenceDataset() error = %v, wantErr %v", tt.name, err, tt.wantErr)
-				return
-			}
-			if gotCheck != tt.wantCheck {
-				t.Errorf("Testcase %v CheckReferenceDataset() = %v, want %v", tt.name, gotCheck, tt.wantCheck)
-			}
-		})
-	}
-}
+	Describe("GetOwnerDatasetUIDFromRuntimeMeta", func() {
+		DescribeTable("extracts the owner dataset UID from runtime metadata",
+			func(meta metav1.ObjectMeta, expectedUID types.UID, expectedError string) {
+				uid, err := GetOwnerDatasetUIDFromRuntimeMeta(meta)
 
-func TestGetPhysicalDatasetSubPath(t *testing.T) {
-	type args struct {
-		dataset *datav1alpha1.Dataset
-	}
-	tests := []struct {
-		name string
-		args args
-		want []string
-	}{
-		{
-			name: "non empty sub path",
-			args: args{
-				dataset: &datav1alpha1.Dataset{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "hbase",
-						Namespace: "fluid",
-					},
-					Spec: datav1alpha1.DatasetSpec{
-						Mounts: []datav1alpha1.Mount{
-							{
-								MountPoint: "dataset://ns-a/ns-b/sub-c/sub-d",
-							},
-						},
-					},
-				},
+				Expect(uid).To(Equal(expectedUID))
+				if expectedError == "" {
+					Expect(err).NotTo(HaveOccurred())
+				} else {
+					Expect(err).To(MatchError(ContainSubstring(expectedError)))
+				}
 			},
-			want: []string{"sub-c/sub-d"},
-		},
-		{
-			name: "empty sub path",
-			args: args{
-				dataset: &datav1alpha1.Dataset{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "hbase",
-						Namespace: "fluid",
-					},
-					Spec: datav1alpha1.DatasetSpec{
-						Mounts: []datav1alpha1.Mount{
-							{
-								MountPoint: "dataset://ns-a/ns-b/",
-							},
-						},
-					},
+			Entry("returns the UID for a matching dataset owner",
+				metav1.ObjectMeta{
+					Name: "spark",
+					OwnerReferences: []metav1.OwnerReference{{
+						Kind: datav1alpha1.Datasetkind,
+						Name: "spark",
+						UID:  types.UID("dataset-uid"),
+					}},
 				},
-			},
-			want: []string{""},
-		},
-		{
-			name: "no sub path",
-			args: args{
-				dataset: &datav1alpha1.Dataset{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "hbase",
-						Namespace: "fluid",
-					},
-					Spec: datav1alpha1.DatasetSpec{
-						Mounts: []datav1alpha1.Mount{
-							{
-								MountPoint: "dataset://ns-a/ns-b",
-							},
-						},
+				types.UID("dataset-uid"),
+				"",
+			),
+			Entry("returns empty UID when no dataset owner exists",
+				metav1.ObjectMeta{
+					Name: "spark",
+					OwnerReferences: []metav1.OwnerReference{{
+						Kind: "Job",
+						Name: "spark",
+						UID:  types.UID("job-uid"),
+					}},
+				},
+				types.UID(""),
+				"",
+			),
+			Entry("rejects multiple dataset owners",
+				metav1.ObjectMeta{
+					Name: "spark",
+					OwnerReferences: []metav1.OwnerReference{
+						{Kind: datav1alpha1.Datasetkind, Name: "spark", UID: types.UID("dataset-uid")},
+						{Kind: datav1alpha1.Datasetkind, Name: "spark", UID: types.UID("dataset-uid-2")},
 					},
 				},
-			},
-			want: nil,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetPhysicalDatasetSubPath(tt.args.dataset); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("GetPhysicalDatasetSubPath() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
+				types.UID(""),
+				"found multiple Dataset owners",
+			),
+			Entry("rejects dataset owners whose name differs from the runtime",
+				metav1.ObjectMeta{
+					Name: "spark",
+					OwnerReferences: []metav1.OwnerReference{{
+						Kind: datav1alpha1.Datasetkind,
+						Name: "other-dataset",
+						UID:  types.UID("dataset-uid"),
+					}},
+				},
+				types.UID(""),
+				"expected to be same",
+			),
+		)
+	})
+})