feat: Pipe and Task Grant resources (#620)

examples/resources/snowflake_pipe_grant/import.sh

@@ -0,0 +1,2 @@
+# format is database name | schema name | pipe name | privilege | true/false for with_grant_option
+terraform import snowflake_pipe_grant.example 'dbName|schemaName|pipeName|OPERATE|false'

examples/resources/snowflake_pipe_grant/resource.tf

@@ -0,0 +1,14 @@
+resource snowflake_pipe_grant grant {
+  database_name = "db"
+  schema_name   = "schema"
+  sequence_name = "sequence"
+
+  privilege = "operate"
+  roles = [
+    "role1",
+    "role2",
+  ]
+
+  on_future         = false
+  with_grant_option = false
+}

examples/resources/snowflake_task_grant/import.sh

@@ -0,0 +1,2 @@
+# format is database name | schema name | task name | privilege | true/false for with_grant_option
+terraform import snowflake_pipe_grant.example 'dbName|schemaName|taskName|OPERATE|false'

examples/resources/snowflake_task_grant/resource.tf

@@ -0,0 +1,14 @@
+resource snowflake_task_grant grant {
+  database_name = "db"
+  schema_name   = "schema"
+  sequence_name = "sequence"
+
+  privilege = "operate"
+  roles = [
+    "role1",
+    "role2",
+  ]
+
+  on_future         = false
+  with_grant_option = false
+}

pkg/provider/provider.go

@@ -138,13 +138,15 @@ func GetGrantResources() resources.TerraformGrantResources {
 		"snowflake_integration_grant":       resources.IntegrationGrant(),
 		"snowflake_masking_policy_grant":    resources.MaskingPolicyGrant(),
 		"snowflake_materialized_view_grant": resources.MaterializedViewGrant(),
+		"snowflake_pipe_grant":              resources.PipeGrant(),
 		"snowflake_procedure_grant":         resources.ProcedureGrant(),
 		"snowflake_resource_monitor_grant":  resources.ResourceMonitorGrant(),
 		"snowflake_schema_grant":            resources.SchemaGrant(),
 		"snowflake_sequence_grant":          resources.SequenceGrant(),
 		"snowflake_stage_grant":             resources.StageGrant(),
 		"snowflake_stream_grant":            resources.StreamGrant(),
 		"snowflake_table_grant":             resources.TableGrant(),
+		"snowflake_task_grant":              resources.TaskGrant(),
 		"snowflake_view_grant":              resources.ViewGrant(),
 		"snowflake_warehouse_grant":         resources.WarehouseGrant(),
 	}

pkg/resources/helpers_test.go

@@ -334,3 +334,19 @@ func maskingPolicyGrant(t *testing.T, id string, params map[string]interface{})
 	d.SetId(id)
 	return d
 }
+
+func pipeGrant(t *testing.T, id string, params map[string]interface{}) *schema.ResourceData {
+	r := require.New(t)
+	d := schema.TestResourceDataRaw(t, resources.PipeGrant().Resource.Schema, params)
+	r.NotNil(d)
+	d.SetId(id)
+	return d
+}
+
+func taskGrant(t *testing.T, id string, params map[string]interface{}) *schema.ResourceData {
+	r := require.New(t)
+	d := schema.TestResourceDataRaw(t, resources.TaskGrant().Resource.Schema, params)
+	r.NotNil(d)
+	d.SetId(id)
+	return d
+}

pkg/resources/pipe_grant.go

@@ -0,0 +1,199 @@
+package resources
+
+import (
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/snowflake"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/validation"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/pkg/errors"
+)
+
+var validPipePrivileges = NewPrivilegeSet(
+	privilegeMonitor,
+	privilegeOperate,
+	privilegeOwnership,
+)
+
+var pipeGrantSchema = map[string]*schema.Schema{
+	"pipe_name": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "The name of the pipe on which to grant privileges immediately (only valid if on_future is false).",
+		ForceNew:    true,
+	},
+	"schema_name": {
+		Type:        schema.TypeString,
+		Required:    true,
+		Description: "The name of the schema containing the current or future pipes on which to grant privileges.",
+		ForceNew:    true,
+	},
+	"database_name": {
+		Type:        schema.TypeString,
+		Required:    true,
+		Description: "The name of the database containing the current or future pipes on which to grant privileges.",
+		ForceNew:    true,
+	},
+	"privilege": {
+		Type:         schema.TypeString,
+		Optional:     true,
+		Description:  "The privilege to grant on the current or future pipe.",
+		Default:      "USAGE",
+		ValidateFunc: validation.ValidatePrivilege(validPipePrivileges.ToList(), true),
+		ForceNew:     true,
+	},
+	"roles": {
+		Type:        schema.TypeSet,
+		Elem:        &schema.Schema{Type: schema.TypeString},
+		Optional:    true,
+		Description: "Grants privilege to these roles.",
+		ForceNew:    true,
+	},
+	"on_future": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "When this is set to true and a schema_name is provided, apply this grant on all future pipes in the given schema. When this is true and no schema_name is provided apply this grant on all future pipes in the given database. The pipe_name field must be unset in order to use on_future.",
+		Default:     false,
+		ForceNew:    true,
+	},
+	"with_grant_option": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Description: "When this is set to true, allows the recipient role to grant the privileges to other roles.",
+		Default:     false,
+		ForceNew:    true,
+	},
+}
+
+// PipeGrant returns a pointer to the resource representing a pipe grant
+func PipeGrant() *TerraformGrantResource {
+	return &TerraformGrantResource{
+		Resource: &schema.Resource{
+			Create: CreatePipeGrant,
+			Read:   ReadPipeGrant,
+			Delete: DeletePipeGrant,
+
+			Schema: pipeGrantSchema,
+			Importer: &schema.ResourceImporter{
+				StateContext: schema.ImportStatePassthroughContext,
+			},
+		},
+		ValidPrivs: validPipePrivileges,
+	}
+}
+
+// CreatePipeGrant implements schema.CreateFunc
+func CreatePipeGrant(d *schema.ResourceData, meta interface{}) error {
+	var pipeName string
+	if name, ok := d.GetOk("pipe_name"); ok {
+		pipeName = name.(string)
+	}
+	dbName := d.Get("database_name").(string)
+	schemaName := d.Get("schema_name").(string)
+	priv := d.Get("privilege").(string)
+	futurePipes := d.Get("on_future").(bool)
+	grantOption := d.Get("with_grant_option").(bool)
+
+	if (pipeName == "") && !futurePipes {
+		return errors.New("pipe_name must be set unless on_future is true.")
+	}
+	if (pipeName != "") && futurePipes {
+		return errors.New("pipe_name must be empty if on_future is true.")
+	}
+
+	var builder snowflake.GrantBuilder
+	if futurePipes {
+		builder = snowflake.FuturePipeGrant(dbName, schemaName)
+	} else {
+		builder = snowflake.PipeGrant(dbName, schemaName, pipeName)
+	}
+
+	err := createGenericGrant(d, meta, builder)
+	if err != nil {
+		return err
+	}
+
+	grant := &grantID{
+		ResourceName: dbName,
+		SchemaName:   schemaName,
+		ObjectName:   pipeName,
+		Privilege:    priv,
+		GrantOption:  grantOption,
+	}
+	dataIDInput, err := grant.String()
+	if err != nil {
+		return err
+	}
+	d.SetId(dataIDInput)
+
+	return ReadPipeGrant(d, meta)
+}
+
+// ReadPipeGrant implements schema.ReadFunc
+func ReadPipeGrant(d *schema.ResourceData, meta interface{}) error {
+	grantID, err := grantIDFromString(d.Id())
+	if err != nil {
+		return err
+	}
+	dbName := grantID.ResourceName
+	schemaName := grantID.SchemaName
+	pipeName := grantID.ObjectName
+	priv := grantID.Privilege
+
+	err = d.Set("database_name", dbName)
+	if err != nil {
+		return err
+	}
+	err = d.Set("schema_name", schemaName)
+	if err != nil {
+		return err
+	}
+	futurePipesEnabled := false
+	if pipeName == "" {
+		futurePipesEnabled = true
+	}
+	err = d.Set("pipe_name", pipeName)
+	if err != nil {
+		return err
+	}
+	err = d.Set("on_future", futurePipesEnabled)
+	if err != nil {
+		return err
+	}
+	err = d.Set("privilege", priv)
+	if err != nil {
+		return err
+	}
+	err = d.Set("with_grant_option", grantID.GrantOption)
+	if err != nil {
+		return err
+	}
+
+	var builder snowflake.GrantBuilder
+	if futurePipesEnabled {
+		builder = snowflake.FuturePipeGrant(dbName, schemaName)
+	} else {
+		builder = snowflake.PipeGrant(dbName, schemaName, pipeName)
+	}
+
+	return readGenericGrant(d, meta, pipeGrantSchema, builder, futurePipesEnabled, validPipePrivileges)
+}
+
+// DeletePipeGrant implements schema.DeleteFunc
+func DeletePipeGrant(d *schema.ResourceData, meta interface{}) error {
+	grantID, err := grantIDFromString(d.Id())
+	if err != nil {
+		return err
+	}
+	dbName := grantID.ResourceName
+	schemaName := grantID.SchemaName
+	pipeName := grantID.ObjectName
+
+	futurePipes := (pipeName == "")
+
+	var builder snowflake.GrantBuilder
+	if futurePipes {
+		builder = snowflake.FuturePipeGrant(dbName, schemaName)
+	} else {
+		builder = snowflake.PipeGrant(dbName, schemaName, pipeName)
+	}
+	return deleteGenericGrant(d, meta, builder)
+}

pkg/resources/pipe_grant_acceptance_test.go

@@ -0,0 +1,92 @@
+package resources_test
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+)
+
+func TestAcc_PipeGrant(t *testing.T) {
+	accName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+
+	resource.Test(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: pipeGrantConfig(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_pipe_grant.test", "database_name", accName),
+					resource.TestCheckResourceAttr("snowflake_pipe_grant.test", "schema_name", accName),
+					resource.TestCheckResourceAttr("snowflake_pipe_grant.test", "pipe_name", accName),
+					resource.TestCheckResourceAttr("snowflake_pipe_grant.test", "with_grant_option", "false"),
+					resource.TestCheckResourceAttr("snowflake_pipe_grant.test", "privilege", "OPERATE"),
+				),
+			},
+		},
+	})
+}
+
+func pipeGrantConfig(name string) string {
+	s := `
+resource "snowflake_database" "test" {
+  name = "%v"
+  comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test" {
+  name = snowflake_database.test.name
+  database = snowflake_database.test.name
+  comment = "Terraform acceptance test"
+}
+
+resource "snowflake_table" "test" {
+  database = snowflake_database.test.name
+  schema   = snowflake_schema.test.name
+  name     = snowflake_schema.test.name
+  column {
+	name = "id"
+	type = "NUMBER(5,0)"
+  }
+  column {
+    name = "data"
+	type = "VARCHAR(16)"
+  }
+}
+
+resource "snowflake_role" "test" {
+  name = "%v"
+}
+
+resource "snowflake_stage" "test" {
+  name = snowflake_schema.test.name
+  database = snowflake_database.test.name
+  schema = snowflake_schema.test.name
+  comment = "Terraform acceptance test"
+}
+
+resource "snowflake_pipe_grant" "test" {
+  pipe_name = snowflake_pipe.test.name
+  database_name = snowflake_database.test.name
+  roles         = [snowflake_role.test.name]
+  schema_name   = snowflake_schema.test.name
+  privilege 	  = "OPERATE"
+}
+
+resource "snowflake_pipe" "test" {
+  database       = snowflake_database.test.name
+  schema         = snowflake_schema.test.name
+  name           = snowflake_schema.test.name
+  comment        = "Terraform acceptance test"
+  copy_statement = <<CMD
+COPY INTO "${snowflake_table.test.database}"."${snowflake_table.test.schema}"."${snowflake_table.test.name}"
+  FROM @"${snowflake_stage.test.database}"."${snowflake_stage.test.schema}"."${snowflake_stage.test.name}"
+  FILE_FORMAT = (TYPE = CSV)
+CMD
+  auto_ingest    = false
+}
+`
+	return fmt.Sprintf(s, name, name)
+}