review comments

internal/catalog/internal/testhelpers/acl_hooks_test_helpers.go

@@ -93,6 +93,16 @@ func RunWorkloadSelectingTypeACLsTests[T WorkloadSelecting](t *testing.T, typ *p
 			WriteOK: resourcetest.ALLOW,
 			ListOK:  resourcetest.DEFAULT,
 		},
+		"service test write with prefixed selectors and a policy with a specific service": {
+			Rules:  `service "test" { policy = "write" } service "workload" { policy = "read" }`,
+			Data:   getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload"}}),
+			Typ:    typ,
+			ReadOK: resourcetest.ALLOW,
+			// TODO (ishustava): this is wrong and should be fixed in a follow up PR. We should not allow
+			// a policy for a specific service when only prefixes are specified in the selector.
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
 	}
 
 	for name, tc := range cases {

internal/catalog/internal/types/dns_policy_test.go

@@ -175,7 +175,6 @@ func TestDNSPolicyACLs(t *testing.T) {
 				Weights:   &pbcatalog.Weights{Passing: 1, Warning: 0},
 			}
 		},
-		func(registry resource.Registry) {
-			RegisterDNSPolicy(registry)
-		})
+		RegisterDNSPolicy,
+	)
 }

internal/catalog/internal/types/health_checks_test.go

@@ -203,7 +203,6 @@ func TestHealthChecksACLs(t *testing.T) {
 		func(selector *pbcatalog.WorkloadSelector) *pbcatalog.HealthChecks {
 			return &pbcatalog.HealthChecks{Workloads: selector}
 		},
-		func(registry resource.Registry) {
-			RegisterHealthChecks(registry)
-		})
+		RegisterHealthChecks,
+	)
 }

internal/catalog/internal/types/health_status_test.go

@@ -292,6 +292,24 @@ func TestHealthStatusACLs(t *testing.T) {
 			WriteOK: resourcetest.ALLOW,
 			ListOK:  resourcetest.DEFAULT,
 		},
+		"node test read with workload owner": {
+			Rules:   `node "test" { policy = "read" }`,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test write with workload owner": {
+			Rules:   `node "test" { policy = "write" }`,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
 	}
 
 	for name, tc := range cases {

internal/catalog/internal/types/service_test.go

@@ -282,7 +282,6 @@ func TestServiceACLs(t *testing.T) {
 		func(selector *pbcatalog.WorkloadSelector) *pbcatalog.Service {
 			return &pbcatalog.Service{Workloads: selector}
 		},
-		func(registry resource.Registry) {
-			RegisterService(registry)
-		})
+		RegisterService,
+	)
 }

internal/mesh/internal/types/destinations_configuration_test.go

@@ -22,9 +22,8 @@ func TestDestinationsConfigurationACLs(t *testing.T) {
 		func(selector *pbcatalog.WorkloadSelector) *pbmesh.DestinationsConfiguration {
 			return &pbmesh.DestinationsConfiguration{Workloads: selector}
 		},
-		func(registry resource.Registry) {
-			RegisterDestinationsConfiguration(registry)
-		})
+		RegisterDestinationsConfiguration,
+	)
 }
 
 func TestValidateDestinationsConfiguration(t *testing.T) {

internal/mesh/internal/types/destinations_test.go

@@ -255,7 +255,6 @@ func TestDestinationsACLs(t *testing.T) {
 		func(selector *pbcatalog.WorkloadSelector) *pbmesh.Destinations {
 			return &pbmesh.Destinations{Workloads: selector}
 		},
-		func(registry resource.Registry) {
-			RegisterDestinations(registry)
-		})
+		RegisterDestinations,
+	)
 }

internal/mesh/internal/types/proxy_configuration_test.go

@@ -24,11 +24,13 @@ import (
 func TestProxyConfigurationACLs(t *testing.T) {
 	catalogtesthelpers.RunWorkloadSelectingTypeACLsTests[*pbmesh.ProxyConfiguration](t, pbmesh.ProxyConfigurationType,
 		func(selector *pbcatalog.WorkloadSelector) *pbmesh.ProxyConfiguration {
-			return &pbmesh.ProxyConfiguration{Workloads: selector}
+			return &pbmesh.ProxyConfiguration{
+				Workloads:     selector,
+				DynamicConfig: &pbmesh.DynamicConfig{},
+			}
 		},
-		func(registry resource.Registry) {
-			RegisterProxyConfiguration(registry)
-		})
+		RegisterProxyConfiguration,
+	)
 }
 
 func TestMutateProxyConfiguration(t *testing.T) {

internal/resource/resourcetest/acls.go

@@ -52,11 +52,15 @@ func RunACLTestCase(t *testing.T, tc ACLTestCase, registry resource.Registry) {
 	reg, ok := registry.Resolve(tc.Typ)
 	require.True(t, ok)
 
+	resolvedType, ok := registry.Resolve(tc.Typ)
+	require.True(t, ok)
+
 	res := Resource(tc.Typ, "test").
-		WithTenancy(resource.DefaultNamespacedTenancy()).
+		WithTenancy(DefaultTenancyForType(t, resolvedType)).
 		WithOwner(tc.Owner).
 		WithData(t, tc.Data).
 		Build()
+
 	ValidateAndNormalize(t, registry, res)
 
 	config := acl.Config{

internal/resource/resourcetest/tenancy.go

@@ -5,6 +5,7 @@ package resourcetest
 
 import (
 	"strings"
+	"testing"
 
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -35,3 +36,18 @@ func Tenancy(s string) *pbresource.Tenancy {
 		return &pbresource.Tenancy{Partition: "BAD", Namespace: "BAD", PeerName: "BAD"}
 	}
 }
+
+func DefaultTenancyForType(t *testing.T, reg resource.Registration) *pbresource.Tenancy {
+	switch reg.Scope {
+	case resource.ScopeNamespace:
+		return resource.DefaultNamespacedTenancy()
+	case resource.ScopePartition:
+		return resource.DefaultPartitionedTenancy()
+	case resource.ScopeCluster:
+		return resource.DefaultClusteredTenancy()
+	default:
+		t.Fatalf("unsupported resource scope: %v", reg.Scope)
+		return nil
+	}
+	return nil
+}