Repository: chanzuckerberg/terraform-provider-snowflake
Pull request #620 (2021-07-26 11:32:02)

feat: Pipe and Task Grant resources
Implemented Task and Pipe Grant Resources

<!-- summary of changes -->

## Test Plan
* [x] acceptance tests


## References
https://docs.snowflake.com/en/sql-reference/sql/grant-privilege.html
* 

alldoami: /ok-to-test sha=f0457c0
github-actions[bot]: Integration tests success for [f0457c0](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/1068822314)

Patch:
diff --git a/examples/resources/snowflake_pipe_grant/import.sh b/examples/resources/snowflake_pipe_grant/import.sh
new file mode 100644
index 00000000..49487226
--- /dev/null
+++ b/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'
diff --git a/examples/resources/snowflake_pipe_grant/resource.tf b/examples/resources/snowflake_pipe_grant/resource.tf
new file mode 100644
index 00000000..d8c92347
--- /dev/null
+++ b/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
+}
diff --git a/examples/resources/snowflake_task_grant/import.sh b/examples/resources/snowflake_task_grant/import.sh
new file mode 100644
index 00000000..767164e7
--- /dev/null
+++ b/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'
diff --git a/examples/resources/snowflake_task_grant/resource.tf b/examples/resources/snowflake_task_grant/resource.tf
new file mode 100644
index 00000000..c4684f9a
--- /dev/null
+++ b/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
+}
diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go
index f5e7d580..1c474d30 100644
--- a/pkg/provider/provider.go
+++ b/pkg/provider/provider.go
@@ -138,6 +138,7 @@ 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(),
@@ -145,6 +146,7 @@ func GetGrantResources() resources.TerraformGrantResources {
 		"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(),
 	}
diff --git a/pkg/resources/pipe_grant.go b/pkg/resources/pipe_grant.go
new file mode 100644
index 00000000..86f4a196
--- /dev/null
+++ b/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)
+}
diff --git a/pkg/resources/task_grant.go b/pkg/resources/task_grant.go
new file mode 100644
index 00000000..b5c7380c
--- /dev/null
+++ b/pkg/resources/task_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 validTaskPrivileges = NewPrivilegeSet(
+	privilegeMonitor,
+	privilegeOperate,
+	privilegeOwnership,
+)
+
+var taskGrantSchema = map[string]*schema.Schema{
+	"task_name": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "The name of the task 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 tasks 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 tasks on which to grant privileges.",
+		ForceNew:    true,
+	},
+	"privilege": {
+		Type:         schema.TypeString,
+		Optional:     true,
+		Description:  "The privilege to grant on the current or future task.",
+		Default:      "USAGE",
+		ValidateFunc: validation.ValidatePrivilege(validTaskPrivileges.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 tasks in the given schema. When this is true and no schema_name is provided apply this grant on all future tasks in the given database. The task_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,
+	},
+}
+
+// TaskGrant returns a pointer to the resource representing a task grant
+func TaskGrant() *TerraformGrantResource {
+	return &TerraformGrantResource{
+		Resource: &schema.Resource{
+			Create: CreateTaskGrant,
+			Read:   ReadTaskGrant,
+			Delete: DeleteTaskGrant,
+
+			Schema: taskGrantSchema,
+			Importer: &schema.ResourceImporter{
+				StateContext: schema.ImportStatePassthroughContext,
+			},
+		},
+		ValidPrivs: validTaskPrivileges,
+	}
+}
+
+// CreateTaskGrant implements schema.CreateFunc
+func CreateTaskGrant(d *schema.ResourceData, meta interface{}) error {
+	var taskName string
+	if name, ok := d.GetOk("task_name"); ok {
+		taskName = name.(string)
+	}
+	dbName := d.Get("database_name").(string)
+	schemaName := d.Get("schema_name").(string)
+	priv := d.Get("privilege").(string)
+	futureTasks := d.Get("on_future").(bool)
+	grantOption := d.Get("with_grant_option").(bool)
+
+	if (taskName == "") && !futureTasks {
+		return errors.New("task_name must be set unless on_future is true.")
+	}
+	if (taskName != "") && futureTasks {
+		return errors.New("task_name must be empty if on_future is true.")
+	}
+
+	var builder snowflake.GrantBuilder
+	if futureTasks {
+		builder = snowflake.FutureTaskGrant(dbName, schemaName)
+	} else {
+		builder = snowflake.TaskGrant(dbName, schemaName, taskName)
+	}
+
+	err := createGenericGrant(d, meta, builder)
+	if err != nil {
+		return err
+	}
+
+	grant := &grantID{
+		ResourceName: dbName,
+		SchemaName:   schemaName,
+		ObjectName:   taskName,
+		Privilege:    priv,
+		GrantOption:  grantOption,
+	}
+	dataIDInput, err := grant.String()
+	if err != nil {
+		return err
+	}
+	d.SetId(dataIDInput)
+
+	return ReadTaskGrant(d, meta)
+}
+
+// ReadTaskGrant implements schema.ReadFunc
+func ReadTaskGrant(d *schema.ResourceData, meta interface{}) error {
+	grantID, err := grantIDFromString(d.Id())
+	if err != nil {
+		return err
+	}
+	dbName := grantID.ResourceName
+	schemaName := grantID.SchemaName
+	taskName := 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
+	}
+	futureTasksEnabled := false
+	if taskName == "" {
+		futureTasksEnabled = true
+	}
+	err = d.Set("task_name", taskName)
+	if err != nil {
+		return err
+	}
+	err = d.Set("on_future", futureTasksEnabled)
+	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 futureTasksEnabled {
+		builder = snowflake.FutureTaskGrant(dbName, schemaName)
+	} else {
+		builder = snowflake.TaskGrant(dbName, schemaName, taskName)
+	}
+
+	return readGenericGrant(d, meta, taskGrantSchema, builder, futureTasksEnabled, validTaskPrivileges)
+}
+
+// DeleteTaskGrant implements schema.DeleteFunc
+func DeleteTaskGrant(d *schema.ResourceData, meta interface{}) error {
+	grantID, err := grantIDFromString(d.Id())
+	if err != nil {
+		return err
+	}
+	dbName := grantID.ResourceName
+	schemaName := grantID.SchemaName
+	taskName := grantID.ObjectName
+
+	futureTasks := (taskName == "")
+
+	var builder snowflake.GrantBuilder
+	if futureTasks {
+		builder = snowflake.FutureTaskGrant(dbName, schemaName)
+	} else {
+		builder = snowflake.TaskGrant(dbName, schemaName, taskName)
+	}
+	return deleteGenericGrant(d, meta, builder)
+}
diff --git a/pkg/snowflake/future_grant.go b/pkg/snowflake/future_grant.go
index 6787d581..7d9ca637 100644
--- a/pkg/snowflake/future_grant.go
+++ b/pkg/snowflake/future_grant.go
@@ -19,6 +19,8 @@ const (
 	futureProcedureType        futureGrantType = "PROCEDURE"
 	futureSequenceType         futureGrantType = "SEQUENCE"
 	futureStreamType           futureGrantType = "STREAM"
+	futurePipeType             futureGrantType = "PIPE"
+	futureTaskType             futureGrantType = "TASK"
 )
 
 const (
@@ -100,7 +102,7 @@ func FutureMaterializedViewGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureStageGrant returns a pointer to a FutureGrantBuilder for a table
+// FutureStageGrant returns a pointer to a FutureGrantBuilder for a stage
 func FutureStageGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -111,7 +113,7 @@ func FutureStageGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureExternalTableGrant returns a pointer to a FutureGrantBuilder for a view
+// FutureExternalTableGrant returns a pointer to a FutureGrantBuilder for a external table
 func FutureExternalTableGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -122,7 +124,7 @@ func FutureExternalTableGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureFileFormatGrant returns a pointer to a FutureGrantBuilder for a view
+// FutureFileFormatGrant returns a pointer to a FutureGrantBuilder for a file format
 func FutureFileFormatGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -133,7 +135,7 @@ func FutureFileFormatGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureFunctionGrant returns a pointer to a FutureGrantBuilder for a view
+// FutureFunctionGrant returns a pointer to a FutureGrantBuilder for a function
 func FutureFunctionGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -144,7 +146,7 @@ func FutureFunctionGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureProcedureGrant returns a pointer to a FutureGrantBuilder for a view
+// FutureProcedureGrant returns a pointer to a FutureGrantBuilder for a procedure
 func FutureProcedureGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -155,7 +157,7 @@ func FutureProcedureGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureSequenceGrant returns a pointer to a FutureGrantBuilder for a view
+// FutureSequenceGrant returns a pointer to a FutureGrantBuilder for a sequence
 func FutureSequenceGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -166,7 +168,7 @@ func FutureSequenceGrant(db, schema string) GrantBuilder {
 	}
 }
 
-// FutureStreamGrant returns a pointer to a FutureGrantBuilder for a view
+// FutureStreamGrant returns a pointer to a FutureGrantBuilder for a stream
 func FutureStreamGrant(db, schema string) GrantBuilder {
 	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
 	return &FutureGrantBuilder{
@@ -177,6 +179,28 @@ func FutureStreamGrant(db, schema string) GrantBuilder {
 	}
 }
 
+// FuturePipeGrant returns a pointer to a FutureGrantBuilder for a pipe
+func FuturePipeGrant(db, schema string) GrantBuilder {
+	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
+	return &FutureGrantBuilder{
+		name:              name,
+		qualifiedName:     qualifiedName,
+		futureGrantType:   futurePipeType,
+		futureGrantTarget: futureTarget,
+	}
+}
+
+// FutureTaskGrant returns a pointer to a FutureGrantBuilder for a task
+func FutureTaskGrant(db, schema string) GrantBuilder {
+	name, qualifiedName, futureTarget := getNameAndQualifiedName(db, schema)
+	return &FutureGrantBuilder{
+		name:              name,
+		qualifiedName:     qualifiedName,
+		futureGrantType:   futureTaskType,
+		futureGrantTarget: futureTarget,
+	}
+}
+
 // Show returns the SQL that will show all privileges on the grant
 func (fgb *FutureGrantBuilder) Show() string {
 	return fmt.Sprintf(`SHOW FUTURE GRANTS IN %v %v`, fgb.futureGrantTarget, fgb.qualifiedName)
diff --git a/pkg/snowflake/grant.go b/pkg/snowflake/grant.go
index 172827e7..86db10d1 100644
--- a/pkg/snowflake/grant.go
+++ b/pkg/snowflake/grant.go
@@ -27,6 +27,8 @@ const (
 	sequenceType         grantType = "SEQUENCE"
 	streamType           grantType = "STREAM"
 	maskingPolicyType    grantType = "MASKING POLICY"
+	pipeType             grantType = "PIPE"
+	taskType             grantType = "TASK"
 )
 
 type GrantExecutable interface {
@@ -262,6 +264,24 @@ func MaskingPolicyGrant(db, schema, maskingPolicy string) GrantBuilder {
 	}
 }
 
+// PipeGrant returns a pointer to a CurrentGrantBuilder for a pipe
+func PipeGrant(db, schema, pipe string) GrantBuilder {
+	return &CurrentGrantBuilder{
+		name:          pipe,
+		qualifiedName: fmt.Sprintf(`"%v"."%v"."%v"`, db, schema, pipe),
+		grantType:     pipeType,
+	}
+}
+
+// TaskGrant returns a pointer to a CurrentGrantBuilder for a task
+func TaskGrant(db, schema, task string) GrantBuilder {
+	return &CurrentGrantBuilder{
+		name:          task,
+		qualifiedName: fmt.Sprintf(`"%v"."%v"."%v"`, db, schema, task),
+		grantType:     taskType,
+	}
+}
+
 type granteeType string
 
 const (


Test patch:
diff --git a/pkg/resources/helpers_test.go b/pkg/resources/helpers_test.go
index 512cbfad..e3ccb45f 100644
--- a/pkg/resources/helpers_test.go
+++ b/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
+}
diff --git a/pkg/resources/pipe_grant_acceptance_test.go b/pkg/resources/pipe_grant_acceptance_test.go
new file mode 100644
index 00000000..f120f1c7
--- /dev/null
+++ b/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)
+}
diff --git a/pkg/resources/pipe_grant_test.go b/pkg/resources/pipe_grant_test.go
new file mode 100644
index 00000000..9d4ee691
--- /dev/null
+++ b/pkg/resources/pipe_grant_test.go
@@ -0,0 +1,153 @@
+package resources_test
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/provider"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/resources"
+	. "github.com/chanzuckerberg/terraform-provider-snowflake/pkg/testhelpers"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPipeGrant(t *testing.T) {
+	r := require.New(t)
+	err := resources.PipeGrant().Resource.InternalValidate(provider.Provider().Schema, true)
+	r.NoError(err)
+}
+
+func TestPipeGrantCreate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"pipe_name":         "test-pipe",
+		"schema_name":       "PUBLIC",
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{"test-role-1", "test-role-2"},
+		"with_grant_option": true,
+	}
+	d := schema.TestResourceDataRaw(t, resources.PipeGrant().Resource.Schema, in)
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(`^GRANT OPERATE ON PIPE "test-db"."PUBLIC"."test-pipe" TO ROLE "test-role-1" WITH GRANT OPTION$`).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(`^GRANT OPERATE ON PIPE "test-db"."PUBLIC"."test-pipe" TO ROLE "test-role-2" WITH GRANT OPTION$`).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadPipeGrant(mock)
+		err := resources.CreatePipeGrant(d, db)
+		r.NoError(err)
+	})
+}
+
+func TestPipeGrantRead(t *testing.T) {
+	r := require.New(t)
+
+	d := pipeGrant(t, "test-db|PUBLIC|test-pipe|OPERATE|false", map[string]interface{}{
+		"pipe_name":         "test-pipe",
+		"schema_name":       "PUBLIC",
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{},
+		"with_grant_option": false,
+	})
+
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		expectReadPipeGrant(mock)
+		err := resources.ReadPipeGrant(d, db)
+		r.NoError(err)
+	})
+
+	roles := d.Get("roles").(*schema.Set)
+	r.True(roles.Contains("test-role-1"))
+	r.True(roles.Contains("test-role-2"))
+	r.Equal(roles.Len(), 2)
+}
+
+func expectReadPipeGrant(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "privilege", "granted_on", "name", "granted_to", "grantee_name", "grant_option", "granted_by",
+	}).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "PIPE", "test-pipe", "ROLE", "test-role-1", false, "bob",
+	).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "PIPE", "test-pipe", "ROLE", "test-role-2", false, "bob",
+	)
+	mock.ExpectQuery(`^SHOW GRANTS ON PIPE "test-db"."PUBLIC"."test-pipe"$`).WillReturnRows(rows)
+}
+
+func TestFuturePipeGrantCreate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"on_future":         true,
+		"schema_name":       "PUBLIC",
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{"test-role-1", "test-role-2"},
+		"with_grant_option": true,
+	}
+	d := schema.TestResourceDataRaw(t, resources.PipeGrant().Resource.Schema, in)
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE PIPES IN SCHEMA "test-db"."PUBLIC" TO ROLE "test-role-1" WITH GRANT OPTION$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE PIPES IN SCHEMA "test-db"."PUBLIC" TO ROLE "test-role-2" WITH GRANT OPTION$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadFuturePipeGrant(mock)
+		err := resources.CreatePipeGrant(d, db)
+		r.NoError(err)
+	})
+
+	b := require.New(t)
+
+	in = map[string]interface{}{
+		"on_future":         true,
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{"test-role-1", "test-role-2"},
+		"with_grant_option": false,
+	}
+	d = schema.TestResourceDataRaw(t, resources.PipeGrant().Resource.Schema, in)
+	b.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE PIPES IN DATABASE "test-db" TO ROLE "test-role-1"$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE PIPES IN DATABASE "test-db" TO ROLE "test-role-2"$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadFuturePipeDatabaseGrant(mock)
+		err := resources.CreatePipeGrant(d, db)
+		b.NoError(err)
+	})
+}
+
+func expectReadFuturePipeGrant(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "privilege", "grant_on", "name", "grant_to", "grantee_name", "grant_option",
+	}).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "PIPE", "test-db.PUBLIC.<SCHEMA>", "ROLE", "test-role-1", false,
+	).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "PIPE", "test-db.PUBLIC.<SCHEMA>", "ROLE", "test-role-2", false,
+	)
+	mock.ExpectQuery(`^SHOW FUTURE GRANTS IN SCHEMA "test-db"."PUBLIC"$`).WillReturnRows(rows)
+}
+
+func expectReadFuturePipeDatabaseGrant(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "privilege", "grant_on", "name", "grant_to", "grantee_name", "grant_option",
+	}).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "PIPE", "test-db.<SCHEMA>", "ROLE", "test-role-1", false,
+	).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "PIPE", "test-db.<SCHEMA>", "ROLE", "test-role-2", false,
+	)
+	mock.ExpectQuery(`^SHOW FUTURE GRANTS IN DATABASE "test-db"$`).WillReturnRows(rows)
+}
diff --git a/pkg/resources/task_grant_acceptance_test.go b/pkg/resources/task_grant_acceptance_test.go
new file mode 100644
index 00000000..8cb1a72d
--- /dev/null
+++ b/pkg/resources/task_grant_acceptance_test.go
@@ -0,0 +1,75 @@
+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_TaskGrant(t *testing.T) {
+	accName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+
+	resource.Test(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: taskGrantConfig(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_task_grant.test", "database_name", accName),
+					resource.TestCheckResourceAttr("snowflake_task_grant.test", "schema_name", accName),
+					resource.TestCheckResourceAttr("snowflake_task_grant.test", "task_name", accName),
+					resource.TestCheckResourceAttr("snowflake_task_grant.test", "with_grant_option", "false"),
+					resource.TestCheckResourceAttr("snowflake_task_grant.test", "privilege", "OPERATE"),
+				),
+			},
+		},
+	})
+}
+
+func taskGrantConfig(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_role" "test" {
+  name = "%v"
+}
+
+resource "snowflake_warehouse" "test" {
+  name = snowflake_database.test.name
+}
+
+resource "snowflake_task" "test" {
+  name     	    = snowflake_schema.test.name
+  database  	= snowflake_database.test.name
+  schema   	  	= snowflake_schema.test.name
+  warehouse 	= snowflake_warehouse.test.name
+  sql_statement = "SHOW FUNCTIONS"
+  enabled  	  	= true
+  schedule 	  	= "15 MINUTES"
+  lifecycle {
+    ignore_changes = [session_parameters]
+  }
+}
+
+resource "snowflake_task_grant" "test" {
+  task_name 	= snowflake_task.test.name
+  database_name = snowflake_database.test.name
+  roles         = [snowflake_role.test.name]
+  schema_name   = snowflake_schema.test.name
+  privilege 	= "OPERATE"
+}
+`
+	return fmt.Sprintf(s, name, name)
+}
diff --git a/pkg/resources/task_grant_test.go b/pkg/resources/task_grant_test.go
new file mode 100644
index 00000000..dbfc3828
--- /dev/null
+++ b/pkg/resources/task_grant_test.go
@@ -0,0 +1,153 @@
+package resources_test
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/provider"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/resources"
+	. "github.com/chanzuckerberg/terraform-provider-snowflake/pkg/testhelpers"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTaskGrant(t *testing.T) {
+	r := require.New(t)
+	err := resources.TaskGrant().Resource.InternalValidate(provider.Provider().Schema, true)
+	r.NoError(err)
+}
+
+func TestTaskGrantCreate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"task_name":         "test-task",
+		"schema_name":       "PUBLIC",
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{"test-role-1", "test-role-2"},
+		"with_grant_option": true,
+	}
+	d := schema.TestResourceDataRaw(t, resources.TaskGrant().Resource.Schema, in)
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(`^GRANT OPERATE ON TASK "test-db"."PUBLIC"."test-task" TO ROLE "test-role-1" WITH GRANT OPTION$`).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(`^GRANT OPERATE ON TASK "test-db"."PUBLIC"."test-task" TO ROLE "test-role-2" WITH GRANT OPTION$`).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadTaskGrant(mock)
+		err := resources.CreateTaskGrant(d, db)
+		r.NoError(err)
+	})
+}
+
+func TestTaskGrantRead(t *testing.T) {
+	r := require.New(t)
+
+	d := taskGrant(t, "test-db|PUBLIC|test-task|OPERATE|false", map[string]interface{}{
+		"task_name":         "test-task",
+		"schema_name":       "PUBLIC",
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{},
+		"with_grant_option": false,
+	})
+
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		expectReadTaskGrant(mock)
+		err := resources.ReadTaskGrant(d, db)
+		r.NoError(err)
+	})
+
+	roles := d.Get("roles").(*schema.Set)
+	r.True(roles.Contains("test-role-1"))
+	r.True(roles.Contains("test-role-2"))
+	r.Equal(roles.Len(), 2)
+}
+
+func expectReadTaskGrant(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "privilege", "granted_on", "name", "granted_to", "grantee_name", "grant_option", "granted_by",
+	}).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "TASK", "test-task", "ROLE", "test-role-1", false, "bob",
+	).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "TASK", "test-task", "ROLE", "test-role-2", false, "bob",
+	)
+	mock.ExpectQuery(`^SHOW GRANTS ON TASK "test-db"."PUBLIC"."test-task"$`).WillReturnRows(rows)
+}
+
+func TestFutureTaskGrantCreate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"on_future":         true,
+		"schema_name":       "PUBLIC",
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{"test-role-1", "test-role-2"},
+		"with_grant_option": true,
+	}
+	d := schema.TestResourceDataRaw(t, resources.TaskGrant().Resource.Schema, in)
+	r.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE TASKS IN SCHEMA "test-db"."PUBLIC" TO ROLE "test-role-1" WITH GRANT OPTION$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE TASKS IN SCHEMA "test-db"."PUBLIC" TO ROLE "test-role-2" WITH GRANT OPTION$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadFutureTaskGrant(mock)
+		err := resources.CreateTaskGrant(d, db)
+		r.NoError(err)
+	})
+
+	b := require.New(t)
+
+	in = map[string]interface{}{
+		"on_future":         true,
+		"database_name":     "test-db",
+		"privilege":         "OPERATE",
+		"roles":             []interface{}{"test-role-1", "test-role-2"},
+		"with_grant_option": false,
+	}
+	d = schema.TestResourceDataRaw(t, resources.TaskGrant().Resource.Schema, in)
+	b.NotNil(d)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE TASKS IN DATABASE "test-db" TO ROLE "test-role-1"$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(
+			`^GRANT OPERATE ON FUTURE TASKS IN DATABASE "test-db" TO ROLE "test-role-2"$`,
+		).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectReadFutureTaskDatabaseGrant(mock)
+		err := resources.CreateTaskGrant(d, db)
+		b.NoError(err)
+	})
+}
+
+func expectReadFutureTaskGrant(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "privilege", "grant_on", "name", "grant_to", "grantee_name", "grant_option",
+	}).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "TASK", "test-db.PUBLIC.<SCHEMA>", "ROLE", "test-role-1", false,
+	).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "TASK", "test-db.PUBLIC.<SCHEMA>", "ROLE", "test-role-2", false,
+	)
+	mock.ExpectQuery(`^SHOW FUTURE GRANTS IN SCHEMA "test-db"."PUBLIC"$`).WillReturnRows(rows)
+}
+
+func expectReadFutureTaskDatabaseGrant(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{
+		"created_on", "privilege", "grant_on", "name", "grant_to", "grantee_name", "grant_option",
+	}).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "TASK", "test-db.<SCHEMA>", "ROLE", "test-role-1", false,
+	).AddRow(
+		time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), "OPERATE", "TASK", "test-db.<SCHEMA>", "ROLE", "test-role-2", false,
+	)
+	mock.ExpectQuery(`^SHOW FUTURE GRANTS IN DATABASE "test-db"$`).WillReturnRows(rows)
+}