Notifies with validation (#5)

modconfig/flowpipe_integration.go

@@ -90,6 +90,10 @@ func (i *EmailIntegration) GetType() string {
 	return i.Type
 }
 
+func (i *EmailIntegration) CtyValue() (cty.Value, error) {
+	return GetCtyValue(i)
+}
+
 func (i *EmailIntegration) Equals(other *EmailIntegration) bool {
 	if i == nil && other == nil {
 		return true

modconfig/flowpipe_pipeline_step.go

@@ -287,8 +287,56 @@ func (ec *ErrorConfig) Equals(other *ErrorConfig) bool {
 }
 
 type PipelineStepInputNotify struct {
-	Channel     *string   `json:"channel,omitempty" hcl:"channel,optional" cty:"channel"`
-	Integration cty.Value `json:"-" hcl:"integration,optional" cty:"integration"`
+	Channel     *string   `json:"channel,omitempty" hcl:"channel,optional"`
+	To          *string   `json:"to,omitempty" hcl:"to,optional"`
+	Integration cty.Value `json:"-" hcl:"integration"`
+}
+
+func (p *PipelineStepInputNotify) Validate() hcl.Diagnostics {
+	var diags hcl.Diagnostics
+
+	if p.Channel == nil && p.To == nil {
+		diags = append(diags, &hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "Either channel or to  must be specified",
+		})
+	}
+
+	if p.Integration == cty.NilVal {
+		diags = append(diags, &hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "integration must be specified",
+		})
+	}
+
+	if !p.Integration.Type().IsObjectType() {
+		diags = append(diags, &hcl.Diagnostic{
+			Severity: hcl.DiagError,
+			Summary:  "integration must be a map",
+		})
+	} else {
+		integrationMap := p.Integration.AsValueMap()
+		integrationType := integrationMap["type"].AsString()
+		if integrationType == "slack" && p.Channel == nil {
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "channel must be specified for slack integration",
+			})
+
+		} else if integrationType == "email" && p.To == nil {
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  "to must be specified for email integration",
+			})
+		} else if integrationType != "slack" && integrationType != "email" {
+			diags = append(diags, &hcl.Diagnostic{
+				Severity: hcl.DiagError,
+				Summary:  fmt.Sprintf("unsupported integration type %s", integrationType),
+			})
+		}
+	}
+
+	return diags
 }
 
 // A common base struct that all pipeline steps must embed
@@ -2217,8 +2265,13 @@ type PipelineStepInput struct {
 	Type *string `json:"type" cty:"type"`
 	// end Integrated 2023 temporary setup
 
-	Notify   *PipelineStepInputNotify
-	Notifies cty.Value // TODO: this is odd, we should just have cty.Value for notify as well I think?
+	Notify *PipelineStepInputNotify
+
+	// This is odd but notifies is an attribute that can be set dynamically, i.e. input from another step. It's also a list of complex
+	// object, so we've decided for now it's the quickest way to implement it.
+	//
+	// We are unable to parse the Notifies for invalid
+	Notifies cty.Value
 }
 
 func (p *PipelineStepInput) Equals(iOther IPipelineStep) bool {
@@ -2854,6 +2907,13 @@ func (p *PipelineStepInput) Validate() hcl.Diagnostics {
 		}
 	}
 
+	if p.Notify != nil {
+		moreDiags := p.Notify.Validate()
+		if len(moreDiags) > 0 {
+			diags = append(diags, moreDiags...)
+		}
+	}
+
 	return diags
 }
 

parse/pipeline_decode.go

@@ -17,6 +17,7 @@ import (
 )
 
 func decodeStep(mod *modconfig.Mod, block *hcl.Block, parseCtx *ModParseContext, pipelineHcl *modconfig.Pipeline) (modconfig.IPipelineStep, hcl.Diagnostics) {
+
 	stepType := block.Labels[0]
 	stepName := block.Labels[1]
 

tests/flowpipe_invalid_tests/invalid_test.go

@@ -77,6 +77,16 @@ var tests = []testSetup{
 		file:          "./pipelines/approval_notify_and_notifies.fp",
 		containsError: "Notify and Notifies attributes are mutualy exclusive: input.input",
 	},
+	{
+		title:         "invalid approval - slack notify missing channel",
+		file:          "./pipelines/approval_invalid_notify_slack.fp",
+		containsError: "channel must be specified for slack integration",
+	},
+	{
+		title:         "invalid approval - email notify missing to",
+		file:          "./pipelines/approval_invalid_notify_email.fp",
+		containsError: "to must be specified for email integration",
+	},
 }
 
 // Simple invalid test. Only single file resources can be evaluated here. This test is unaable to test

tests/flowpipe_invalid_tests/pipelines/approval_invalid_notify_attribute.fp

@@ -0,0 +1,15 @@
+integration "slack" "my_slack_app" {
+  token           = "xoxp-111111"
+
+  # optional - if you want to verify the source
+  signing_secret  = "Q#$$#@#$$#W"
+}
+
+pipeline "approval" {
+  step "input" "input" {
+    notify {
+      integration = integration.slack.my_slack_app
+      bad_attribute = "foo"
+    }
+  }
+}

tests/flowpipe_invalid_tests/pipelines/approval_invalid_notify_email.fp

@@ -0,0 +1,14 @@
+integration "email" "email_integration" {
+  smtp_host = "foo bar baz"
+  default_subject = "bar foo baz"
+  smtp_username = "baz bar foo"
+}
+
+pipeline "approval" {
+  step "input" "input" {
+    notify {
+      integration = integration.email.email_integration
+      channel = "foo"
+    }
+  }
+}

tests/flowpipe_invalid_tests/pipelines/approval_invalid_notify_slack.fp

@@ -0,0 +1,15 @@
+integration "slack" "my_slack_app" {
+  token           = "xoxp-111111"
+
+  # optional - if you want to verify the source
+  signing_secret  = "Q#$$#@#$$#W"
+}
+
+pipeline "approval" {
+  step "input" "input" {
+    notify {
+      integration = integration.slack.my_slack_app
+      to = "foo"
+    }
+  }
+}

tests/flowpipe_parsing_tests/approval_notifies_test.go

@@ -20,7 +20,7 @@ func TestApprovalNotifies(t *testing.T) {
 		return
 	}
 
-	assert.Equal(3, len(mod.ResourceMaps.Integrations))
+	assert.Equal(2, len(mod.ResourceMaps.Integrations))
 
 	integration := mod.ResourceMaps.Integrations["local.integration.slack.my_slack_app"]
 	if integration == nil {

tests/flowpipe_parsing_tests/approval_test.go

@@ -63,11 +63,26 @@ func TestApproval(t *testing.T) {
 	assert.NotNil(integrationMap)
 	assert.Equal("xoxp-111111", integrationMap["token"].AsString())
 
-	assert.Equal("remove this after integrated", *inputStep.Token)
-
 	inputsAfterEval, err := inputStep.GetInputs(&hcl.EvalContext{})
 	// the notify should override the inline definition (the inline definition should not be there after integrated 2023)
 	assert.Nil(err)
 
 	assert.Equal("xoxp-111111", inputsAfterEval["token"].(string))
+
+	pipeline = mod.ResourceMaps.Pipelines["local.pipeline.approval_email"]
+	if pipeline == nil {
+		assert.Fail("Pipeline not found")
+		return
+	}
+
+	inputStep, ok = pipeline.Steps[0].(*modconfig.PipelineStepInput)
+	if !ok {
+		assert.Fail("Pipeline step not found")
+		return
+	}
+
+	assert.Equal("input_email", inputStep.Name)
+	assert.NotNil(inputStep.Notify)
+	assert.Equal("[email protected]", *inputStep.Notify.To)
+
 }

tests/flowpipe_parsing_tests/pipelines/approval.fp

@@ -22,21 +22,32 @@ integration "email" "email_integration" {
 
 pipeline "approval" {
   step "input" "input" {
-    token = "remove this after integrated"
+
     notify {
       integration = integration.slack.my_slack_app
       channel = "foo"
     }
   }
 }
 
+pipeline "approval_email" {
+  step "input" "input_email" {
+
+    notify {
+      integration = integration.email.email_integration
+      to = "[email protected]"
+    }
+  }
+}
+
+// TODO: param doesn't work yet
 pipeline "approval_dynamic_integration" {
 
   param "integration_param" {
   }
 
   step "input" "input" {
-    token = "remove this after integrated"
+
     notify {
       integration = integration.slack.my_slack_app
       channel = "foo"

tests/flowpipe_parsing_tests/pipelines/approval_notifies.fp

@@ -2,14 +2,6 @@
 
 integration "slack" "my_slack_app" {
   token           = "xoxp-111111"
-
-  # optional - if you want to verify the source
-  signing_secret  = "Q#$$#@#$$#W"
-}
-
-integration "slack" "my_slack_app_two" {
-  token           = "xoxp-111111"
-
   # optional - if you want to verify the source
   signing_secret  = "Q#$$#@#$$#W"
 }
@@ -36,13 +28,14 @@ pipeline "approval_with_notifies" {
       {
         integration = integration.slack.my_slack_app
         channel = "foo"
+        to = "[email protected]"
         # channel = param.slack_channel
-
         # if      = param.slack_integration == null ? false : true
       },
       {
         integration = integration.email.email_integration
         to = "[email protected]"
+        channel = "bar"
       }
     ]
   }
@@ -68,7 +61,6 @@ pipeline "approval_with_notifies_depend_another_step" {
         integration = integration.slack.my_slack_app
         channel = step.echo.echo.text
         # channel = param.slack_channel
-
         # if      = param.slack_integration == null ? false : true
       },
       {