fix(permission-grant): validate namespace exists during plan

pulsar/resource_pulsar_permission_grant.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"strings"
 	"sync"
 	"time"
 
@@ -72,6 +73,7 @@ func resourcePulsarPermissionGrant() *schema.Resource {
 		ReadContext:   resourcePulsarPermissionGrantRead,
 		UpdateContext: resourcePulsarPermissionGrantUpdate,
 		DeleteContext: resourcePulsarPermissionGrantDelete,
+		CustomizeDiff: resourcePulsarPermissionGrantCustomizeDiff,
 
 		Description: `Provides a resource for managing permissions on either Pulsar namespaces or topics.
 Permission can be granted to specific roles using this resource.
@@ -123,6 +125,80 @@ See the ` + "`permission_grant`" + ` attribute of ` + "`pulsar_namespace`" + ` a
 	}
 }
 
+// resourcePulsarPermissionGrantCustomizeDiff runs during the plan phase and fails
+// the plan when the grant references a namespace that is known not to exist. Without
+// this check the missing namespace is only discovered at apply time as a
+// "404 Namespace not found" error, after the change has already been merged.
+func resourcePulsarPermissionGrantCustomizeDiff(_ context.Context, diff *schema.ResourceDiff,
+	meta interface{}) error {
+	// meta is nil when the provider is not configured (e.g. `terraform validate`).
+	// There is no client to consult, so skip the check rather than failing.
+	if meta == nil {
+		return nil
+	}
+	client := getClientFromMeta(meta)
+
+	namespace, _ := diff.GetOk("namespace")
+	topic, _ := diff.GetOk("topic")
+
+	return verifyGrantNamespaceExists(client, namespace.(string), topic.(string))
+}
+
+// verifyGrantNamespaceExists returns a diagnostic-style error when the namespace
+// targeted by the grant (either directly, or the namespace owning the topic) is
+// known not to exist.
+//
+// It intentionally errs on the side of NOT failing the plan: if the namespace
+// cannot be determined (unparseable input, or the tenant listing call fails —
+// which also covers the tenant itself not existing yet, and transient/network
+// errors), the check is skipped. It only returns an error when the tenant's
+// namespace listing succeeds AND the target namespace is provably absent. This
+// avoids false positives during a fresh apply where the tenant/namespace are
+// being created in the same run.
+func verifyGrantNamespaceExists(client admin.Client, namespace, topic string) error {
+	var nsName *utils.NameSpaceName
+
+	switch {
+	case namespace != "":
+		parsed, err := utils.GetNamespaceName(namespace)
+		if err != nil {
+			// Malformed input is reported by the CRUD path; don't block the plan here.
+			return nil
+		}
+		nsName = parsed
+	case topic != "":
+		topicName, err := utils.GetTopicName(topic)
+		if err != nil {
+			return nil
+		}
+		parsed, err := utils.GetNameSpaceName(topicName.GetTenant(), topicName.GetNamespace())
+		if err != nil {
+			return nil
+		}
+		nsName = parsed
+	default:
+		// Neither set: the schema's ExactlyOneOf validation handles this.
+		return nil
+	}
+
+	// NameSpaceName only exposes String() ("tenant/namespace"); the tenant is the
+	// first segment and is required to list the tenant's namespaces.
+	tenant := strings.SplitN(nsName.String(), "/", 2)[0]
+	namespaces, err := client.Namespaces().GetNamespaces(tenant)
+	if err != nil {
+		// Cannot determine existence (tenant missing, network error, etc.). Skip.
+		return nil
+	}
+
+	if !contains(namespaces, nsName.String()) {
+		return fmt.Errorf(
+			"namespace %q does not exist; create it before granting permissions "+
+				"(referenced by pulsar_permission_grant)", nsName.String())
+	}
+
+	return nil
+}
+
 func resourcePulsarPermissionGrantCreate(ctx context.Context, d *schema.ResourceData,
 	meta interface{}) diag.Diagnostics {
 	client := getClientFromMeta(meta)

pulsar/resource_pulsar_permission_grant_test.go

@@ -712,3 +712,93 @@ resource "pulsar_permission_grant" "test" {
 }
 `, wsURL, role)
 }
+
+// TestPermissionGrantNamespaceDoesNotExist verifies that the plan fails when a
+// grant references a namespace that does not exist under an existing tenant —
+// exactly the misconfiguration that previously only surfaced as a 404 at apply
+// time.
+//
+// The test is intentionally two steps. The plan-time check deliberately skips
+// when the tenant's namespace listing fails (e.g. the tenant does not exist yet
+// during a fresh apply), so the tenant must be created first. Step 1 provisions
+// only the cluster and tenant; step 2 introduces the grant against a namespace
+// that is never declared, so the listing succeeds and provably lacks it.
+func TestPermissionGrantNamespaceDoesNotExist(t *testing.T) {
+	cName := acctest.RandString(10)
+	tName := acctest.RandString(10)
+	nsName := acctest.RandString(10)
+	roleName := acctest.RandString(10)
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:          func() { testAccPreCheck(t) },
+		ProviderFactories: testAccProviderFactories,
+		Steps: []resource.TestStep{
+			{
+				// Create the tenant (and cluster) so the namespace listing succeeds later.
+				Config: testPulsarPermissionGrantTenantOnly(testWebServiceURL, cName, tName),
+			},
+			{
+				// Add a grant referencing a namespace that was never created.
+				Config: testPulsarPermissionGrantMissingNamespace(
+					testWebServiceURL, cName, tName, nsName, roleName),
+				ExpectError: regexp.MustCompile(`namespace .* does not exist`),
+			},
+		},
+	})
+}
+
+func testPulsarPermissionGrantTenantOnly(wsURL, cluster, tenant string) string {
+	return fmt.Sprintf(`
+provider "pulsar" {
+    web_service_url = "%s"
+}
+
+resource "pulsar_cluster" "test_cluster" {
+    cluster = "%s"
+
+    cluster_data {
+        web_service_url    = "http://localhost:8080"
+        broker_service_url = "pulsar://localhost:6050"
+        peer_clusters      = ["standalone"]
+    }
+}
+
+resource "pulsar_tenant" "test_tenant" {
+    tenant           = "%s"
+    allowed_clusters = [pulsar_cluster.test_cluster.cluster, "standalone"]
+}
+`, wsURL, cluster, tenant)
+}
+
+// testPulsarPermissionGrantMissingNamespace keeps the cluster and tenant from the
+// previous step and adds a permission grant referencing a namespace under that
+// tenant that is never declared. The tenant's namespace listing succeeds and
+// does not contain the referenced namespace, so the plan-time check must fail.
+func testPulsarPermissionGrantMissingNamespace(wsURL, cluster, tenant, namespace, role string) string {
+	return fmt.Sprintf(`
+provider "pulsar" {
+    web_service_url = "%s"
+}
+
+resource "pulsar_cluster" "test_cluster" {
+    cluster = "%s"
+
+    cluster_data {
+        web_service_url    = "http://localhost:8080"
+        broker_service_url = "pulsar://localhost:6050"
+        peer_clusters      = ["standalone"]
+    }
+}
+
+resource "pulsar_tenant" "test_tenant" {
+    tenant           = "%s"
+    allowed_clusters = [pulsar_cluster.test_cluster.cluster, "standalone"]
+}
+
+resource "pulsar_permission_grant" "test" {
+    namespace = "${pulsar_tenant.test_tenant.tenant}/%s"
+    role      = "%s"
+    actions   = ["produce", "consume"]
+}
+`, wsURL, cluster, tenant, namespace, role)
+}

pulsar/resource_pulsar_permission_grant_unit_test.go

@@ -27,11 +27,130 @@ import (
 	"testing"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// fakeNamespaces is a minimal admin.Namespaces implementation for unit tests.
+// It embeds the interface so unimplemented methods compile (calling any of them
+// would panic, which these tests never do) and overrides only GetNamespaces.
+type fakeNamespaces struct {
+	admin.Namespaces
+	namespaces []string
+	err        error
+	gotTenant  string
+	called     bool
+}
+
+func (f *fakeNamespaces) GetNamespaces(tenant string) ([]string, error) {
+	f.called = true
+	f.gotTenant = tenant
+	return f.namespaces, f.err
+}
+
+// fakeAdminClient embeds admin.Client and returns a canned Namespaces impl.
+type fakeAdminClient struct {
+	admin.Client
+	ns *fakeNamespaces
+}
+
+func (f *fakeAdminClient) Namespaces() admin.Namespaces {
+	return f.ns
+}
+
+func TestVerifyGrantNamespaceExists(t *testing.T) {
+	tests := []struct {
+		name          string
+		namespace     string
+		topic         string
+		nsList        []string
+		nsErr         error
+		wantErr       bool
+		wantErrSubstr string
+		wantTenant    string // expected tenant passed to GetNamespaces ("" = not called)
+	}{
+		{
+			name:       "namespace exists",
+			namespace:  "my-tenant/my-ns",
+			nsList:     []string{"my-tenant/my-ns", "my-tenant/other"},
+			wantErr:    false,
+			wantTenant: "my-tenant",
+		},
+		{
+			name:          "namespace missing",
+			namespace:     "my-tenant/missing",
+			nsList:        []string{"my-tenant/other"},
+			wantErr:       true,
+			wantErrSubstr: `namespace "my-tenant/missing" does not exist`,
+			wantTenant:    "my-tenant",
+		},
+		{
+			name:       "topic namespace exists",
+			topic:      "persistent://my-tenant/my-ns/my-topic",
+			nsList:     []string{"my-tenant/my-ns"},
+			wantErr:    false,
+			wantTenant: "my-tenant",
+		},
+		{
+			name:          "topic namespace missing",
+			topic:         "persistent://my-tenant/missing/my-topic",
+			nsList:        []string{"my-tenant/my-ns"},
+			wantErr:       true,
+			wantErrSubstr: `namespace "my-tenant/missing" does not exist`,
+			wantTenant:    "my-tenant",
+		},
+		{
+			name:       "listing error is skipped (tenant missing / network)",
+			namespace:  "my-tenant/my-ns",
+			nsErr:      rest.Error{Code: http.StatusNotFound, Reason: "Tenant does not exist"},
+			wantErr:    false,
+			wantTenant: "my-tenant",
+		},
+		{
+			name:       "malformed namespace is skipped",
+			namespace:  "not-a-valid-namespace",
+			wantErr:    false,
+			wantTenant: "", // GetNamespaces never reached
+		},
+		{
+			name:       "malformed topic is skipped",
+			topic:      "a/b", // 2-segment short name is rejected by GetTopicName
+			wantErr:    false,
+			wantTenant: "",
+		},
+		{
+			name:       "neither namespace nor topic is skipped",
+			wantErr:    false,
+			wantTenant: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ns := &fakeNamespaces{namespaces: tt.nsList, err: tt.nsErr}
+			client := &fakeAdminClient{ns: ns}
+
+			err := verifyGrantNamespaceExists(client, tt.namespace, tt.topic)
+
+			if tt.wantErr {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErrSubstr)
+			} else {
+				require.NoError(t, err)
+			}
+
+			if tt.wantTenant == "" {
+				assert.False(t, ns.called, "GetNamespaces should not have been called")
+			} else {
+				assert.True(t, ns.called, "GetNamespaces should have been called")
+				assert.Equal(t, tt.wantTenant, ns.gotTenant)
+			}
+		})
+	}
+}
+
 // TestGetPermissionLock_SameKeyReturnsSamePointer verifies that repeated calls
 // with the same key return the identical *sync.Mutex.
 func TestGetPermissionLock_SameKeyReturnsSamePointer(t *testing.T) {