acceptance_test.go

// Copyright © 2022 Meroxa, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package hubspot

import (
	"context"
	"net/http"
	"os"
	"testing"
	"time"

	"github.com/brianvoe/gofakeit/v6"
	"github.com/conduitio-labs/conduit-connector-hubspot/config"
	"github.com/conduitio-labs/conduit-connector-hubspot/hubspot"
	"github.com/conduitio-labs/conduit-connector-hubspot/source"
	"github.com/conduitio/conduit-commons/opencdc"
	sdk "github.com/conduitio/conduit-connector-sdk"
	"go.uber.org/goleak"
)

// testResource is a test resource that we use for integration tests.
const testResource = "crm.contacts"

// acceptanceTestTimeout is a timeout used for both read and write.
const acceptanceTestTimeout = time.Second * 20

// The list of HubSpot field names that are used in acceptance tests.
const (
	testIDFieldName        = "id"
	testEmailFieldName     = "email"
	testFirstNameFieldName = "firstname"
	testLastNameFieldName  = "lastname"
)

// testAccessToken holds a value of HUBSPOT_ACCESS_TOKEN which is required for the acceptance tests.
var testAccessToken = os.Getenv("HUBSPOT_ACCESS_TOKEN")

type driver struct {
	sdk.ConfigurableAcceptanceTestDriver

	hubspotClient *hubspot.Client
}

// GenerateRecord overides the [sdk.ConfigurableAcceptanceTestDriver] [GenerateRecord] method.
func (d driver) GenerateRecord(t *testing.T, operation opencdc.Operation) opencdc.Record {
	t.Helper()

	return opencdc.Record{
		Operation: operation,
		Payload: opencdc.Change{
			After: opencdc.StructuredData{
				"properties": map[string]any{
					testFirstNameFieldName: gofakeit.FirstName(),
					testLastNameFieldName:  gofakeit.LastName(),
					testEmailFieldName:     gofakeit.Email(),
				},
			},
		},
	}
}

// ReadFromDestination overrides the [sdk.ConfigurableAcceptanceTestDriver] [ReadFromDestination] method.
// It removes some redundant fields that are unknown when we insert data.
func (d driver) ReadFromDestination(t *testing.T, records []opencdc.Record) []opencdc.Record {
	t.Helper()

	// the search endpoint lags behind, query it and wait for all results to disappear
	hubspotClient := hubspot.NewClient(testAccessToken, &http.Client{
		Timeout: acceptanceTestTimeout,
	})
	for i := 0; i < 5; i++ {
		time.Sleep(time.Second * 5)
		listResp, err := hubspotClient.Search(context.Background(), testResource, &hubspot.SearchRequest{})
		if err != nil {
			t.Fatalf("error querying search: %v", err)
		}
		if listResp.Total >= len(records) {
			break
		}
	}

	newRecords := d.ConfigurableAcceptanceTestDriver.ReadFromDestination(t, records)

	out := make([]opencdc.Record, len(newRecords))

	for i, newRecord := range newRecords {
		newRecordStructuredPayload, ok := newRecord.Payload.After.(opencdc.StructuredData)
		if !ok {
			// this shouldn't happen
			t.Fatalf("expected payload to contain opencdc.StructuredData, got %T", newRecord.Payload.After)
		}

		out[i] = opencdc.Record{
			Operation: newRecord.Operation,
			Position:  newRecord.Position,
			Payload: opencdc.Change{
				After: opencdc.StructuredData{
					"properties": map[string]any{
						testIDFieldName:        newRecordStructuredPayload[testIDFieldName],
						testFirstNameFieldName: newRecordStructuredPayload[testFirstNameFieldName],
						testLastNameFieldName:  newRecordStructuredPayload[testLastNameFieldName],
						testEmailFieldName:     newRecordStructuredPayload[testEmailFieldName],
					},
				},
			},
		}
	}

	return records
}

// WriteToSource overrides the [sdk.ConfigurableAcceptanceTestDriver] [WriteToSource] method.
// It takes items from HubSpot and returns them as [opencdc.Record]s, because Source returns fields
// that are unknown when we insert them.
func (d driver) WriteToSource(t *testing.T, records []opencdc.Record) []opencdc.Record {
	t.Helper()

	newRecords := d.ConfigurableAcceptanceTestDriver.WriteToSource(t, records)

	listResp, err := d.hubspotClient.List(context.Background(), testResource, nil)
	if err != nil {
		t.Errorf("list items: %v", err)
	}

	// create a map that holds emails and corresponding list response results
	listRespMap := make(map[string]hubspot.ListResponseResult)

	for _, listRespResult := range listResp.Results {
		listRespResultName, ok := listRespResult["properties"].(map[string]any)[testEmailFieldName].(string)
		if !ok {
			t.Errorf("list resp result email is not a string or doesn't exist")
		}

		listRespMap[listRespResultName] = listRespResult
	}

	// fill records payload with the newly created HubSpot items properties
	for i := range newRecords {
		recordPayloadAfter, ok := newRecords[i].Payload.After.(opencdc.StructuredData)
		if !ok {
			t.Fatal("record's payload after is not structured")
		}

		recordPayloadAfterProperties, ok := recordPayloadAfter["properties"].(map[string]any)
		if !ok {
			t.Fatal("record's payload after does not contain properties or it's not a map")
		}

		recordPayloadAfterEmail, ok := recordPayloadAfterProperties[testEmailFieldName].(string)
		if !ok {
			t.Fatal("record's payload after email is not a string or doesn't exist")
		}

		listRespResult, ok := listRespMap[recordPayloadAfterEmail]
		if !ok {
			t.Fatal("can't find a list resp result by email")
		}

		newRecords[i].Key = opencdc.StructuredData{hubspot.ResultsFieldID: listRespResult["id"]}
		newRecords[i].Payload.After = opencdc.StructuredData(listRespResult)
	}

	return newRecords
}

func TestAcceptance(t *testing.T) {
	if testAccessToken == "" {
		t.Skip("HUBSPOT_ACCESS_TOKEN env var must be set")
	}

	cfg := map[string]string{
		config.KeyAccessToken:         testAccessToken,
		config.KeyResource:            testResource,
		source.ConfigKeyPollingPeriod: "1s",
	}

	sdk.AcceptanceTest(t, driver{
		ConfigurableAcceptanceTestDriver: sdk.ConfigurableAcceptanceTestDriver{
			Config: sdk.ConfigurableAcceptanceTestDriverConfig{
				Connector:         Connector,
				SourceConfig:      cfg,
				DestinationConfig: cfg,
				GoleakOptions: []goleak.Option{
					// these leaks mainly come from the go-retryablehttp
					goleak.IgnoreTopFunction("internal/poll.runtime_pollWait"),
					goleak.IgnoreTopFunction("net/http.(*persistConn).writeLoop"),
				},
				AfterTest:    afterTest,
				ReadTimeout:  acceptanceTestTimeout,
				WriteTimeout: acceptanceTestTimeout,
				Skip: []string{
					// ResumeAtPosition tests are failing, it's hard to get a
					// stable test run with the hubspot API as the search API
					// has quite a delay between writes and reads. Also, the
					// "updatedAt" field gets updated randomly even after the
					// resource has been created.
					// To top it off, the API has a rate limit which we are
					// hitting, so we can easily hit the test timeout.
					"TestSource_Open_ResumeAtPositionCDC",
					"TestSource_Open_ResumeAtPositionSnapshot",
					// TestDestination_Write_Success is also flaky, although
					// succeeds more often than not.
				},
			},
		},
		hubspotClient: hubspot.NewClient(testAccessToken, &http.Client{
			Timeout: acceptanceTestTimeout,
		}),
	})
}

// afterTest is a test helper that deletes all resource items after each test.
func afterTest(t *testing.T) {
	t.Helper()
	ctx := context.Background()

	hubspotClient := hubspot.NewClient(testAccessToken, &http.Client{
		Timeout: acceptanceTestTimeout,
	})

	listResp, err := hubspotClient.List(ctx, testResource, nil)
	if err != nil {
		t.Errorf("hubspot list: %v", err)
	}

	for _, item := range listResp.Results {
		itemID, ok := item[testIDFieldName].(string)
		if !ok {
			t.Errorf("item is not a string")
		}

		if err := hubspotClient.Delete(context.Background(), testResource, itemID); err != nil {
			t.Errorf("hubspot delete: %v", err)
		}
	}

	// the search endpoint lags behind, query it and wait for all results to disappear
	searchEmpty := false
	for i := 0; i < 5; i++ {
		listResp, err := hubspotClient.Search(ctx, testResource, &hubspot.SearchRequest{})
		if err != nil {
			t.Fatalf("error querying search: %v", err)
		}
		if listResp.Total == 0 {
			searchEmpty = true
			break
		}
		time.Sleep(time.Second * 5)
	}

	if !searchEmpty {
		t.Log("WARNING: hubspot still returns data in the search endpoint, next tests might fail because of this")
	}
}