diff --git a/.github/scripts/build-all-images.sh b/.github/scripts/build-all-images.sh
index bee4c510e96..dfc25427585 100755
--- a/.github/scripts/build-all-images.sh
+++ b/.github/scripts/build-all-images.sh
@@ -11,6 +11,7 @@ get_image_tag() {
 
 build_images() {
     oss_emulator_img="${IMG_REPO}/oss-emulator:e2e"
+    mooncake_img="${IMG_REPO}/mooncake:e2e"
     images=(
         "${IMG_REPO}/dataset-controller:${IMAGE_TAG}"
         "${IMG_REPO}/application-controller:${IMAGE_TAG}"
@@ -23,10 +24,12 @@ build_images() {
         "${IMG_REPO}/fluid-webhook:${IMAGE_TAG}"
         "${IMG_REPO}/fluid-crd-upgrader:${IMAGE_TAG}"
         "${oss_emulator_img}"
+        "${mooncake_img}"
     )
 
     make docker-build-all
     docker build -t "${oss_emulator_img}" test/gha-e2e/jindo/oss-emulator
+    docker build -t "${mooncake_img}" test/gha-e2e/mooncake/image
 
     echo ">>> Cleaning docker build caches before loading images to free disk space..."
     docker builder prune -a -f
diff --git a/.github/scripts/gha-e2e.sh b/.github/scripts/gha-e2e.sh
index 2c7cf04464e..9dde03a9fb5 100755
--- a/.github/scripts/gha-e2e.sh
+++ b/.github/scripts/gha-e2e.sh
@@ -105,8 +105,14 @@ function curvine_e2e() {
     bash test/gha-e2e/curvine/test.sh
 }
 
+function mooncake_e2e() {
+    set -e
+    bash test/gha-e2e/mooncake/test.sh
+}
+
 check_control_plane_status
 alluxio_e2e
 jindo_e2e
 juicefs_e2e
 curvine_e2e
+mooncake_e2e
diff --git a/test/gha-e2e/mooncake/bad_mount_pod.yaml b/test/gha-e2e/mooncake/bad_mount_pod.yaml
new file mode 100644
index 00000000000..d3d8e21d4f2
--- /dev/null
+++ b/test/gha-e2e/mooncake/bad_mount_pod.yaml
@@ -0,0 +1,29 @@
+# Negative case: without a client component there is no FUSE mount point, so an
+# application pod that mounts this Dataset's PVC is bound to fail. Fluid still
+# creates the PVC/PV and reports them as Bound, which easily gives the
+# impression that they can be mounted, so the docs call this out in their FAQ
+# and this test pins the behaviour down.
+#
+# This pod is expected never to start; it is deleted as soon as the check is done.
+apiVersion: v1
+kind: Pod
+metadata:
+  name: mooncake-bad-mount
+  namespace: default
+spec:
+  # The mount keeps being retried anyway, no need to have kubelet restart the
+  # container over and over
+  restartPolicy: Never
+  automountServiceAccountToken: false
+  containers:
+    - name: app
+      image: busybox:1.36
+      imagePullPolicy: IfNotPresent
+      command: ["sleep", "infinity"]
+      volumeMounts:
+        - name: mc-vol
+          mountPath: /data
+  volumes:
+    - name: mc-vol
+      persistentVolumeClaim:
+        claimName: mooncake-demo
diff --git a/test/gha-e2e/mooncake/cacheruntime.yaml b/test/gha-e2e/mooncake/cacheruntime.yaml
new file mode 100644
index 00000000000..ffe26868311
--- /dev/null
+++ b/test/gha-e2e/mooncake/cacheruntime.yaml
@@ -0,0 +1,20 @@
+# One worker replica: the CI kind cluster is single-node, so extra replicas would
+# show no scheduling difference and only make the case slower. Replica scaling is
+# already covered by the curvine case and is not repeated here.
+apiVersion: data.fluid.io/v1alpha1
+kind: CacheRuntime
+metadata:
+  name: mooncake-demo
+  namespace: default
+spec:
+  runtimeClassName: mooncake-demo
+  master:
+    replicas: 1
+  worker:
+    replicas: 1
+    tieredStore:
+      levels:
+        - emptyDir:
+            quota: 1Gi
+          high: "0.8"
+          low: "0.5"
diff --git a/test/gha-e2e/mooncake/cacheruntimeclass.yaml b/test/gha-e2e/mooncake/cacheruntimeclass.yaml
new file mode 100644
index 00000000000..76d6da126f7
--- /dev/null
+++ b/test/gha-e2e/mooncake/cacheruntimeclass.yaml
@@ -0,0 +1,92 @@
+# Mooncake is a cache system without POSIX mount semantics: applications talk to
+# the cache service directly through its own client, so this topology declares
+# only master and worker and omits the client component that would do the
+# mounting. That is exactly the scenario this case guards: the controller must
+# not panic when the client component is omitted.
+#
+# Note: do not declare resources on the containers here. When the
+# CacheRuntimeClass template sets resources but the CacheRuntime does not,
+# syncRuntimeSpec passes the zero value to updateResources, which overwrites the
+# template's resources with an empty value. That has two consequences:
+#   1. the resources declared in the template are silently dropped and the pod
+#      ends up with {};
+#   2. the patch triggers one ASTS rollout, during which the runtime is briefly
+#      not-ready, and once the Dataset flips to Failed it does not recover on its
+#      own (even after the component phase returns to Ready).
+apiVersion: data.fluid.io/v1alpha1
+kind: CacheRuntimeClass
+metadata:
+  name: mooncake-demo
+fileSystemType: mooncakefs
+topology:
+  master:
+    service:
+      headless: {}
+    executionEntries:
+      reportSummary:
+        command:
+          - bash
+          - -c
+          - /reportSummary.sh
+        timeout: 30
+    template:
+      spec:
+        restartPolicy: Always
+        containers:
+          - name: master
+            image: fluidcloudnative/mooncake:e2e
+            command:
+              - /custom-entrypoint.sh
+            args:
+              - master
+              - start
+            imagePullPolicy: IfNotPresent
+            readinessProbe:
+              tcpSocket:
+                port: 50051
+              initialDelaySeconds: 10
+              periodSeconds: 5
+              failureThreshold: 12
+            env:
+              - name: POD_NAME
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.name
+            ports:
+              - containerPort: 50051
+                name: rpc
+              - containerPort: 8080
+                name: metadata
+              - containerPort: 9003
+                name: metrics
+  worker:
+    service:
+      headless: {}
+    template:
+      spec:
+        restartPolicy: Always
+        containers:
+          - name: worker
+            image: fluidcloudnative/mooncake:e2e
+            command:
+              - /custom-entrypoint.sh
+            args:
+              - worker
+              - start
+            imagePullPolicy: IfNotPresent
+            readinessProbe:
+              tcpSocket:
+                port: 50052
+              initialDelaySeconds: 5
+              periodSeconds: 5
+              failureThreshold: 12
+            env:
+              - name: POD_NAME
+                valueFrom:
+                  fieldRef:
+                    fieldPath: metadata.name
+            ports:
+              - containerPort: 50052
+                name: data
+              - containerPort: 9300
+                name: http
diff --git a/test/gha-e2e/mooncake/dataset.yaml b/test/gha-e2e/mooncake/dataset.yaml
new file mode 100644
index 00000000000..bcf0427230f
--- /dev/null
+++ b/test/gha-e2e/mooncake/dataset.yaml
@@ -0,0 +1,16 @@
+# Mooncake mounts no underlying storage (UFS); data is written directly into the
+# cache by the client. The mountPoint here is only a placeholder: the
+# CacheRuntimeClass declares no mountUfs execution entry, so Fluid never performs
+# an actual mount.
+apiVersion: data.fluid.io/v1alpha1
+kind: Dataset
+metadata:
+  name: mooncake-demo
+  namespace: default
+spec:
+  placement: Shared
+  accessModes:
+    - ReadWriteMany
+  mounts:
+    - name: mc
+      mountPoint: "mooncakefs:///"
diff --git a/test/gha-e2e/mooncake/image/Dockerfile b/test/gha-e2e/mooncake/image/Dockerfile
new file mode 100644
index 00000000000..bbe128f2bbf
--- /dev/null
+++ b/test/gha-e2e/mooncake/image/Dockerfile
@@ -0,0 +1,41 @@
+# The Mooncake image used by the e2e case.
+#
+# It is built in-repo rather than pulled from an external registry, for the same
+# reason as test/gha-e2e/jindo/oss-emulator: e2e runs on every PR, an external
+# image going away turns the whole pipeline red, and the two scripts the image
+# provides for Fluid to invoke have to be reviewable.
+#
+# The only external dependency is Mooncake's official package on PyPI.
+# python:3.12.13-slim — slim is required: the full variant carries an extra
+# ~700MB build toolchain, growing the image from ~470MB to ~1.9GB, and e2e also
+# has to kind-load it onto the node and keep a second copy there.
+FROM python:3.12.13-slim@sha256:229a2c5bfa27522db7815ea81f9bed70af17ccb9de9fc7ad142b1877b5830d36
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        libcurl4 \
+        libibverbs1 \
+        rdma-core \
+        librdmacm1 \
+        libnuma1 \
+        liburing2 \
+        curl \
+        jq \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN pip install --no-cache-dir \
+        mooncake-transfer-engine-non-cuda==0.3.12.post1 \
+        nvidia-cuda-runtime-cu12
+
+ENV LD_LIBRARY_PATH=/usr/local/lib/python3.12/site-packages/nvidia/cuda_runtime/lib:$LD_LIBRARY_PATH
+
+# 50051 master RPC / 50052 worker data / 8080 metadata service / 9003 master metrics / 9300 worker http
+EXPOSE 50051 50052 8080 9003 9300
+
+# The two scripts Fluid invokes:
+#   custom-entrypoint.sh  starts the right process based on the role (master/worker)
+#   reportSummary.sh      collects cache usage and emits it as JSON in the format Fluid expects
+COPY custom-entrypoint.sh /custom-entrypoint.sh
+COPY reportSummary.sh /reportSummary.sh
+RUN chmod +x /custom-entrypoint.sh /reportSummary.sh
+
+CMD ["python3"]
diff --git a/test/gha-e2e/mooncake/image/custom-entrypoint.sh b/test/gha-e2e/mooncake/image/custom-entrypoint.sh
new file mode 100644
index 00000000000..85e45fc2b0d
--- /dev/null
+++ b/test/gha-e2e/mooncake/image/custom-entrypoint.sh
@@ -0,0 +1,71 @@
+#!/bin/sh
+# custom-entrypoint.sh
+# Usage: /custom-entrypoint.sh <master|worker|client> <start>
+
+set -e
+
+ROLE="$1"
+ACTION="$2"
+
+if [ "$ACTION" != "start" ]; then
+  echo "Error: unsupported action '$ACTION'"
+  exit 1
+fi
+
+case "$ROLE" in
+
+  master)
+    exec mooncake_master \
+      -v=1 \
+      --rpc_interface=eth0 \
+      --enable_http_metadata_server=true \
+      --http_metadata_server_host=0.0.0.0 \
+      --http_metadata_server_port=8080 \
+      --enable_metadata_cleanup_on_timeout=true \
+      --client_ttl=10
+    ;;
+
+  worker)
+    # Read the dynamic values from the RuntimeConfig JSON mounted by Fluid
+    if [ -z "$FLUID_RUNTIME_CONFIG_PATH" ] || [ ! -f "$FLUID_RUNTIME_CONFIG_PATH" ]; then
+      echo "Error: FLUID_RUNTIME_CONFIG_PATH not set or file not found"
+      exit 1
+    fi
+
+    CONFIG=$(cat "$FLUID_RUNTIME_CONFIG_PATH")
+
+    MASTER_SVC=$(echo "$CONFIG" | jq -r '.master.service.name')
+    WORKER_SVC=$(echo "$CONFIG" | jq -r '.worker.service.name')
+    QUOTA=$(echo "$CONFIG" | jq -r '.worker.tieredStoreLevels[0].quotas[0] // "1GiB"')
+
+    # Quota format conversion: Fluid hands over K8s-style "1Gi", Mooncake wants "1GB"
+    SEGMENT_SIZE=$(echo "$QUOTA" | sed 's/Gi$/GB/; s/Mi$/MB/')
+
+    NAMESPACE="${FLUID_DATASET_NAMESPACE:-default}"
+    MASTER_ADDR="${MASTER_SVC}.${NAMESPACE}.svc.cluster.local:50051"
+    METADATA_ADDR="http://${MASTER_SVC}.${NAMESPACE}.svc.cluster.local:8080/metadata"
+    WORKER_HOST="${POD_NAME}.${WORKER_SVC}.${NAMESPACE}.svc.cluster.local"
+
+    echo "Starting worker: master=$MASTER_ADDR, segment_size=$SEGMENT_SIZE, host=$WORKER_HOST"
+
+    exec mooncake_client \
+      --host="$WORKER_HOST" \
+      --port=50052 \
+      --global_segment_size="$SEGMENT_SIZE" \
+      --master_server_address="$MASTER_ADDR" \
+      --metadata_server="$METADATA_ADDR" \
+      --protocol=tcp \
+      --enable_http_server=true \
+      --http_port=9300
+    ;;
+
+  client)
+    echo "Error: client role not yet implemented for Mooncake"
+    exit 1
+    ;;
+
+  *)
+    echo "Error: unknown role '$ROLE'"
+    exit 1
+    ;;
+esac
diff --git a/test/gha-e2e/mooncake/image/reportSummary.sh b/test/gha-e2e/mooncake/image/reportSummary.sh
new file mode 100644
index 00000000000..46ba3d74db2
--- /dev/null
+++ b/test/gha-e2e/mooncake/image/reportSummary.sh
@@ -0,0 +1,47 @@
+#!/bin/bash
+set -euo pipefail
+
+RAW=$(curl -s http://localhost:9003/metrics/summary)
+
+if [ -z "$RAW" ]; then
+  echo "Error: empty response from metrics endpoint" >&2
+  exit 1
+fi
+
+# Extract the "Mem Storage: 0 B / 2.00 GB (0.0%)" part
+MEM_LINE=$(echo "$RAW" | grep -oE 'Mem Storage: [^|]+' || true)
+
+CACHED_RAW=$(echo "$MEM_LINE" | sed -E 's/Mem Storage: ([^/]+) \/.*/\1/' | xargs)
+CAPACITY_RAW=$(echo "$MEM_LINE" | sed -E 's/.*\/ ([^(]+) \(.*/\1/' | xargs)
+PERCENT_RAW=$(echo "$MEM_LINE" | grep -oE '\([0-9.]+%\)' | tr -d '()%')
+
+# Unit format conversion ("2.00 GB" -> "2.00GiB")
+normalize_unit() {
+  echo "$1" | sed -E 's/ ?GB$/GiB/; s/ ?MB$/MiB/; s/ ?B$/B/' | tr -d ' '
+}
+CACHED=$(normalize_unit "$CACHED_RAW")
+CACHE_CAPACITY=$(normalize_unit "$CAPACITY_RAW")
+
+# The number of keys is used as fileNum
+FILE_NUM=$(echo "$RAW" | grep -oE 'Keys: [0-9]+' | grep -oE '[0-9]+' || echo "0")
+
+# Approximate the hit ratio from the success rate of Get requests
+GET_STATS=$(echo "$RAW" | grep -oE 'Get=[0-9.]+/[0-9.]+' || echo "Get=0.00/0.00")
+GET_SUCCESS=$(echo "$GET_STATS" | cut -d= -f2 | cut -d/ -f1)
+GET_TOTAL=$(echo "$GET_STATS" | cut -d/ -f2)
+HIT_RATIO=$(awk -v s="$GET_SUCCESS" -v t="$GET_TOTAL" \
+  'BEGIN{ if (t>0) printf "%.0f", (s/t*100); else print "0" }')
+
+# Mooncake has no real underlying UFS total, so use the cache capacity instead
+UFS_TOTAL="$CACHE_CAPACITY"
+
+cat <<EOF
+{
+  "cached": "$CACHED",
+  "cachedPercentage": "$PERCENT_RAW",
+  "cacheCapacity": "$CACHE_CAPACITY",
+  "cacheHitRatio": "$HIT_RATIO",
+  "fileNum": "$FILE_NUM",
+  "ufsTotal": "$UFS_TOTAL"
+}
+EOF
diff --git a/test/gha-e2e/mooncake/rw_job.yaml b/test/gha-e2e/mooncake/rw_job.yaml
new file mode 100644
index 00000000000..b6d97d9f7ee
--- /dev/null
+++ b/test/gha-e2e/mooncake/rw_job.yaml
@@ -0,0 +1,86 @@
+# Verifies the data path: read and write against the master directly through
+# Mooncake's own Python client. Note that the pod declares no volumes /
+# volumeMounts — without a client component there is no FUSE mount point, so the
+# data path goes entirely through Mooncake's own client protocol, not the PVC.
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: mooncake-rw-job
+  namespace: default
+  labels:
+    app: mooncake-rw-job
+spec:
+  backoffLimit: 2
+  template:
+    metadata:
+      name: mooncake-rw-job
+      labels:
+        app: mooncake-rw-job
+    spec:
+      restartPolicy: Never
+      automountServiceAccountToken: false
+      containers:
+        - name: rw
+          image: fluidcloudnative/mooncake:e2e
+          imagePullPolicy: IfNotPresent
+          resources:
+            limits:
+              memory: "1Gi"
+          env:
+            # The client's local_hostname must be the pod's own IP so that other
+            # nodes can connect back to fetch data
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
+          # Override the image's default entrypoint so it is not started as a
+          # master or a worker
+          command: ["python3", "-c"]
+          args:
+            - |
+              import hashlib
+              import os
+              import sys
+
+              from mooncake.store import MooncakeDistributedStore
+
+              MASTER = "mooncake-demo-master-0.svc-mooncake-demo-master"
+
+              store = MooncakeDistributedStore()
+              rc = store.setup(
+                  local_hostname=os.environ["POD_IP"],
+                  metadata_server="http://%s:8080/metadata" % MASTER,
+                  master_server_addr="%s:50051" % MASTER,
+                  global_segment_size=0,
+                  local_buffer_size=128 * 1024 * 1024,
+                  protocol="tcp",
+                  rdma_devices="",
+              )
+              if rc != 0:
+                  sys.exit("setup failed, rc=%s" % rc)
+
+              # Mooncake's put silently skips a key that already exists (it does
+              # not overwrite the old value and still returns 0). With a fixed
+              # key, a job retry would read back the previous run's data and
+              # report an md5 mismatch, producing a false failure, so use a
+              # random key on every run.
+              key = "e2e-" + os.urandom(8).hex()
+              payload = os.urandom(4 * 1024 * 1024)
+
+              rc = store.put(key, payload)
+              if rc != 0:
+                  sys.exit("put failed, rc=%s" % rc)
+
+              got = store.get(key)
+              if got is None:
+                  sys.exit("get returned None for key %s" % key)
+              if len(got) != len(payload):
+                  sys.exit("length mismatch: got %d, want %d" % (len(got), len(payload)))
+
+              want = hashlib.md5(payload).hexdigest()
+              have = hashlib.md5(got).hexdigest()
+              print("key=%s put_md5=%s get_md5=%s" % (key, want, have))
+              if want != have:
+                  sys.exit("md5 mismatch: got %s, want %s" % (have, want))
+
+              print("mooncake put/get verified")
diff --git a/test/gha-e2e/mooncake/test.sh b/test/gha-e2e/mooncake/test.sh
new file mode 100644
index 00000000000..d235c3fafe2
--- /dev/null
+++ b/test/gha-e2e/mooncake/test.sh
@@ -0,0 +1,381 @@
+#!/bin/bash
+
+# Mooncake is a cache system without POSIX mount semantics; its
+# CacheRuntimeClass declares only master and worker, and no client. This case
+# guards exactly that scenario: when the client component is omitted the
+# controller must not panic and the Dataset must still reach Bound. The curvine
+# case ships a client component and does not cover this path.
+
+testname="mooncake cache runtime (client-less topology) e2e"
+
+dataset_name="mooncake-demo"
+rw_job_name="mooncake-rw-job"
+bad_mount_pod_name="mooncake-bad-mount"
+testdir="test/gha-e2e/mooncake"
+
+function syslog() {
+    echo ">>> $1"
+}
+
+function panic() {
+    local err_msg=$1
+    syslog "test \"$testname\" failed: $err_msg"
+    exit 1
+}
+
+function create_dataset() {
+    kubectl create -f $testdir/cacheruntimeclass.yaml
+    kubectl create -f $testdir/dataset.yaml
+    kubectl create -f $testdir/cacheruntime.yaml
+
+    if [[ -z "$(kubectl get cacheruntimeclass $dataset_name -oname)" ]]; then
+        panic "failed to create mooncake cache runtime class $dataset_name"
+    fi
+
+    if [[ -z "$(kubectl get dataset $dataset_name -oname)" ]]; then
+        panic "failed to create dataset $dataset_name"
+    fi
+
+    if [[ -z "$(kubectl get cacheruntime $dataset_name -oname)" ]]; then
+        panic "failed to create mooncake cache runtime $dataset_name"
+    fi
+}
+
+# Regression guard: a client-less topology used to make cacheruntime-controller
+# panic on a nil pointer. If it regresses the Dataset gets stuck in NotBound, but
+# the panic itself surfaces earlier and says more, so check for it while waiting
+# for Bound and report the root cause directly on failure.
+function check_controller_not_panicked() {
+    local logs=""
+    logs=$(kubectl logs -n fluid-system -l control-plane=cacheruntime-controller \
+        -c manager --tail=200 2>/dev/null || true)
+    if echo "$logs" | grep -qE "panic:|invalid memory address or nil pointer dereference"; then
+        syslog "--- cacheruntime-controller panic detected ---"
+        echo "$logs" | grep -A 20 -E "panic:|nil pointer dereference" || true
+        panic "cacheruntime-controller panicked on a client-less topology (regression)"
+    fi
+    syslog "No panic found in cacheruntime-controller logs"
+}
+
+function wait_dataset_bound() {
+    local deadline=600 # 10 minutes
+    local last_state=""
+    local counter=0
+    while true; do
+        last_state=$(kubectl get dataset $dataset_name -ojsonpath='{@.status.phase}')
+        if [[ "$last_state" == "Bound" ]]; then
+            break
+        fi
+
+        counter=$((counter + 1))
+        if [[ $((counter % 3)) -eq 0 ]]; then
+            syslog "checking dataset.status.phase==Bound (already $((counter * 5))s, last state: ${last_state:-<empty>})"
+            check_controller_not_panicked
+        fi
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            panic "timeout ${deadline}s waiting for dataset $dataset_name to be Bound (last state: ${last_state:-<empty>})"
+        fi
+        sleep 5
+    done
+    syslog "Found dataset $dataset_name status.phase==Bound"
+}
+
+function wait_cache_worker_ready() {
+    local deadline=180 # 3 minutes
+    local worker_component_name="${dataset_name}-worker"
+    local worker_selector="cacheruntime.fluid.io/component-name=${worker_component_name}"
+    local last_phase=""
+    local ready_replicas=""
+    local desired_replicas=""
+    local counter=0
+
+    while true; do
+        last_phase=$(kubectl get cacheruntime "$dataset_name" -ojsonpath='{@.status.worker.phase}')
+        ready_replicas=$(kubectl get cacheruntime "$dataset_name" -ojsonpath='{@.status.worker.readyReplicas}')
+        desired_replicas=$(kubectl get cacheruntime "$dataset_name" -ojsonpath='{@.status.worker.desiredReplicas}')
+
+        if [[ "$last_phase" == "Ready" ]] && \
+            [[ -n "$desired_replicas" ]] && \
+            [[ "$desired_replicas" != "0" ]] && \
+            [[ "$ready_replicas" == "$desired_replicas" ]] && \
+            kubectl wait --for=condition=Ready --timeout=5s pod -l "$worker_selector" >/dev/null 2>&1; then
+            break
+        fi
+
+        counter=$((counter + 1))
+        if [[ $((counter % 3)) -eq 0 ]]; then
+            syslog "checking cache worker readiness (already $((counter * 5))s, phase: ${last_phase:-<empty>}, ready/desired: ${ready_replicas:-<empty>}/${desired_replicas:-<empty>})"
+        fi
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            panic "timeout ${deadline}s waiting for cache worker pod ready"
+        fi
+        sleep 5
+    done
+    syslog "Found ready cache worker pod for $dataset_name"
+}
+
+# A client-less topology must produce no client-side artifacts. If the controller
+# ever starts creating a fallback DaemonSet for the missing client component,
+# this is what catches it first.
+function check_no_client_component() {
+    local client_component_name="${dataset_name}-client"
+    local client_selector="cacheruntime.fluid.io/component-name=${client_component_name}"
+
+    if kubectl get daemonset "$client_component_name" >/dev/null 2>&1; then
+        panic "unexpected client DaemonSet $client_component_name created for a client-less topology"
+    fi
+
+    local client_pods=""
+    client_pods=$(kubectl get pod -l "$client_selector" -oname 2>/dev/null)
+    if [[ -n "$client_pods" ]]; then
+        panic "unexpected client pods for a client-less topology: $client_pods"
+    fi
+
+    # Note: status.client itself does exist (as {"phase":""}) and spec.client is
+    # filled in by the CRD defaults; neither means a client component was
+    # actually started. The criterion is that phase stays empty.
+    local client_phase=""
+    client_phase=$(kubectl get cacheruntime "$dataset_name" -ojsonpath='{@.status.client.phase}' 2>/dev/null)
+    if [[ -n "$client_phase" ]]; then
+        panic "expected empty cacheruntime.status.client.phase for a client-less topology, got: $client_phase"
+    fi
+
+    syslog "Confirmed no client component was created"
+}
+
+# Evidence that the ReportSummary script ran: cacheStates gets populated.
+# Mooncake has no UFS, so its reportSummary.sh fills ufsTotal with the cache
+# capacity, which is why the two are expected to be equal.
+function check_dataset_cache_state() {
+    local deadline=180
+    local counter=0
+    local cache_capacity=""
+    while true; do
+        cache_capacity=$(kubectl get dataset ${dataset_name} -ojsonpath='{.status.cacheStates.cacheCapacity}' 2>/dev/null)
+        if [[ -n "$cache_capacity" ]] && [[ "$cache_capacity" != "0B" ]]; then
+            break
+        fi
+        counter=$((counter + 1))
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            panic "timeout ${deadline}s waiting for cacheStates.cacheCapacity to be reported (got: ${cache_capacity:-<empty>}), report summary may have failed"
+        fi
+        sleep 5
+    done
+
+    local ufs_total=""
+    ufs_total=$(kubectl get dataset ${dataset_name} -ojsonpath='{.status.cacheStates.ufsTotal}' 2>/dev/null)
+    if [[ "$ufs_total" != "$cache_capacity" ]]; then
+        panic "expected ufsTotal to equal cacheCapacity for a UFS-less cache system, got ufsTotal=${ufs_total:-<empty>} cacheCapacity=${cache_capacity}"
+    fi
+
+    syslog "Found reported cacheStates (cacheCapacity=$cache_capacity, ufsTotal=$ufs_total)"
+}
+
+function create_job() {
+    local job_file=$1
+    local job_name=$2
+    kubectl create -f "$job_file"
+
+    if [[ -z "$(kubectl get job "$job_name" -oname)" ]]; then
+        panic "failed to create job $job_name"
+    fi
+}
+
+function wait_job_completed() {
+    local job_name=$1
+    local succeed=""
+    local deadline=600
+    local counter=0
+    local job_failed=""
+    while true; do
+        succeed=$(kubectl get job "$job_name" -ojsonpath='{@.status.succeeded}')
+        [[ -z "$succeed" ]] && succeed=0
+
+        if [[ "$succeed" -ge "1" ]]; then
+            break
+        fi
+
+        # Only fail when the job controller itself marks the job as Failed
+        # (i.e. all backoffLimit retries are exhausted), not on first pod failure.
+        job_failed=$(kubectl get job "$job_name" \
+            -ojsonpath='{.status.conditions[?(@.type=="Failed")].status}' 2>/dev/null || true)
+        if [[ "$job_failed" == "True" ]]; then
+            syslog "--- logs of failed job $job_name ---"
+            kubectl logs job/"$job_name" --tail=100 2>&1 || true
+            panic "job $job_name failed when accessing data (all retries exhausted)"
+        fi
+
+        counter=$((counter + 1))
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            panic "timeout ${deadline}s waiting for job $job_name to complete"
+        fi
+        sleep 5
+    done
+    syslog "Found succeeded job $job_name"
+    kubectl logs job/"$job_name" --tail=20 2>&1 || true
+}
+
+# After a write, cached should reflect the actual usage. ReportSummary runs
+# periodically, so one round has to elapse first.
+function check_cached_after_write() {
+    local deadline=180
+    local counter=0
+    local cached=""
+    local file_num=""
+    while true; do
+        cached=$(kubectl get dataset ${dataset_name} -ojsonpath='{.status.cacheStates.cached}' 2>/dev/null)
+        file_num=$(kubectl get dataset ${dataset_name} -ojsonpath='{.status.cacheStates.fileNum}' 2>/dev/null)
+        if [[ -n "$cached" ]] && [[ "$cached" != "0B" ]] && [[ "$file_num" != "0" ]]; then
+            break
+        fi
+        counter=$((counter + 1))
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            panic "timeout ${deadline}s waiting for cacheStates to reflect written data (cached=${cached:-<empty>}, fileNum=${file_num:-<empty>})"
+        fi
+        sleep 5
+    done
+    syslog "Found cacheStates reflecting written data (cached=$cached, fileNum=$file_num)"
+}
+
+# Negative case: without a client component there is no FUSE mount point, so an
+# application pod mounting this PVC is bound to fail. The docs' FAQ promises this
+# behaviour (the PVC/PV do reach Bound, but cannot be mounted); pin it down here
+# so the docs do not quietly go stale if the CSI-side behaviour ever changes.
+function check_pvc_not_mountable() {
+    # First confirm the PVC/PV really are Bound — this is the misleading part
+    local pvc_phase=""
+    pvc_phase=$(kubectl get pvc $dataset_name -ojsonpath='{@.status.phase}' 2>/dev/null)
+    if [[ "$pvc_phase" != "Bound" ]]; then
+        panic "expected PVC $dataset_name to be Bound, got: ${pvc_phase:-<empty>}"
+    fi
+    syslog "PVC $dataset_name is Bound as documented (but must not be mountable)"
+
+    kubectl create -f $testdir/bad_mount_pod.yaml
+
+    local deadline=150
+    local counter=0
+    local failed_mount=""
+    local pod_phase=""
+    while true; do
+        failed_mount=$(kubectl get events \
+            --field-selector "involvedObject.name=${bad_mount_pod_name},reason=FailedMount" \
+            -ojsonpath='{.items[*].message}' 2>/dev/null)
+        if [[ -n "$failed_mount" ]]; then
+            break
+        fi
+
+        # If the pod ever becomes Running the mount actually succeeded, which is
+        # a behaviour change
+        pod_phase=$(kubectl get pod "$bad_mount_pod_name" -ojsonpath='{@.status.phase}' 2>/dev/null)
+        if [[ "$pod_phase" == "Running" ]]; then
+            panic "pod $bad_mount_pod_name mounted the PVC successfully; a client-less runtime is expected to have no FUSE mount point"
+        fi
+
+        counter=$((counter + 1))
+        if [[ $((counter % 6)) -eq 0 ]]; then
+            syslog "waiting for FailedMount event (already $((counter * 5))s, pod phase: ${pod_phase:-<empty>})"
+        fi
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            syslog "--- describe of $bad_mount_pod_name ---"
+            kubectl describe pod "$bad_mount_pod_name" 2>&1 || true
+            panic "timeout ${deadline}s waiting for a FailedMount event on $bad_mount_pod_name"
+        fi
+        sleep 5
+    done
+
+    syslog "Found expected FailedMount event: $failed_mount"
+
+    # The failure reason should point at the missing FUSE mount point, not some
+    # other mount problem
+    if ! echo "$failed_mount" | grep -qi "fuse mount point"; then
+        syslog "WARNING: FailedMount message does not mention the FUSE mount point; the docs' FAQ wording may need updating"
+    fi
+
+    kubectl delete --ignore-not-found -f $testdir/bad_mount_pod.yaml --force --grace-period=0 >/dev/null 2>&1
+    syslog "Confirmed the Dataset PVC cannot be mounted by application pods"
+}
+
+function delete_dataset_and_runtime() {
+    kubectl delete -f $testdir/dataset.yaml
+    kubectl delete -f $testdir/cacheruntime.yaml
+}
+
+function wait_runtime_deleted() {
+    local deadline=120
+    local counter=0
+    while true; do
+        local remaining=""
+        remaining=$(kubectl get advancedstatefulset,daemonset,svc -l fluid.io/managed-by=fluid -n default -oname 2>/dev/null)
+        if [[ -z "$remaining" ]]; then
+            break
+        fi
+        counter=$((counter + 1))
+        if [[ $((counter * 5)) -ge $deadline ]]; then
+            syslog "remaining resources after deletion: $remaining"
+            panic "timeout ${deadline}s waiting for runtime resources to be garbage collected"
+        fi
+        sleep 5
+    done
+    syslog "All runtime resources (AdvancedStatefulSet, Service) garbage collected"
+
+    if kubectl get pvc $dataset_name -n default >/dev/null 2>&1; then
+        panic "PVC $dataset_name still exists after deletion"
+    fi
+    if kubectl get pv default-$dataset_name >/dev/null 2>&1; then
+        panic "PV default-$dataset_name still exists after deletion"
+    fi
+    syslog "PV/PVC cleaned up successfully"
+}
+
+function dump_env_and_clean_up() {
+    local exit_code=$?
+    if [[ $exit_code -ne 0 ]]; then
+        syslog "=== Diagnostic logs for failed test ==="
+        syslog "--- cacheruntime-controller logs (last 100 lines) ---"
+        kubectl logs -n fluid-system -l control-plane=cacheruntime-controller -c manager --tail=100 2>&1 || true
+        syslog "--- CacheRuntime describe ---"
+        kubectl describe cacheruntime $dataset_name 2>&1 || true
+        syslog "--- Dataset describe ---"
+        kubectl describe dataset $dataset_name 2>&1 || true
+        syslog "--- Pods in default namespace ---"
+        kubectl get pods -n default -owide 2>&1 || true
+        syslog "--- bad-mount pod describe ---"
+        kubectl describe pod $bad_mount_pod_name 2>&1 || true
+        syslog "--- Job logs ---"
+        kubectl logs job/$rw_job_name --tail=100 2>&1 || true
+        syslog "--- Events in default namespace ---"
+        kubectl get events -n default --sort-by='.lastTimestamp' 2>&1 || true
+        syslog "=== End of diagnostic logs ==="
+    fi
+    syslog "Cleaning up resources for testcase $testname"
+    kubectl delete --ignore-not-found -f $testdir/bad_mount_pod.yaml --force --grace-period=0
+    kubectl delete --ignore-not-found -f $testdir/rw_job.yaml
+    kubectl delete --ignore-not-found -f $testdir/dataset.yaml
+    kubectl delete --ignore-not-found -f $testdir/cacheruntime.yaml
+    kubectl delete --ignore-not-found -f $testdir/cacheruntimeclass.yaml
+}
+
+function main() {
+    syslog "[TESTCASE $testname STARTS AT $(date)]"
+    trap dump_env_and_clean_up EXIT
+
+    create_dataset
+    wait_dataset_bound
+    check_controller_not_panicked
+    wait_cache_worker_ready
+    check_no_client_component
+    check_dataset_cache_state
+
+    create_job $testdir/rw_job.yaml $rw_job_name
+    wait_job_completed $rw_job_name
+    check_cached_after_write
+
+    check_pvc_not_mountable
+
+    delete_dataset_and_runtime
+    wait_runtime_deleted
+
+    syslog "[TESTCASE $testname SUCCEEDED AT $(date)]"
+}
+
+main