Review

pkg/acceptance/bettertestspoc/assert/objectassert/stream_snowflake_ext.go

@@ -1,13 +1,15 @@
 package objectassert
 
 import (
+	"errors"
 	"fmt"
+	"slices"
 	"testing"
 
 	"github.com/Snowflake-Labs/terraform-provider-snowflake/pkg/sdk"
 )
 
-func (s *StreamAssert) HasTableId(expected string) *StreamAssert {
+func (s *StreamAssert) HasTableId(expected sdk.SchemaObjectIdentifier) *StreamAssert {
 	s.AddAssertion(func(t *testing.T, o *sdk.Stream) error {
 		t.Helper()
 		if o.TableName == nil {
@@ -17,7 +19,7 @@ func (s *StreamAssert) HasTableId(expected string) *StreamAssert {
 		if err != nil {
 			return err
 		}
-		if gotTableId.FullyQualifiedName() != expected {
+		if gotTableId.FullyQualifiedName() != expected.FullyQualifiedName() {
 			return fmt.Errorf("expected table name: %v; got: %v", expected, *o.TableName)
 		}
 		return nil
@@ -53,21 +55,21 @@ func (s *StreamAssert) HasSourceType(expected sdk.StreamSourceType) *StreamAsser
 	return s
 }
 
-func (s *StreamAssert) HasBaseTables(expected []sdk.SchemaObjectIdentifier) *StreamAssert {
+func (s *StreamAssert) HasBaseTables(expected ...sdk.SchemaObjectIdentifier) *StreamAssert {
 	s.AddAssertion(func(t *testing.T, o *sdk.Stream) error {
 		t.Helper()
-		if o.BaseTables == nil {
-			return fmt.Errorf("expected base tables to have value; got: nil")
-		}
 		if len(o.BaseTables) != len(expected) {
 			return fmt.Errorf("expected base tables length: %v; got: %v", len(expected), len(o.BaseTables))
 		}
-		for i := range o.BaseTables {
-			if o.BaseTables[i].FullyQualifiedName() != expected[i].FullyQualifiedName() {
-				return fmt.Errorf("expected base table id: %v; got: %v", expected[i], o.BaseTables[i])
+		var errs []error
+		for _, wantId := range expected {
+			if !slices.ContainsFunc(o.BaseTables, func(gotId sdk.SchemaObjectIdentifier) bool {
+				return wantId.FullyQualifiedName() == gotId.FullyQualifiedName()
+			}) {
+				errs = append(errs, fmt.Errorf("expected id: %s, to be in the list ids: %v", wantId.FullyQualifiedName(), o.BaseTables))
 			}
 		}
-		return nil
+		return errors.Join(errs...)
 	})
 	return s
 }

pkg/resources/stream_common.go

@@ -58,3 +58,79 @@ func DeleteStreamContext(ctx context.Context, d *schema.ResourceData, meta any)
 	d.SetId("")
 	return nil
 }
+
+var atSchema = &schema.Schema{
+	Type:        schema.TypeList,
+	Optional:    true,
+	MaxItems:    1,
+	Description: "This field specifies that the request is inclusive of any changes made by a statement or transaction with a timestamp equal to the specified parameter. Due to Snowflake limitations, the provider does not detect external changes on this field.",
+	Elem: &schema.Resource{
+		Schema: map[string]*schema.Schema{
+			"timestamp": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				Description:  "Specifies an exact date and time to use for Time Travel. The value must be explicitly cast to a TIMESTAMP, TIMESTAMP_LTZ, TIMESTAMP_NTZ, or TIMESTAMP_TZ data type.",
+				ExactlyOneOf: []string{"at.0.timestamp", "at.0.offset", "at.0.statement", "at.0.stream"},
+			},
+			"offset": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				Description:  "Specifies the difference in seconds from the current time to use for Time Travel, in the form -N where N can be an integer or arithmetic expression (e.g. -120 is 120 seconds, -30*60 is 1800 seconds or 30 minutes).",
+				ExactlyOneOf: []string{"at.0.timestamp", "at.0.offset", "at.0.statement", "at.0.stream"},
+			},
+			"statement": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				Description:  "Specifies the query ID of a statement to use as the reference point for Time Travel. This parameter supports any statement of one of the following types: DML (e.g. INSERT, UPDATE, DELETE), TCL (BEGIN, COMMIT transaction), SELECT.",
+				ExactlyOneOf: []string{"at.0.timestamp", "at.0.offset", "at.0.statement", "at.0.stream"},
+			},
+			"stream": {
+				Type:             schema.TypeString,
+				Optional:         true,
+				Description:      "Specifies the identifier (i.e. name) for an existing stream on the queried table or view. The current offset in the stream is used as the AT point in time for returning change data for the source object.",
+				ExactlyOneOf:     []string{"at.0.timestamp", "at.0.offset", "at.0.statement", "at.0.stream"},
+				DiffSuppressFunc: suppressIdentifierQuoting,
+				ValidateDiagFunc: IsValidIdentifier[sdk.SchemaObjectIdentifier](),
+			},
+		},
+	},
+	ConflictsWith: []string{"before"},
+}
+
+var beforeSchema = &schema.Schema{
+	Type:        schema.TypeList,
+	Optional:    true,
+	MaxItems:    1,
+	Description: "This field specifies that the request refers to a point immediately preceding the specified parameter. This point in time is just before the statement, identified by its query ID, is completed.  Due to Snowflake limitations, the provider does not detect external changes on this field.",
+	Elem: &schema.Resource{
+		Schema: map[string]*schema.Schema{
+			"timestamp": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				Description:  "Specifies an exact date and time to use for Time Travel. The value must be explicitly cast to a TIMESTAMP, TIMESTAMP_LTZ, TIMESTAMP_NTZ, or TIMESTAMP_TZ data type.",
+				ExactlyOneOf: []string{"before.0.timestamp", "before.0.offset", "before.0.statement", "before.0.stream"},
+			},
+			"offset": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				Description:  "Specifies the difference in seconds from the current time to use for Time Travel, in the form -N where N can be an integer or arithmetic expression (e.g. -120 is 120 seconds, -30*60 is 1800 seconds or 30 minutes).",
+				ExactlyOneOf: []string{"before.0.timestamp", "before.0.offset", "before.0.statement", "before.0.stream"},
+			},
+			"statement": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				Description:  "Specifies the query ID of a statement to use as the reference point for Time Travel. This parameter supports any statement of one of the following types: DML (e.g. INSERT, UPDATE, DELETE), TCL (BEGIN, COMMIT transaction), SELECT.",
+				ExactlyOneOf: []string{"before.0.timestamp", "before.0.offset", "before.0.statement", "before.0.stream"},
+			},
+			"stream": {
+				Type:             schema.TypeString,
+				Optional:         true,
+				Description:      "Specifies the identifier (i.e. name) for an existing stream on the queried table or view. The current offset in the stream is used as the AT point in time for returning change data for the source object.",
+				ExactlyOneOf:     []string{"before.0.timestamp", "before.0.offset", "before.0.statement", "before.0.stream"},
+				DiffSuppressFunc: suppressIdentifierQuoting,
+				ValidateDiagFunc: IsValidIdentifier[sdk.SchemaObjectIdentifier](),
+			},
+		},
+	},
+	ConflictsWith: []string{"at"},
+}

pkg/resources/stream_on_table.go

@@ -76,8 +76,8 @@ var streamOnTableSchema = map[string]*schema.Schema{
 		Optional:    true,
 		Description: "Specifies a comment for the stream.",
 	},
-	AtAttributeName:     schemas.AtSchema,
-	BeforeAttributeName: schemas.BeforeSchema,
+	AtAttributeName:     atSchema,
+	BeforeAttributeName: beforeSchema,
 	ShowOutputAttributeName: {
 		Type:        schema.TypeList,
 		Computed:    true,
@@ -139,7 +139,7 @@ func ImportStreamOnTable(ctx context.Context, d *schema.ResourceData, meta any)
 	if err := d.Set("schema", id.SchemaName()); err != nil {
 		return nil, err
 	}
-	if err := d.Set("append_only", booleanStringFromBool(*v.Mode == sdk.StreamModeAppendOnly)); err != nil {
+	if err := d.Set("append_only", booleanStringFromBool(v.IsAppendOnly())); err != nil {
 		return nil, err
 	}
 	return []*schema.ResourceData{d}, nil
@@ -253,7 +253,7 @@ func ReadStreamOnTable(withExternalChangesMarking bool) schema.ReadContextFunc {
 				mode = *stream.Mode
 			}
 			if err = handleExternalChangesToObjectInShow(d,
-				showMapping{"mode", "append_only", string(mode), booleanStringFromBool(mode == sdk.StreamModeAppendOnly), nil},
+				showMapping{"mode", "append_only", string(mode), booleanStringFromBool(stream.IsAppendOnly()), nil},
 			); err != nil {
 				return diag.FromErr(err)
 			}