From 1613ac54914096e2d56f03352b8115718c32598a Mon Sep 17 00:00:00 2001
From: igavra <igavra@redhat.com>
Date: Wed, 26 Aug 2026 22:18:50 +0300
Subject: [PATCH] test(e2e): add k8s storage SP API, NATS, and cluster checks
 (FLPATH-4434)

Assisted-by: Cursor
Signed-off-by: igavra <igavra@redhat.com>
---
 .cursor/rules/e2e-tests.mdc               |  16 +-
 .gitignore                                |   1 +
 CLAUDE.md                                 |  12 +-
 Makefile                                  |  13 +-
 README.md                                 |  15 +-
 scripts/deploy-dcm.sh                     |  83 ++++
 tests/e2e/api_helpers_test.go             |   2 +
 tests/e2e/sp_storage_api_test.go          | 369 ++++++++++++++++++
 tests/e2e/sp_storage_helpers_test.go      | 455 ++++++++++++++++++++++
 tests/e2e/sp_storage_registration_test.go |  48 +++
 tests/e2e/sp_storage_status_test.go       | 263 +++++++++++++
 tests/run-e2e.sh                          |  46 ++-
 12 files changed, 1312 insertions(+), 11 deletions(-)
 create mode 100644 tests/e2e/sp_storage_api_test.go
 create mode 100644 tests/e2e/sp_storage_helpers_test.go
 create mode 100644 tests/e2e/sp_storage_registration_test.go
 create mode 100644 tests/e2e/sp_storage_status_test.go

diff --git a/.cursor/rules/e2e-tests.mdc b/.cursor/rules/e2e-tests.mdc
index 665c18d..4fafcb9 100644
--- a/.cursor/rules/e2e-tests.mdc
+++ b/.cursor/rules/e2e-tests.mdc
@@ -15,6 +15,7 @@ Files follow `{layer}_{subject}_test.go` with a matching `{layer}_helpers_test.g
 | `api_` | Control plane HTTP API (`doRequest` → `localhost:8080`) | `api_helpers_test.go` |
 | `cli_` | CLI binary (`runDCM` → `os/exec`) | `cli_helpers_test.go` |
 | `sp_container_` | Container SP direct API + NATS (`doContainerSPRequest` → `localhost:8082`) | `sp_helpers_test.go` |
+| `sp_storage_` | Storage SP direct API + NATS (`doStorageSPRequest` → `localhost:8089`) | `sp_storage_helpers_test.go` |
 | `sp_acm_cluster_` | ACM Cluster SP direct API (`doAcmClusterSPRequest` → `localhost:8083`) | `sp_acm_cluster_helpers_test.go` |
 | `core_` | Cross-service provisioning flow via control plane (`doRequest`) | `api_helpers_test.go`, `sp_helpers_test.go` |
 
@@ -23,9 +24,9 @@ When adding a new service provider, add `sp_{provider}_{concern}_test.go` (e.g.
 ## Go Conventions
 - All files start with `//go:build e2e` and `package e2e_test`
 - Dot-import Ginkgo and Gomega: `. "github.com/onsi/ginkgo/v2"`
-- Use Ginkgo labels for selective runs: `Label("sp", "container")`, `Label("sp", "acm-cluster")`, `Label("core", "platform")`, `Label("cli")`, `Label("smoke")`, `Label("cluster")`, `Label("disruptive")`
+- Use Ginkgo labels for selective runs: `Label("sp", "container")`, `Label("sp", "storage")`, `Label("sp", "storage", "registration")`, `Label("sp", "acm-cluster")`, `Label("core", "platform")`, `Label("cli")`, `Label("smoke")`, `Label("cluster")`, `Label("disruptive")`, `Label("nats")`
 - Ordered contexts need `AfterAll` cleanup; standalone tests use `BeforeEach`/`AfterEach`
-- Tests that need optional infrastructure call `requireCLI()`, `requireContainerSP()`, `requireAcmClusterSP()`, `requireKubectl()`, or `requirePodman()` to skip gracefully
+- Tests that need optional infrastructure call `requireCLI()`, `requireContainerSP()`, `requireStorageSP()`, `requireEnvironmentAgent()`, `requireAcmClusterSP()`, `requireKubectl()`, or `requirePodman()` to skip gracefully
 - `BeforeSuite` in `api_helpers_test.go` is the single initialization point — add `init*()` calls there for new subsystems
 - Tests that modify infrastructure (e.g. stop NATS) use `Label("disruptive")` so they can be excluded from normal runs
 - Tests that require cluster access use `Label("cluster")` so they can be filtered when no cluster is available
@@ -39,10 +40,20 @@ When adding a new service provider, add `sp_{provider}_{concern}_test.go` (e.g.
 |----------|---------|---------|
 | `DCM_GATEWAY_URL` | `http://localhost:8080/api/v1alpha1` | Control plane API |
 | `DCM_CONTAINER_SP_URL` | `http://localhost:8082/api/v1alpha1` | Container SP direct API |
+| `DCM_STORAGE_SP_URL` | `http://localhost:8089/api/v1alpha1` | Storage SP direct API |
 | `DCM_ACM_CLUSTER_SP_URL` | `http://localhost:8083/api/v1alpha1` | ACM Cluster SP direct API |
 | `DCM_NATS_URL` | `nats://localhost:4222` | NATS for status events |
 | `DCM_CLI_PATH` | (auto-resolved) | CLI binary path |
 | `DCM_CONTAINER_PROVIDER_NAME` | (first container provider) | Target a specific container provider in core platform tests |
+| `DCM_ENVIRONMENT_AGENT_URL` | (unset) | Environment agent for storage registration tests (`storage && registration`) |
+| `K8S_STORAGE_SP_NAMESPACE` | `default` | Namespace for storage SP PVCs |
+| `K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS` | (from `.dcm-e2e.env`) | E2E harness: SP fallback when requests omit `provider_hints` (`SP_K8S_DEFAULT_STORAGE_CLASS`) |
+| `E2E_CATALOG_STORAGE_CLASS` | `standard` | Optional override of simulated catalog hint `storage_class` in SP-direct E2E |
+| `K8S_STORAGE_SP_REGISTERED_ENDPOINT` | (compose internal URL) | Expected registered volumes URL in registration tests |
+
+Storage class sources in storage E2E (do not conflate):
+- **Catalog hints** (`provider_hints.kubernetes.storage_class`): `defaultCatalogStorageClass` constant (`standard`) via `catalogStorageClassHint()`; optional `E2E_CATALOG_STORAGE_CLASS` override
+- **SP default** (no hints, TC-2.1.6): from `K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS` via `spDefaultStorageClass()`; set by `deploy-dcm.sh` and `.dcm-e2e.env`
 
 ## Code Quality Patterns
 
@@ -103,4 +114,5 @@ When testing message ordering via NATS, comment whether the test relies on singl
 ## Compose Overrides
 SP compose overrides are managed by the provider registry (`providers/*.conf`). Each provider's `COMPOSE_OVERRIDE` key specifies its overlay file, which `deploy-dcm.sh` injects automatically when the provider is enabled:
 - `tests/compose-sp-test.yaml` — publishes container SP port 8082 (via `k8s-container.conf`)
+- `tests/compose-storage-sp-test.yaml` — publishes storage SP port 8089 (via `k8s-storage.conf`)
 - `tests/compose-acm-cluster-sp.yaml` — adds the ACM cluster SP service on port 8083 (via `acm-cluster.conf`)
diff --git a/.gitignore b/.gitignore
index e61836e..03d4d8c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -29,5 +29,6 @@ Thumbs.db
 # Env / secrets
 .env
 .env.*
+.dcm-e2e.env
 *.pem
 *.key
diff --git a/CLAUDE.md b/CLAUDE.md
index cfeeabf..03a058a 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -178,6 +178,8 @@ All test targets support JUnit XML output: `make test-e2e JUNIT_REPORT=results.x
 | **Core platform tests** | Full provisioning flow through control plane | `core`, `platform` |
 | **API tests** | HTTP CRUD operations against the control plane | (none) |
 | **SP tests** | Container SP direct API + NATS status events | `sp`, `container` |
+| **Storage SP tests** | K8s storage SP direct API + NATS status events | `sp`, `storage` |
+| **Storage SP registration** | Env-agent registration placeholder (skipped without env) | `sp`, `storage`, `registration` |
 | **ACM SP tests** | ACM Cluster SP API (health, registration, validation, CRUD) | `sp`, `acm-cluster` |
 | **Cluster tests** | Tests requiring `kubectl`/`oc` cluster access | `cluster` |
 | **Disruptive tests** | Tests that stop/start infrastructure (e.g. NATS) | `disruptive` |
@@ -204,12 +206,18 @@ CLI tests are skipped (not failed) if no binary is available.
 - `DCM_GATEWAY_URL` env var overrides the control plane API endpoint (default: `http://localhost:8080/api/v1alpha1`)
 - `DCM_CONTAINER_SP_URL` env var overrides the container SP endpoint (default: `http://localhost:8082/api/v1alpha1`)
 - `DCM_STORAGE_SP_URL` env var overrides the storage SP endpoint (default: `http://localhost:8089/api/v1alpha1`)
+- `K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS` — **E2E harness only:** SP fallback when `provider_hints` omit `storage_class`; auto-detected by `deploy-dcm.sh`, written to `.dcm-e2e.env` (loaded by `make test-storage-sp`, `make test-sp`, and `run-e2e.sh` when storage tests run)
+- `K8S_STORAGE_SP_DEFAULT_ACCESS_MODE` — **E2E harness only:** SP fallback when hints omit `access_mode` (compose default `ReadWriteOnce`; not cluster-detected)
+- Catalog `provider_hints.kubernetes.storage_class` in storage E2E defaults to `standard` (simulated catalog hint for SP-direct tests; override with `E2E_CATALOG_STORAGE_CLASS`)
+- `K8S_STORAGE_SP_NAMESPACE` — PVC namespace for storage SP tests (from deploy `--k8s-storage-namespace`, default `default`)
+- `DCM_ENVIRONMENT_AGENT_URL` enables storage registration placeholder tests (`registration` label filter)
+- `K8S_STORAGE_SP_REGISTERED_ENDPOINT` overrides the expected registered volumes URL in registration tests (default: compose internal `http://k8s-storage-service-provider:8080/api/v1alpha1/volumes`)
 - `DCM_ACM_CLUSTER_SP_URL` env var overrides the ACM cluster SP endpoint (default: `http://localhost:8083/api/v1alpha1`)
 - `DCM_NATS_URL` env var overrides the NATS server (default: `nats://localhost:4222`)
 - `DCM_CLI_PATH` env var specifies the CLI binary path
 - `DCM_CONTAINER_PROVIDER_NAME` env var overrides which container provider to target in core platform tests (default: first `service_type=container` provider found)
-- Ginkgo labels (`smoke`, `cli`, `sp`, `container`, `acm-cluster`, `nats`, `cluster`, `disruptive`, `core`, `platform`, `rehydration`, `happy-path`, `failover`, `policy`, `negative`, `integrity`, `contract`) enable selective test runs via `--label-filter`
-- SP tests skip gracefully if the container SP or ACM cluster SP isn't reachable (no hard failure)
+- Ginkgo labels (`smoke`, `cli`, `sp`, `container`, `storage`, `registration`, `acm-cluster`, `nats`, `cluster`, `disruptive`, `core`, `platform`, `rehydration`, `happy-path`, `failover`, `policy`, `negative`, `integrity`, `contract`) enable selective test runs via `--label-filter`
+- SP tests skip gracefully if the container SP, storage SP, or ACM cluster SP isn't reachable (no hard failure); storage registration tests skip when `DCM_ENVIRONMENT_AGENT_URL` is unset
 - Cluster tests skip gracefully if `kubectl`/`oc` is unavailable or the cluster is unreachable
 - Disruptive tests skip if `podman` is unavailable; exclude from normal runs with `--label-filter '!disruptive'`
 
diff --git a/Makefile b/Makefile
index 5643642..765c520 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: help e2e-up test-e2e test-smoke test-cli test-sp test-acm-sp test-kubevirt-sp test-core test-rehydration test-rehydration-safe test-rehydration-cli e2e-down test-e2e-full download-cli cli-version lint
+.PHONY: help e2e-up test-e2e test-smoke test-cli test-sp test-storage-sp test-storage-sp-registration test-acm-sp test-kubevirt-sp test-core test-rehydration test-rehydration-safe test-rehydration-cli e2e-down test-e2e-full download-cli cli-version lint
 
 # Set JUNIT_REPORT to a filename to produce JUnit XML output.
 # Example: make test-e2e JUNIT_REPORT=results.xml
@@ -9,6 +9,9 @@ ifdef JUNIT_REPORT
 GINKGO_BASE += --junit-report=$(JUNIT_REPORT)
 endif
 
+# Source storage SP deploy exports when present (storage E2E targets only).
+STORAGE_E2E_ENV_LOAD = set -a; [ -f .dcm-e2e.env ] && . ./.dcm-e2e.env; set +a;
+
 help: ## Show all available targets
 	@grep -hE '^[a-zA-Z0-9_-]+:.*## ' $(MAKEFILE_LIST) | awk -F ':.*## ' '{printf "  %-18s %s\n", $$1, $$2}'
 
@@ -25,7 +28,13 @@ test-cli: ## Run CLI tests only (stack must be running)
 	cd tests/e2e && $(GINKGO_BASE) --label-filter=cli .
 
 test-sp: ## Run all service provider tests (SPs must be deployed with ports published)
-	cd tests/e2e && $(GINKGO_BASE) --label-filter=sp .
+	@$(STORAGE_E2E_ENV_LOAD) cd tests/e2e && $(GINKGO_BASE) --label-filter=sp .
+
+test-storage-sp: ## Run k8s storage SP API + NATS tests (excludes registration placeholder)
+	@$(STORAGE_E2E_ENV_LOAD) cd tests/e2e && $(GINKGO_BASE) --label-filter='storage && !registration' .
+
+test-storage-sp-registration: ## Run k8s storage SP env-agent registration placeholder only
+	@$(STORAGE_E2E_ENV_LOAD) cd tests/e2e && $(GINKGO_BASE) --label-filter='storage && registration' .
 
 test-acm-sp: ## Run ACM cluster SP tests only
 	cd tests/e2e && $(GINKGO_BASE) --label-filter=acm-cluster .
diff --git a/README.md b/README.md
index d772848..f932369 100644
--- a/README.md
+++ b/README.md
@@ -101,8 +101,10 @@ make e2e-up        # Deploy the stack
 make test-e2e      # Run all tests (stack must be running)
 make test-smoke    # Run health checks + CLI version only
 make test-cli      # Run CLI tests only
-make test-sp       # Run container SP tests (SP must be deployed)
-make test-acm-sp   # Run ACM cluster SP tests (ACM SP must be deployed)
+make test-sp          # Run all SP tests (SPs must be deployed)
+make test-storage-sp            # Storage SP API + NATS (default; excludes registration placeholder)
+make test-storage-sp-registration # Env-agent registration placeholder (needs DCM_ENVIRONMENT_AGENT_URL)
+make test-acm-sp      # Run ACM cluster SP tests (ACM SP must be deployed)
 make test-core     # Run core platform tests (full provisioning flow)
 make e2e-down      # Tear down
 make download-cli  # Download latest DCM CLI without running tests
@@ -127,9 +129,15 @@ make help
 | `DCM_STORAGE_SP_URL` | `http://localhost:8089/api/v1alpha1` | Storage SP direct URL (requires published port) |
 | `DCM_ACM_CLUSTER_SP_URL` | `http://localhost:8083/api/v1alpha1` | ACM Cluster SP direct URL (requires published port) |
 | `DCM_NATS_URL` | `nats://localhost:4222` | NATS server URL for status event tests |
+| `K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS` | (auto-detected at deploy) | **E2E harness only:** sets `SP_K8S_DEFAULT_STORAGE_CLASS` on the storage SP for requests that omit `provider_hints` (TC-2.1.6). Not used for catalog hints. Written to `.dcm-e2e.env` so `make test-storage-sp` sees the same SP config after deploy exits. |
+| `K8S_STORAGE_SP_DEFAULT_ACCESS_MODE` | `ReadWriteOnce` | **E2E harness only:** sets `SP_K8S_DEFAULT_ACCESS_MODE` when requests omit access_mode hints (compose default; not auto-detected). |
+| `E2E_CATALOG_STORAGE_CLASS` | `standard` | **Tests only:** optional override of simulated catalog `provider_hints.kubernetes.storage_class` in SP-direct E2E (default `standard`, common on kind). Deploy does not set this. |
+| `K8S_STORAGE_SP_NAMESPACE` | `default` | Namespace for storage PVCs (from deploy `--k8s-storage-namespace`) |
 | `DCM_CLI_PATH` | (auto-resolved) | Path to `dcm` CLI binary |
 | `JUNIT_REPORT` | (none) | JUnit XML report filename (e.g. `make test-e2e JUNIT_REPORT=results.xml`) |
 
+When the k8s storage SP is deployed for E2E, `deploy-dcm.sh` writes `.dcm-e2e.env` (gitignored) with the detected **SP default** storage class (no-hints path) and SP URLs. `make test-storage-sp`, `make test-storage-sp-registration`, and `make test-sp` source it; other targets do not. Catalog hint tests use `defaultCatalogStorageClass` (`standard`), not deploy env.
+
 ### Test Harness Flags
 
 The test harness (`tests/run-e2e.sh`) supports additional flags for fine-grained control:
@@ -146,7 +154,8 @@ The test harness (`tests/run-e2e.sh`) supports additional flags for fine-grained
 # Service provider tests
 ./tests/run-e2e.sh --k8s-container-service-provider --cluster-api https://api.example.com:6443
 ./tests/run-e2e.sh --k8s-storage-service-provider --kubeconfig ~/.kube/config
-./tests/run-e2e.sh --skip-deploy --label-filter "sp && container"
+./tests/run-e2e.sh --skip-deploy --label-filter "storage && !registration"
+./tests/run-e2e.sh --skip-deploy --label-filter "storage && registration"   # requires DCM_ENVIRONMENT_AGENT_URL
 
 # ACM cluster SP tests
 ./tests/run-e2e.sh --acm-cluster-service-provider --kubeconfig ~/.kube/config
diff --git a/scripts/deploy-dcm.sh b/scripts/deploy-dcm.sh
index fc943c1..e4c5260 100755
--- a/scripts/deploy-dcm.sh
+++ b/scripts/deploy-dcm.sh
@@ -36,6 +36,7 @@ readonly VERSION_ENV_VARS=(
 )
 
 REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+readonly E2E_ENV_FILE="${REPO_ROOT}/.dcm-e2e.env"
 
 # --- Provider registry ----------------------------------------------------- #
 #
@@ -290,9 +291,47 @@ tear_down() {
         rm -rf "${deploy_dir}"
     fi
 
+    rm -f "${E2E_ENV_FILE}"
     log "Tear-down complete"
 }
 
+# Persist storage SP deploy exports for storage E2E (child processes cannot inherit deploy-dcm.sh exports).
+write_e2e_env_file() {
+    local tmp="${E2E_ENV_FILE}.tmp.$$"
+    local wrote=false
+
+    {
+        echo "# Generated by deploy-dcm.sh — do not commit"
+        echo "# Load: set -a && . ${E2E_ENV_FILE} && set +a"
+        if [[ -n "${K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS:-}" ]]; then
+            printf 'export K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS=%q\n' "${K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS}"
+            wrote=true
+        fi
+        if [[ -n "${K8S_STORAGE_SP_NAMESPACE:-}" ]]; then
+            printf 'export K8S_STORAGE_SP_NAMESPACE=%q\n' "${K8S_STORAGE_SP_NAMESPACE}"
+            wrote=true
+        fi
+        local i
+        for i in $(seq 0 $((PROV_COUNT - 1))); do
+            [[ "${PROV_ENABLED[$i]}" == true ]] || continue
+            case "${PROV_FLAGS[$i]}" in
+                k8s-storage-service-provider)
+                    printf 'export DCM_STORAGE_SP_URL=%q\n' "http://localhost:8089/api/v1alpha1"
+                    printf 'export DCM_NATS_URL=%q\n' "nats://localhost:4222"
+                    wrote=true
+                    ;;
+            esac
+        done
+    } > "${tmp}"
+
+    if [[ "${wrote}" == true ]]; then
+        mv -f "${tmp}" "${E2E_ENV_FILE}"
+        info "Wrote ${E2E_ENV_FILE} for storage E2E (make test-storage-sp / test-sp load it)"
+    else
+        rm -f "${tmp}" "${E2E_ENV_FILE}"
+    fi
+}
+
 # --- Provider validation hooks -------------------------------------------- #
 #
 # Each hook receives: (kubeconfig, namespace, cli_binary)
@@ -340,6 +379,49 @@ validate_k8s_container_provider() {
 validate_k8s_storage_provider() {
     log "Validating k8s storage provider prerequisites"
     ensure_provider_namespace "$1" "$2" "$3"
+    ensure_storage_class_for_e2e "$1" "$3"
+}
+
+# ensure_storage_class_for_e2e configures SP_K8S_DEFAULT_STORAGE_CLASS for the utilities E2E stack only.
+# Production/catalog flows set provider_hints.kubernetes.storage_class via CatalogItem + SpecBuilder;
+# this hook covers the no-hints path (e.g. TC-2.1.6) where the SP falls back to SP_K8S_DEFAULT_STORAGE_CLASS.
+# Access mode is not auto-detected here — compose already sets K8S_STORAGE_SP_DEFAULT_ACCESS_MODE (ReadWriteOnce).
+ensure_storage_class_for_e2e() {
+    local kubeconfig="$1"
+    local cli="$2"
+    local selected_sc binding_mode
+
+    if [[ -n "${K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS:-}" ]]; then
+        selected_sc="${K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS}"
+        info "Using configured storage class: ${selected_sc}"
+    else
+        selected_sc=$("${cli}" --kubeconfig="${kubeconfig}" get sc \
+            -o jsonpath='{.items[?(@.metadata.annotations.storageclass\.kubernetes\.io/is-default-class=="true")].metadata.name}' 2>/dev/null | awk '{print $1}')
+        if [[ -z "${selected_sc}" ]]; then
+            selected_sc=$("${cli}" --kubeconfig="${kubeconfig}" get sc -o jsonpath='{.items[0].metadata.name}' 2>/dev/null | awk '{print $1}')
+        fi
+        if [[ -z "${selected_sc}" ]]; then
+            err "No StorageClasses found on the cluster"
+            err "Install a default StorageClass or set K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS before deploy"
+            return 1
+        fi
+    fi
+
+    if ! binding_mode=$("${cli}" --kubeconfig="${kubeconfig}" get sc "${selected_sc}" \
+            -o jsonpath='{.volumeBindingMode}' 2>/dev/null); then
+        err "StorageClass '${selected_sc}' not found"
+        if [[ -n "${K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS:-}" ]]; then
+            err "Unset K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS or choose an existing StorageClass"
+        fi
+        return 1
+    fi
+
+    export K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS="${selected_sc}"
+    if [[ "${binding_mode}" == "WaitForFirstConsumer" ]]; then
+        info "Storage class '${selected_sc}' uses WaitForFirstConsumer — E2E tests schedule PVC consumer pods"
+    else
+        info "E2E storage class: ${selected_sc} (${binding_mode:-Immediate})"
+    fi
 }
 
 validate_acm_cluster_provider() {
@@ -1102,6 +1184,7 @@ get_running_versions "${CONTROL_PLANE_TMP_DIR}/deploy/compose.yaml" ${COMPOSE_EX
 
 GATEWAY_URL="http://localhost:${CONTROL_PLANE_PORT}"
 log "DCM stack is up and healthy at ${GATEWAY_URL}"
+write_e2e_env_file
 if [[ "${CONTROL_PLANE_TMP_DIR}" != "${DEFAULT_CONTROL_PLANE_TMP_DIR}" ]]; then
     info "To tear down: $(basename "$0") --control-plane-dir ${CONTROL_PLANE_TMP_DIR} --tear-down"
 else
diff --git a/tests/e2e/api_helpers_test.go b/tests/e2e/api_helpers_test.go
index f849675..7d8d1ce 100644
--- a/tests/e2e/api_helpers_test.go
+++ b/tests/e2e/api_helpers_test.go
@@ -56,6 +56,8 @@ var _ = BeforeSuite(func() {
 
 	// Probe service providers (tests skip gracefully if not deployed).
 	initContainerSP()
+	initStorageSP()
+	initEnvironmentAgent()
 	initAcmClusterSP()
 
 	// Resolve cluster CLI for tests that need kubectl/oc.
diff --git a/tests/e2e/sp_storage_api_test.go b/tests/e2e/sp_storage_api_test.go
new file mode 100644
index 0000000..3d8f34d
--- /dev/null
+++ b/tests/e2e/sp_storage_api_test.go
@@ -0,0 +1,369 @@
+//go:build e2e
+
+package e2e_test
+
+import (
+	"net/http"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Storage SP API", Label("sp", "storage"), func() {
+	BeforeEach(func() {
+		requireStorageSP()
+	})
+
+	Context("health", func() {
+		It("returns healthy status", func() {
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes/health", "")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+			var body map[string]interface{}
+			decodeJSON(resp, &body)
+			Expect(body["status"]).To(Equal("healthy"))
+			Expect(body).To(HaveKey("uptime"))
+			Expect(body).To(HaveKey("version"))
+			Expect(body).To(HaveKey("type"))
+		})
+	})
+
+	Context("CRUD lifecycle", Ordered, func() {
+		var volumeIDs []string
+
+		AfterAll(func() {
+			for _, id := range volumeIDs {
+				deleteTestVolume(id)
+			}
+		})
+
+		It("creates a volume with valid spec", func() {
+			name := uniqueName("e2e-vol")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+
+			id := body["id"].(string)
+			Expect(id).To(Equal(name))
+			volumeIDs = append(volumeIDs, id)
+		})
+
+		It("creates a volume with a client-assigned id query parameter", func() {
+			customID := uniqueName("e2e-custom")
+			placeholderName := uniqueName("e2e-placeholder")
+			resp, err := doStorageSPRequest(http.MethodPost,
+				"/volumes?id="+customID,
+				volumeSpec(placeholderName, defaultTestCapacity))
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusCreated))
+
+			var body map[string]interface{}
+			decodeJSON(resp, &body)
+			Expect(body["id"]).To(Equal(customID))
+
+			spec, ok := body["spec"].(map[string]interface{})
+			Expect(ok).To(BeTrue())
+			meta, ok := spec["metadata"].(map[string]interface{})
+			Expect(ok).To(BeTrue())
+			Expect(meta["name"]).To(Equal(customID),
+				"SP should canonicalize metadata.name to the id query parameter")
+
+			volumeIDs = append(volumeIDs, customID)
+		})
+
+		It("derives the id from metadata.name when no query parameter is provided", func() {
+			name := uniqueName("e2e-autoid")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+
+			id := body["id"].(string)
+			Expect(id).To(Equal(name))
+			volumeIDs = append(volumeIDs, id)
+		})
+
+		It("gets an existing volume by ID", Label("cluster"), func() {
+			Expect(volumeIDs).NotTo(BeEmpty())
+			id := volumeIDs[0]
+
+			waitForStorageVolumeRunning(id, 120*time.Second)
+
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes/"+id, "")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+			var body map[string]interface{}
+			decodeJSON(resp, &body)
+			Expect(body["status"]).To(Equal("RUNNING"))
+		})
+
+		It("lists all managed volumes", func() {
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes", "")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+			var body map[string]interface{}
+			decodeJSON(resp, &body)
+			Expect(body).To(HaveKey("volumes"))
+
+			volumes, ok := body["volumes"].([]interface{})
+			Expect(ok).To(BeTrue())
+			Expect(len(volumes)).To(BeNumerically(">=", len(volumeIDs)),
+				"list should include at least all volumes we created")
+		})
+
+		It("paginates volume list", func() {
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes?max_page_size=2", "")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+			var page1 map[string]interface{}
+			decodeJSON(resp, &page1)
+			volumes := page1["volumes"].([]interface{})
+			Expect(len(volumes)).To(BeNumerically("<=", 2))
+
+			if token, ok := page1["next_page_token"].(string); ok && token != "" {
+				resp2, err := doStorageSPRequest(http.MethodGet,
+					"/volumes?max_page_size=2&page_token="+token, "")
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resp2.StatusCode).To(Equal(http.StatusOK))
+
+				var page2 map[string]interface{}
+				decodeJSON(resp2, &page2)
+				Expect(page2).To(HaveKey("volumes"))
+			}
+		})
+	})
+
+	Context("provider hints and PVC verification", Label("cluster"), func() {
+		var volumeID string
+
+		BeforeEach(func() {
+			requireClusterStoragePrereqs()
+			volumeID = ""
+		})
+
+		AfterEach(func() {
+			if volumeID != "" {
+				deleteTestVolume(volumeID)
+			}
+		})
+
+		It("creates a PVC named after the client-assigned id (TC-2.1.1)", func() {
+			customID := uniqueName("e2e-pvc-id")
+			placeholderName := uniqueName("e2e-placeholder")
+			resp, err := doStorageSPRequest(http.MethodPost,
+				"/volumes?id="+customID,
+				volumeSpecMinimal(placeholderName, defaultTestCapacity))
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusCreated))
+			volumeID = customID
+
+			pvc := getStoragePVCJSON(customID)
+			meta, _ := pvc["metadata"].(map[string]interface{})
+			Expect(meta["name"]).To(Equal(customID))
+		})
+
+		It("creates a PVC with DCM labels and storage class hint (TC-2.1.2)", func() {
+			name := uniqueName("e2e-hints")
+			body := createTestVolume(volumeSpecWithHints(name, defaultTestCapacity, catalogStorageClassHint(), "", ""))
+			volumeID = body["id"].(string)
+
+			pvc := getStoragePVCJSON(volumeID)
+			meta, _ := pvc["metadata"].(map[string]interface{})
+			labels, _ := meta["labels"].(map[string]interface{})
+			Expect(labels).To(HaveKeyWithValue("dcm.project/managed-by", "dcm"))
+			Expect(labels).To(HaveKeyWithValue("dcm.project/dcm-instance-id", volumeID))
+			Expect(labels).To(HaveKeyWithValue("dcm.project/dcm-service-type", "storage"))
+
+			spec, _ := pvc["spec"].(map[string]interface{})
+			Expect(spec["storageClassName"]).To(Equal(catalogStorageClassHint()))
+		})
+
+		It("applies access mode and volume mode hints (TC-2.1.3, TC-2.1.4)", func() {
+			name := uniqueName("e2e-modes")
+			body := createTestVolume(volumeSpecWithHints(name, defaultTestCapacity, catalogStorageClassHint(), "ReadWriteOnce", "Filesystem"))
+			volumeID = body["id"].(string)
+
+			pvc := getStoragePVCJSON(volumeID)
+			spec, _ := pvc["spec"].(map[string]interface{})
+			Expect(spec["accessModes"]).To(Equal([]interface{}{"ReadWriteOnce"}))
+			Expect(spec["volumeMode"]).To(Equal("Filesystem"))
+		})
+
+		It("applies all kubernetes provider hints together (TC-2.1.5)", func() {
+			name := uniqueName("e2e-all-hints")
+			body := createTestVolume(volumeSpecWithHints(name, defaultTestCapacity, catalogStorageClassHint(), "ReadWriteOnce", "Filesystem"))
+			volumeID = body["id"].(string)
+
+			pvc := getStoragePVCJSON(volumeID)
+			meta, _ := pvc["metadata"].(map[string]interface{})
+			labels, _ := meta["labels"].(map[string]interface{})
+			Expect(labels).To(HaveKeyWithValue("dcm.project/dcm-service-type", "storage"))
+
+			spec, _ := pvc["spec"].(map[string]interface{})
+			Expect(spec["storageClassName"]).To(Equal(catalogStorageClassHint()))
+			Expect(spec["accessModes"]).To(Equal([]interface{}{"ReadWriteOnce"}))
+			Expect(spec["volumeMode"]).To(Equal("Filesystem"))
+		})
+
+		It("uses SP defaults when provider hints are omitted (TC-2.1.6)", func() {
+			name := uniqueName("e2e-defaults")
+			body := createTestVolume(volumeSpecMinimal(name, defaultTestCapacity))
+			volumeID = body["id"].(string)
+
+			pvc := getStoragePVCJSON(volumeID)
+			spec, _ := pvc["spec"].(map[string]interface{})
+			Expect(spec["accessModes"]).To(Equal([]interface{}{"ReadWriteOnce"}))
+			Expect(spec["volumeMode"]).To(Equal("Filesystem"))
+			Expect(spec["storageClassName"]).To(Equal(spDefaultStorageClass()))
+		})
+	})
+
+	Context("validation errors", func() {
+		It("rejects an empty body", func() {
+			resp, err := doStorageSPRequest(http.MethodPost, "/volumes", "")
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusBadRequest))
+		})
+
+		It("rejects a body with missing required fields", func() {
+			resp, err := doStorageSPRequest(http.MethodPost, "/volumes", `{"spec":{}}`)
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusBadRequest))
+		})
+
+		It("rejects invalid field types", func() {
+			resp, err := doStorageSPRequest(http.MethodPost, "/volumes", `{"spec": {"service_type": 12345}}`)
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusBadRequest))
+		})
+
+		It("rejects the reserved volume id health", func() {
+			resp, err := doStorageSPRequest(http.MethodPost, "/volumes?id=health", volumeSpec("health", defaultTestCapacity))
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusBadRequest))
+		})
+
+		It("rejects a non-existent StorageClass", func() {
+			name := uniqueName("e2e-badsc")
+			resp, err := doStorageSPRequest(http.MethodPost, "/volumes",
+				volumeSpecWith(name, defaultTestCapacity, "does-not-exist-sc"))
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusUnprocessableEntity))
+		})
+
+		It("rejects wrong content type", func() {
+			url := storageSPBaseURL + "/volumes"
+			req, err := http.NewRequest(http.MethodPost, url, nil)
+			Expect(err).NotTo(HaveOccurred())
+			req.Header.Set("Content-Type", "text/plain")
+
+			resp, err := httpClient.Do(req)
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusBadRequest))
+		})
+
+		It("returns 409 when creating the same volume twice", func() {
+			name := uniqueName("e2e-dup")
+			createTestVolume(volumeSpec(name, defaultTestCapacity))
+			defer deleteTestVolume(name)
+
+			resp, err := doStorageSPRequest(http.MethodPost, "/volumes", volumeSpec(name, defaultTestCapacity))
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusConflict))
+		})
+	})
+
+	Context("not found", func() {
+		It("returns 404 for GET on non-existent volume", func() {
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes/does-not-exist", "")
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusNotFound))
+		})
+
+		It("returns 404 for DELETE on non-existent volume", func() {
+			resp, err := doStorageSPRequest(http.MethodDelete, "/volumes/does-not-exist", "")
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusNotFound))
+		})
+	})
+
+	Context("label filtering", Label("cluster"), Ordered, func() {
+		manualPVCName := uniqueName("e2e-manual")
+
+		BeforeAll(func() {
+			requireKubectl()
+		})
+
+		AfterAll(func() {
+			_, _ = runStorageKubectl("delete", "pvc", manualPVCName, "--ignore-not-found")
+		})
+
+		It("excludes non-DCM PVCs from list", func() {
+			By("creating a PVC without DCM labels via kubectl")
+			manifest := `apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: ` + manualPVCName + `
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: ` + catalogStorageClassHint() + `
+  resources:
+    requests:
+      storage: 1Gi
+`
+			err := applyStorageManifest(manifest)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("listing volumes via the SP API")
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes", "")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+			var body map[string]interface{}
+			decodeJSON(resp, &body)
+			volumes, _ := body["volumes"].([]interface{})
+
+			for _, v := range volumes {
+				vol := v.(map[string]interface{})
+				id, _ := vol["id"].(string)
+				Expect(id).NotTo(Equal(manualPVCName),
+					"non-DCM PVC %q should not appear in SP list", manualPVCName)
+			}
+		})
+	})
+
+	Context("delete lifecycle", Ordered, func() {
+		var deleteID string
+
+		It("creates a volume to delete", func() {
+			name := uniqueName("e2e-delete")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+			deleteID = body["id"].(string)
+		})
+
+		It("deletes the volume and its PVC", func() {
+			Expect(deleteID).NotTo(BeEmpty())
+
+			resp, err := doStorageSPRequest(http.MethodDelete, "/volumes/"+deleteID, "")
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusNoContent))
+
+			By("confirming the volume is gone")
+			resp, err = doStorageSPRequest(http.MethodGet, "/volumes/"+deleteID, "")
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusNotFound))
+		})
+	})
+})
diff --git a/tests/e2e/sp_storage_helpers_test.go b/tests/e2e/sp_storage_helpers_test.go
new file mode 100644
index 0000000..206bdd9
--- /dev/null
+++ b/tests/e2e/sp_storage_helpers_test.go
@@ -0,0 +1,455 @@
+//go:build e2e
+
+package e2e_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+const (
+	defaultStorageSPURL              = "http://localhost:8089/api/v1alpha1"
+	defaultStorageRegisteredEndpoint = "http://k8s-storage-service-provider:8080/api/v1alpha1/volumes"
+	natsStorageSubject               = "dcm.storage"
+	defaultTestCapacity = "1Gi"
+	// defaultCatalogStorageClass simulates catalog provider_hints.kubernetes.storage_class in SP-direct E2E.
+	// "standard" matches common kind/minikube default StorageClass names; override with E2E_CATALOG_STORAGE_CLASS.
+	defaultCatalogStorageClass = "standard"
+)
+
+var (
+	storageSPBaseURL            string
+	storageSPReady              bool
+	storageSPNamespace          string
+	storageSPRegisteredEndpoint string
+	environmentAgentBaseURL     string
+	environmentAgentReady       bool
+)
+
+func initStorageSP() {
+	storageSPBaseURL = os.Getenv("DCM_STORAGE_SP_URL")
+	if storageSPBaseURL == "" {
+		storageSPBaseURL = defaultStorageSPURL
+	}
+	storageSPBaseURL = strings.TrimRight(storageSPBaseURL, "/")
+
+	storageSPNamespace = os.Getenv("K8S_STORAGE_SP_NAMESPACE")
+	if storageSPNamespace == "" {
+		storageSPNamespace = "default"
+	}
+
+	resp, err := httpClient.Get(storageSPBaseURL + "/volumes/health")
+	if err != nil {
+		GinkgoWriter.Printf("Storage SP not reachable at %s: %v — storage SP tests will be skipped\n", storageSPBaseURL, err)
+		return
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		GinkgoWriter.Printf("Storage SP health returned %d — storage SP tests will be skipped\n", resp.StatusCode)
+		return
+	}
+	storageSPReady = true
+	GinkgoWriter.Printf(
+		"Storage SP ready at %s (namespace: %s, catalog hint storage_class: %q, SP default: %q)\n",
+		storageSPBaseURL, storageSPNamespace, catalogStorageClassHint(), spDefaultStorageClass(),
+	)
+
+	storageSPRegisteredEndpoint = os.Getenv("K8S_STORAGE_SP_REGISTERED_ENDPOINT")
+	if storageSPRegisteredEndpoint == "" {
+		storageSPRegisteredEndpoint = defaultStorageRegisteredEndpoint
+	}
+}
+
+func initEnvironmentAgent() {
+	environmentAgentBaseURL = os.Getenv("DCM_ENVIRONMENT_AGENT_URL")
+	if environmentAgentBaseURL == "" {
+		GinkgoWriter.Println("DCM_ENVIRONMENT_AGENT_URL unset — storage registration tests will be skipped")
+		return
+	}
+	environmentAgentBaseURL = strings.TrimRight(environmentAgentBaseURL, "/")
+
+	resp, err := httpClient.Get(environmentAgentBaseURL + "/health")
+	if err != nil {
+		GinkgoWriter.Printf("Environment agent not reachable at %s: %v — storage registration tests will be skipped\n", environmentAgentBaseURL, err)
+		return
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		GinkgoWriter.Printf("Environment agent health returned %d — storage registration tests will be skipped\n", resp.StatusCode)
+		return
+	}
+	environmentAgentReady = true
+	GinkgoWriter.Printf("Environment agent ready at %s\n", environmentAgentBaseURL)
+}
+
+func requireEnvironmentAgent() {
+	if !environmentAgentReady {
+		Skip("Environment agent not available (set DCM_ENVIRONMENT_AGENT_URL and deploy environment-agent)")
+	}
+}
+
+func requireStorageSP() {
+	if !storageSPReady {
+		Skip("Storage SP not available (deploy with --k8s-storage-service-provider and publish port 8089)")
+	}
+}
+
+// requireClusterStoragePrereqs validates catalog hint and SP default classes before cluster tests.
+func requireClusterStoragePrereqs() {
+	requireCatalogStorageClass()
+	requireSPDefaultStorageClass()
+}
+
+func requireCatalogStorageClass() {
+	requireKubectl()
+	sc := catalogStorageClassHint()
+	out, err := runStorageKubectl("get", "sc", sc)
+	if err != nil {
+		Fail(fmt.Sprintf(
+			"StorageClass %q not found on cluster (catalog provider_hints.kubernetes.storage_class). "+
+				"Set E2E_CATALOG_STORAGE_CLASS to an existing class or deploy with --k8s-storage-service-provider "+
+				"(writes .dcm-e2e.env). kubectl: %s",
+			sc, strings.TrimSpace(out),
+		))
+	}
+}
+
+func requireSPDefaultStorageClass() {
+	requireKubectl()
+	sc := spDefaultStorageClass()
+	if sc == "" {
+		Fail("K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS unset — deploy with --k8s-storage-service-provider " +
+			"(ensure_storage_class_for_e2e sets SP_K8S_DEFAULT_STORAGE_CLASS and writes .dcm-e2e.env)")
+	}
+	out, err := runStorageKubectl("get", "sc", sc)
+	if err != nil {
+		Fail(fmt.Sprintf(
+			"StorageClass %q not found on cluster (SP_K8S_DEFAULT_STORAGE_CLASS / K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS). "+
+				"Deploy-dcm.sh validates and writes .dcm-e2e.env. kubectl: %s",
+			sc, strings.TrimSpace(out),
+		))
+	}
+}
+
+// catalogStorageClassHint returns the storage_class simulated as a CatalogItem default in SP-direct E2E.
+// Override with E2E_CATALOG_STORAGE_CLASS when the cluster does not have a "standard" StorageClass.
+func catalogStorageClassHint() string {
+	if sc := os.Getenv("E2E_CATALOG_STORAGE_CLASS"); sc != "" {
+		return sc
+	}
+	return defaultCatalogStorageClass
+}
+
+// spDefaultStorageClass returns the SP fallback when provider_hints omit storage_class (TC-2.1.6).
+// Set by deploy-dcm.sh → SP_K8S_DEFAULT_STORAGE_CLASS, not by catalog.
+func spDefaultStorageClass() string {
+	if sc := os.Getenv("K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS"); sc != "" {
+		return sc
+	}
+	return ""
+}
+
+// doStorageSPRequest sends a request to the storage SP's direct API.
+func doStorageSPRequest(method, path string, body string) (*http.Response, error) {
+	url := storageSPBaseURL + path
+
+	var reqBody io.Reader
+	if body != "" {
+		reqBody = strings.NewReader(body)
+	}
+
+	req, err := http.NewRequest(method, url, reqBody)
+	if err != nil {
+		return nil, err
+	}
+	if body != "" {
+		req.Header.Set("Content-Type", "application/json")
+	}
+
+	return httpClient.Do(req)
+}
+
+// createTestVolume creates a volume via the SP API and returns the parsed response body.
+func createTestVolume(spec string) map[string]interface{} {
+	resp, err := doStorageSPRequest(http.MethodPost, "/volumes", spec)
+	Expect(err).NotTo(HaveOccurred())
+	Expect(resp.StatusCode).To(Equal(http.StatusCreated),
+		"create volume failed with status %d", resp.StatusCode)
+
+	var body map[string]interface{}
+	decodeJSON(resp, &body)
+	Expect(body).To(HaveKey("id"))
+	return body
+}
+
+// deleteTestVolume removes a volume by ID, ignoring 404 (already gone).
+func deleteTestVolume(id string) {
+	deletePVCConsumer(id)
+	resp, err := doStorageSPRequest(http.MethodDelete, "/volumes/"+id, "")
+	if err != nil {
+		GinkgoWriter.Printf("Warning: cleanup DELETE failed for volume %s: %v\n", id, err)
+		return
+	}
+	resp.Body.Close()
+}
+
+// volumeSpec returns a JSON body with catalog-derived provider_hints (simulates SpecBuilder output).
+func volumeSpec(name, capacity string) string {
+	return volumeSpecWith(name, capacity, catalogStorageClassHint())
+}
+
+// volumeSpecMinimal returns a JSON body without provider_hints (SP/cluster defaults).
+func volumeSpecMinimal(name, capacity string) string {
+	spec := map[string]interface{}{
+		"service_type": "storage",
+		"capacity":     capacity,
+		"metadata": map[string]interface{}{
+			"name": name,
+		},
+	}
+	body := map[string]interface{}{"spec": spec}
+	data, _ := json.Marshal(body)
+	return string(data)
+}
+
+// volumeSpecWith returns a JSON body with an explicit StorageClass hint.
+func volumeSpecWith(name, capacity, storageClass string) string {
+	return volumeSpecWithHints(name, capacity, storageClass, "", "")
+}
+
+// volumeSpecWithHints returns a JSON body with optional kubernetes provider hints.
+func volumeSpecWithHints(name, capacity, storageClass, accessMode, volumeMode string) string {
+	spec := map[string]interface{}{
+		"service_type": "storage",
+		"capacity":     capacity,
+		"metadata": map[string]interface{}{
+			"name": name,
+		},
+	}
+	k8sHints := map[string]interface{}{}
+	if storageClass != "" {
+		k8sHints["storage_class"] = storageClass
+	}
+	if accessMode != "" {
+		k8sHints["access_mode"] = accessMode
+	}
+	if volumeMode != "" {
+		k8sHints["volume_mode"] = volumeMode
+	}
+	if len(k8sHints) > 0 {
+		spec["provider_hints"] = map[string]interface{}{
+			"kubernetes": k8sHints,
+		}
+	}
+	body := map[string]interface{}{"spec": spec}
+	data, _ := json.Marshal(body)
+	return string(data)
+}
+
+// getStoragePVCJSON fetches a PVC by name via kubectl and returns parsed JSON.
+func getStoragePVCJSON(name string) map[string]interface{} {
+	out, err := runStorageKubectl("get", "pvc", name, "-o", "json")
+	ExpectWithOffset(1, err).NotTo(HaveOccurred(), "failed to get PVC %s", name)
+
+	var pvc map[string]interface{}
+	ExpectWithOffset(1, json.Unmarshal([]byte(out), &pvc)).To(Succeed())
+	return pvc
+}
+
+// runStorageKubectl executes kubectl/oc in the storage SP namespace.
+func runStorageKubectl(args ...string) (string, error) {
+	fullArgs := append([]string{"-n", storageSPNamespace}, args...)
+	cmd := exec.Command(kubectlBin, fullArgs...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		GinkgoWriter.Printf("kubectl %v failed: %s\n", args, string(out))
+	}
+	return string(out), err
+}
+
+// applyStorageManifest applies a Kubernetes manifest in the storage SP namespace.
+func applyStorageManifest(manifest string) error {
+	cmd := exec.Command(kubectlBin, "-n", storageSPNamespace, "apply", "-f", "-")
+	cmd.Stdin = strings.NewReader(manifest)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		GinkgoWriter.Printf("kubectl apply failed: %s\n", string(out))
+	}
+	return err
+}
+
+// pvcConsumerPodName returns a deterministic pod name for binding a PVC on WaitForFirstConsumer clusters.
+func pvcConsumerPodName(pvcName string) string {
+	const prefix = "e2e-pvc-bind-"
+	name := prefix + pvcName
+	if len(name) <= 63 {
+		return name
+	}
+	return prefix + pvcName[len(pvcName)-(63-len(prefix)):]
+}
+
+// ensurePVCConsumer schedules a pod that mounts the PVC so local-path (WaitForFirstConsumer) can bind it.
+func ensurePVCConsumer(pvcName string) {
+	requireKubectl()
+
+	podName := pvcConsumerPodName(pvcName)
+	manifest := fmt.Sprintf(`apiVersion: v1
+kind: Pod
+metadata:
+  name: %s
+  labels:
+    dcm.project/managed-by: dcm-e2e
+spec:
+  restartPolicy: Never
+  containers:
+  - name: pause
+    image: registry.k8s.io/pause:3.9
+    volumeMounts:
+    - name: vol
+      mountPath: /data
+  volumes:
+  - name: vol
+    persistentVolumeClaim:
+      claimName: %s
+`, podName, pvcName)
+
+	Expect(applyStorageManifest(manifest)).To(Succeed(), "failed to schedule PVC consumer pod for %s", pvcName)
+
+	Eventually(func() string {
+		out, err := runStorageKubectl("get", "pod", podName, "-o", "jsonpath={.status.phase}")
+		if err != nil {
+			return ""
+		}
+		return strings.TrimSpace(out)
+	}).WithTimeout(60 * time.Second).WithPolling(2 * time.Second).Should(
+		SatisfyAny(Equal("Running"), Equal("Succeeded")),
+		"PVC consumer pod %s should start", podName)
+}
+
+func deletePVCConsumer(pvcName string) {
+	if kubectlBin == "" {
+		return
+	}
+	podName := pvcConsumerPodName(pvcName)
+	_, _ = runStorageKubectl("delete", "pod", podName, "--ignore-not-found")
+}
+
+// describeStorageVolumeDebug returns PVC status context for RUNNING wait failures.
+func describeStorageVolumeDebug(volumeID string) string {
+	phase, err := runStorageKubectl("get", "pvc", volumeID,
+		"-o", "jsonpath={.status.phase}/{.spec.storageClassName}")
+	if err != nil {
+		return fmt.Sprintf("PVC lookup failed (catalog hint storage_class: %q): %v", catalogStorageClassHint(), err)
+	}
+	phase = strings.TrimSpace(phase)
+	events, _ := runStorageKubectl("get", "events",
+		"--field-selector", fmt.Sprintf("involvedObject.name=%s", volumeID),
+		"-o", "jsonpath={range .items[*]}{.reason}: {.message}{\"\\n\"}{end}")
+	events = strings.TrimSpace(events)
+	msg := fmt.Sprintf("PVC phase/class=%s (catalog hint storage_class=%q, SP default K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS=%q)",
+		phase, catalogStorageClassHint(), os.Getenv("K8S_STORAGE_SP_DEFAULT_STORAGE_CLASS"))
+	if events != "" {
+		msg += "; events: " + events
+	}
+	return msg
+}
+
+// waitForStorageVolumeRunning waits until the SP reports RUNNING, scheduling a PVC consumer when needed.
+func waitForStorageVolumeRunning(volumeID string, timeout time.Duration) {
+	requireCatalogStorageClass()
+	ensurePVCConsumer(volumeID)
+
+	Eventually(func() string {
+		resp, err := doStorageSPRequest(http.MethodGet, "/volumes/"+volumeID, "")
+		if err != nil {
+			return ""
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			return ""
+		}
+		var body map[string]interface{}
+		decodeJSON(resp, &body)
+		s, _ := body["status"].(string)
+		return s
+	}).WithTimeout(timeout).WithPolling(3 * time.Second).Should(Equal("RUNNING"),
+		"volume %s did not reach RUNNING; %s", volumeID, describeStorageVolumeDebug(volumeID))
+}
+
+// doEnvironmentAgentRequest sends a request to the environment agent API.
+func doEnvironmentAgentRequest(method, path string, body string) (*http.Response, error) {
+	url := environmentAgentBaseURL + path
+
+	var reqBody io.Reader
+	if body != "" {
+		reqBody = strings.NewReader(body)
+	}
+
+	req, err := http.NewRequest(method, url, reqBody)
+	if err != nil {
+		return nil, err
+	}
+	if body != "" {
+		req.Header.Set("Content-Type", "application/json")
+	}
+
+	return httpClient.Do(req)
+}
+
+// listEnvironmentAgentProviders returns all providers from the environment agent,
+// following pagination tokens until exhausted.
+func listEnvironmentAgentProviders() []map[string]interface{} {
+	var all []map[string]interface{}
+	token := ""
+
+	for {
+		path := "/providers"
+		if token != "" {
+			path += "?page_token=" + url.QueryEscape(token)
+		}
+
+		resp, err := doEnvironmentAgentRequest(http.MethodGet, path, "")
+		Expect(err).NotTo(HaveOccurred())
+		Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+		var body map[string]interface{}
+		decodeJSON(resp, &body)
+
+		providers, ok := body["providers"].([]interface{})
+		Expect(ok).To(BeTrue(), "providers list response missing providers array")
+
+		for _, p := range providers {
+			provider, ok := p.(map[string]interface{})
+			Expect(ok).To(BeTrue())
+			all = append(all, provider)
+		}
+
+		next, _ := body["next_page_token"].(string)
+		if next == "" {
+			break
+		}
+		token = next
+	}
+
+	return all
+}
+
+func storageProvidersFromAgent() []map[string]interface{} {
+	var matched []map[string]interface{}
+	for _, p := range listEnvironmentAgentProviders() {
+		if st, _ := p["service_type"].(string); st == "storage" {
+			matched = append(matched, p)
+		}
+	}
+	return matched
+}
diff --git a/tests/e2e/sp_storage_registration_test.go b/tests/e2e/sp_storage_registration_test.go
new file mode 100644
index 0000000..8fdd3ad
--- /dev/null
+++ b/tests/e2e/sp_storage_registration_test.go
@@ -0,0 +1,48 @@
+//go:build e2e
+
+package e2e_test
+
+import (
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+// Registration contract for the storage SP against environment-agent
+// (control-plane#51 removed direct /providers; external SPs register via the agent).
+// Same Eventually/Consistently shape as other SP registration E2E specs.
+//
+// Label("registration") — optional env-agent checks; skipped until deploy-dcm wires
+// environment-agent and DCM_ENVIRONMENT_AGENT_URL is set.
+var _ = Describe("Storage SP registration with environment-agent", Label("sp", "storage", "registration"), func() {
+	BeforeEach(func() {
+		requireEnvironmentAgent()
+	})
+
+	// TC-E2E-020 analogue for storage
+	It("registers exactly one storage-type provider pointing at the SP volumes endpoint", func() {
+		var found map[string]interface{}
+		Eventually(func() int {
+			providers := storageProvidersFromAgent()
+			found = nil
+			for _, p := range providers {
+				found = p
+			}
+			return len(providers)
+		}, 60*time.Second, 2*time.Second).Should(Equal(1),
+			"expected exactly one storage provider registered with environment-agent")
+
+		Expect(found).NotTo(BeNil())
+		Expect(found["service_type"]).To(Equal("storage"))
+		Expect(found["endpoint"]).To(Equal(storageSPRegisteredEndpoint))
+	})
+
+	// TC-E2E-040 analogue — idempotent re-registration must not create duplicates
+	It("does not duplicate the storage registration across a re-registration cycle", func() {
+		Consistently(func() int {
+			return len(storageProvidersFromAgent())
+		}, 90*time.Second, 10*time.Second).Should(Equal(1),
+			"storage SP periodic re-registration must stay idempotent on name, not create duplicates")
+	})
+})
diff --git a/tests/e2e/sp_storage_status_test.go b/tests/e2e/sp_storage_status_test.go
new file mode 100644
index 0000000..7024e0e
--- /dev/null
+++ b/tests/e2e/sp_storage_status_test.go
@@ -0,0 +1,263 @@
+//go:build e2e
+
+package e2e_test
+
+import (
+	"net/http"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Storage SP Status Events", Label("sp", "storage", "nats"), func() {
+	BeforeEach(func() {
+		requireStorageSP()
+	})
+
+	Context("CloudEvent format", Ordered, func() {
+		var collector *NATSCollector
+		var volumeID string
+
+		BeforeAll(func() {
+			collector = newNATSCollector(natsStorageSubject)
+		})
+
+		AfterAll(func() {
+			collector.Close()
+			if volumeID != "" {
+				deleteTestVolume(volumeID)
+			}
+		})
+
+		It("publishes a CloudEvent on volume creation", func() {
+			name := uniqueName("e2e-ce")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+			volumeID = body["id"].(string)
+
+			Eventually(func() int {
+				return len(collector.EventsForInstance(volumeID))
+			}).WithTimeout(30 * time.Second).WithPolling(1 * time.Second).Should(BeNumerically(">", 0),
+				"expected at least one NATS event for volume %s", volumeID)
+
+			events := collector.EventsForInstance(volumeID)
+			evt := events[0]
+
+			By("validating CloudEvent required fields")
+			Expect(evt.SpecVersion).To(Equal("1.0"))
+			Expect(evt.ID).NotTo(BeEmpty())
+			Expect(evt.Source).To(HavePrefix("dcm/providers/"), "source should be dcm/providers/<providerName>")
+			Expect(evt.Type).NotTo(BeEmpty())
+			Expect(evt.Time).NotTo(BeEmpty())
+			Expect(evt.DataContentType).To(Equal("application/json"))
+		})
+	})
+
+	Context("healthy volume lifecycle", Label("cluster"), Ordered, func() {
+		var collector *NATSCollector
+		var volumeID string
+
+		BeforeAll(func() {
+			requireCatalogStorageClass()
+			collector = newNATSCollector(natsStorageSubject)
+		})
+
+		AfterAll(func() {
+			collector.Close()
+			if volumeID != "" {
+				deleteTestVolume(volumeID)
+			}
+		})
+
+		It("transitions to RUNNING and emits status events", func() {
+			name := uniqueName("e2e-run")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+			volumeID = body["id"].(string)
+
+			By("scheduling a PVC consumer pod for WaitForFirstConsumer storage classes")
+			ensurePVCConsumer(volumeID)
+
+			By("waiting for RUNNING status via NATS")
+			evt := collector.WaitForStatus(volumeID, "RUNNING", 120*time.Second)
+			Expect(evt.Data["status"]).To(Equal("RUNNING"))
+			Expect(evt.Data["id"]).To(Equal(volumeID))
+
+			By("confirming GET also shows RUNNING")
+			resp, err := doStorageSPRequest(http.MethodGet, "/volumes/"+volumeID, "")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(resp.StatusCode).To(Equal(http.StatusOK))
+
+			var getBody map[string]interface{}
+			decodeJSON(resp, &getBody)
+			Expect(getBody["status"]).To(Equal("RUNNING"))
+		})
+
+		It("has a valid status progression ending in RUNNING", func() {
+			events := collector.EventsForInstance(volumeID)
+			Expect(events).NotTo(BeEmpty())
+
+			validStatuses := map[string]bool{"PROVISIONING": true, "RUNNING": true}
+			for _, e := range events {
+				s, _ := e.Data["status"].(string)
+				Expect(validStatuses).To(HaveKey(s), "unexpected status %q in event history", s)
+			}
+			last := events[len(events)-1]
+			Expect(last.Data["status"]).To(Equal("RUNNING"))
+		})
+	})
+
+	Context("first observed status", Ordered, func() {
+		var collector *NATSCollector
+		var volumeID string
+
+		BeforeAll(func() {
+			collector = newNATSCollector(natsStorageSubject)
+		})
+
+		AfterAll(func() {
+			collector.Close()
+			if volumeID != "" {
+				deleteTestVolume(volumeID)
+			}
+		})
+
+		It("emits PROVISIONING or RUNNING as the first observed status", func() {
+			name := uniqueName("e2e-init")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+			volumeID = body["id"].(string)
+
+			Eventually(func() int {
+				return len(collector.EventsForInstance(volumeID))
+			}).WithTimeout(15 * time.Second).WithPolling(1 * time.Second).Should(BeNumerically(">", 0))
+
+			events := collector.EventsForInstance(volumeID)
+			first := events[0].Data["status"].(string)
+			Expect(first).To(SatisfyAny(Equal("PROVISIONING"), Equal("RUNNING")),
+				"first event should be either PROVISIONING or RUNNING, got %q", first)
+		})
+	})
+
+	Context("delete status event", Label("cluster"), Ordered, func() {
+		var collector *NATSCollector
+		var volumeID string
+
+		BeforeAll(func() {
+			requireCatalogStorageClass()
+			collector = newNATSCollector(natsStorageSubject)
+		})
+
+		AfterAll(func() {
+			collector.Close()
+			if volumeID != "" {
+				deleteTestVolume(volumeID)
+			}
+		})
+
+		It("creates a volume, waits for RUNNING, then deletes", func() {
+			name := uniqueName("e2e-del")
+			body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+			volumeID = body["id"].(string)
+
+			ensurePVCConsumer(volumeID)
+			collector.WaitForStatus(volumeID, "RUNNING", 120*time.Second)
+
+			deletePVCConsumer(volumeID)
+			resp, err := doStorageSPRequest(http.MethodDelete, "/volumes/"+volumeID, "")
+			Expect(err).NotTo(HaveOccurred())
+			resp.Body.Close()
+			Expect(resp.StatusCode).To(Equal(http.StatusNoContent))
+
+			By("waiting for a DELETED status event")
+			Eventually(func() bool {
+				for _, e := range collector.EventsForInstance(volumeID) {
+					s, _ := e.Data["status"].(string)
+					if s == "DELETED" {
+						return true
+					}
+				}
+				return false
+			}).WithTimeout(30 * time.Second).WithPolling(2 * time.Second).Should(BeTrue(),
+				"expected a DELETED status event after deletion")
+		})
+	})
+
+	Context("label filtering", Label("cluster"), Ordered, func() {
+		var collector *NATSCollector
+		manualPVCName := uniqueName("e2e-unlbl")
+
+		BeforeAll(func() {
+			requireCatalogStorageClass()
+			collector = newNATSCollector(natsStorageSubject)
+		})
+
+		AfterAll(func() {
+			collector.Close()
+			_, _ = runStorageKubectl("delete", "pvc", manualPVCName, "--ignore-not-found")
+		})
+
+		It("does not emit events for non-DCM PVCs", func() {
+			By("creating a PVC without DCM labels via kubectl")
+			manifest := `apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: ` + manualPVCName + `
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: ` + catalogStorageClassHint() + `
+  resources:
+    requests:
+      storage: 1Gi
+`
+			err := applyStorageManifest(manifest)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("waiting and confirming no events appear for the manual PVC")
+			Consistently(func() int {
+				return len(collector.EventsForInstance(manualPVCName))
+			}).WithTimeout(15 * time.Second).WithPolling(2 * time.Second).Should(Equal(0),
+				"no NATS events should be emitted for non-DCM PVC %s", manualPVCName)
+		})
+	})
+
+	Context("concurrent volumes", Ordered, func() {
+		var collector *NATSCollector
+		var volumeIDs []string
+
+		BeforeAll(func() {
+			collector = newNATSCollector(natsStorageSubject)
+		})
+
+		AfterAll(func() {
+			collector.Close()
+			for _, id := range volumeIDs {
+				deleteTestVolume(id)
+			}
+		})
+
+		It("emits independent event streams for each volume", func() {
+			const count = 3
+			for i := 0; i < count; i++ {
+				name := uniqueName("e2e-con")
+				body := createTestVolume(volumeSpec(name, defaultTestCapacity))
+				volumeIDs = append(volumeIDs, body["id"].(string))
+			}
+
+			for _, id := range volumeIDs {
+				Eventually(func() int {
+					return len(collector.EventsForInstance(id))
+				}).WithTimeout(120 * time.Second).WithPolling(2 * time.Second).Should(BeNumerically(">", 0),
+					"expected at least one event for volume %s", id)
+			}
+
+			By("verifying events are isolated per instance")
+			for _, id := range volumeIDs {
+				events := collector.EventsForInstance(id)
+				for _, e := range events {
+					Expect(e.Data["id"]).To(Equal(id),
+						"event for volume %s contains wrong instance ID", id)
+				}
+			}
+		})
+	})
+})
diff --git a/tests/run-e2e.sh b/tests/run-e2e.sh
index 2bc4500..2e7cab1 100755
--- a/tests/run-e2e.sh
+++ b/tests/run-e2e.sh
@@ -88,6 +88,34 @@ log()  { echo "==> $*"; }
 info() { echo "    $*"; }
 err()  { echo "ERROR: $*" >&2; }
 
+source_e2e_env_file() {
+    local env_file="${REPO_ROOT}/.dcm-e2e.env"
+    if [[ -f "${env_file}" ]]; then
+        set -a
+        # shellcheck source=/dev/null
+        . "${env_file}"
+        set +a
+        info "Loaded storage E2E environment from .dcm-e2e.env"
+    fi
+}
+
+# Load .dcm-e2e.env only when storage SP tests will run (not smoke/cli/acm/etc.).
+needs_storage_e2e_env() {
+    if [[ "${ENABLE_STORAGE_SP}" == "true" ]]; then
+        return 0
+    fi
+    if [[ -z "${LABEL_FILTER}" ]]; then
+        return 0
+    fi
+    if [[ "${LABEL_FILTER}" == *storage* ]]; then
+        return 0
+    fi
+    if [[ "${LABEL_FILTER}" == sp || "${LABEL_FILTER}" == sp* ]]; then
+        return 0
+    fi
+    return 1
+}
+
 # --- CLI binary resolution ------------------------------------------------- #
 
 download_dcm_cli() {
@@ -161,6 +189,7 @@ LABEL_FILTER=""
 JUNIT_REPORT=""
 DEPLOY_ARGS=()
 ENABLE_CONTAINER_SP=false
+ENABLE_STORAGE_SP=false
 ENABLE_ACM_CLUSTER_SP=false
 ENABLE_KUBEVIRT_SP=false
 KUBEVIRT_VM_NS_ARG=""
@@ -201,8 +230,13 @@ while [[ $# -gt 0 ]]; do
             ENABLE_CONTAINER_SP=true
             DEPLOY_ARGS+=("$1")
             shift ;;
+        --k8s-storage-service-provider)
+            ENABLE_STORAGE_SP=true
+            DEPLOY_ARGS+=("$1")
+            shift ;;
         --all-service-providers)
             ENABLE_CONTAINER_SP=true
+            ENABLE_STORAGE_SP=true
             ENABLE_ACM_CLUSTER_SP=true
             ENABLE_KUBEVIRT_SP=true
             DEPLOY_ARGS+=("$1")
@@ -218,7 +252,7 @@ while [[ $# -gt 0 ]]; do
         --deploy-acm|--deploy-mce)
             DEPLOY_ARGS+=("$1")
             shift ;;
-        --compose-file|--kubeconfig|--k8s-container-namespace|--acm-cluster-namespace|--cluster-api|--cluster-username|--cluster-password|--acm-cluster-sp-repo|--acm-cluster-sp-branch)
+        --compose-file|--kubeconfig|--k8s-container-namespace|--k8s-storage-namespace|--acm-cluster-namespace|--cluster-api|--cluster-username|--cluster-password|--acm-cluster-sp-repo|--acm-cluster-sp-branch)
             DEPLOY_ARGS+=("$1" "$2")
             shift 2 ;;
         --kubevirt-vm-namespace)
@@ -250,6 +284,10 @@ else
     log "Skipping deployment (--skip-deploy)"
 fi
 
+if needs_storage_e2e_env; then
+    source_e2e_env_file
+fi
+
 # Resolve CLI binary.
 if [[ "${SKIP_CLI}" == "false" ]]; then
     if resolve_dcm_cli; then
@@ -267,7 +305,7 @@ if [[ -n "${GATEWAY_URL}" ]]; then
 fi
 
 # Export SP URLs when providers are enabled.
-if [[ "${ENABLE_CONTAINER_SP}" == "true" ]] || [[ "${ENABLE_ACM_CLUSTER_SP}" == "true" ]]; then
+if [[ "${ENABLE_CONTAINER_SP}" == "true" ]] || [[ "${ENABLE_STORAGE_SP}" == "true" ]] || [[ "${ENABLE_ACM_CLUSTER_SP}" == "true" ]]; then
     export DCM_NATS_URL="${DCM_NATS_URL:-nats://localhost:4222}"
     info "DCM_NATS_URL=${DCM_NATS_URL}"
 fi
@@ -275,6 +313,10 @@ if [[ "${ENABLE_CONTAINER_SP}" == "true" ]]; then
     export DCM_CONTAINER_SP_URL="${DCM_CONTAINER_SP_URL:-http://localhost:8082/api/v1alpha1}"
     info "DCM_CONTAINER_SP_URL=${DCM_CONTAINER_SP_URL}"
 fi
+if [[ "${ENABLE_STORAGE_SP}" == "true" ]]; then
+    export DCM_STORAGE_SP_URL="${DCM_STORAGE_SP_URL:-http://localhost:8089/api/v1alpha1}"
+    info "DCM_STORAGE_SP_URL=${DCM_STORAGE_SP_URL}"
+fi
 if [[ "${ENABLE_ACM_CLUSTER_SP}" == "true" ]]; then
     export DCM_ACM_CLUSTER_SP_URL="${DCM_ACM_CLUSTER_SP_URL:-http://localhost:8083/api/v1alpha1}"
     info "DCM_ACM_CLUSTER_SP_URL=${DCM_ACM_CLUSTER_SP_URL}"