Repository: amp-labs/connectors
Pull request #1663 (2025-05-17 00:30:13)

[CON-1682] feat(salesforce): AccountEngagement - Read
# Testing
## Emails
Note: you can query dot seperated fields to get nested fields, but they won't appear under the fields only under raw unless we need to change that.
![image](https://github.com/user-attachments/assets/bd7f0c4a-c2e8-4481-ae8e-1b76c1ffb19a)


Cobalt0s: > [!WARNING]
> <b>This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1663?utm_source=stack-comment-downstack-mergeability-warning" >on Graphite</a>.</b>
> <a href="https://graphite.dev/docs/merge-pull-requests">Learn more</a>

* **#1663** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1663?utm_source=stack-comment-icon" target="_blank"><img src="https://static.graphite.dev/graphite-32x32-black.png" alt="Graphite" width="10px" height="10px"/></a> 👈 <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1663?utm_source=stack-comment-view-in-graphite" target="_blank">(View in Graphite)</a>
* **#1662** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/1662?utm_source=stack-comment-icon" target="_blank"><img src="https://static.graphite.dev/graphite-32x32-black.png" alt="Graphite" width="10px" height="10px"/></a>
* `main`


<h2></h2>

This stack of pull requests is managed by <a href="https://graphite.dev?utm-source=stack-comment"><b>Graphite</b></a>. Learn more about <a href="https://stacking.dev/?utm_source=stack-comment">stacking</a>.
<!-- Current dependencies on/for this PR: -->

Patch:
diff --git a/common/paramsbuilder/metadata.go b/common/paramsbuilder/metadata.go
index 3060cb5e..7f4d90ae 100644
--- a/common/paramsbuilder/metadata.go
+++ b/common/paramsbuilder/metadata.go
@@ -60,3 +60,7 @@ func (m *Metadata) GetCatalogVars() []catalogreplacer.CustomCatalogVariable {
 
 	return result
 }
+
+func (m *Metadata) Value(key string) string {
+	return m.Map[key]
+}
diff --git a/internal/datautils/timing.go b/internal/datautils/timing.go
index fa43f58b..79282187 100644
--- a/internal/datautils/timing.go
+++ b/internal/datautils/timing.go
@@ -26,3 +26,8 @@ func (timing) FormatRFC3339inUTCWithMilliseconds(input time.Time) string {
 
 	return utcTime.Format("2006-01-02T15:04:05.000Z")
 }
+
+// FormatRFC3339WithOffset is RFC3339 but includes time zone offset.
+func (timing) FormatRFC3339WithOffset(input time.Time) string {
+	return input.Format("2006-01-02T15:04:05-07:00")
+}
diff --git a/providers/salesforce.go b/providers/salesforce.go
index 93aa6877..76c7c6f7 100644
--- a/providers/salesforce.go
+++ b/providers/salesforce.go
@@ -123,6 +123,15 @@ func init() { // nolint:funlen
 					DisplayName: "Subdomain",
 					DocsURL:     "https://help.salesforce.com/s/articleView?language=en_US&id=sf.faq_domain_name_what.htm&type=5",
 				},
+				{
+					Name:        "businessUnitId",
+					DisplayName: "Business Unit ID",
+					DocsURL:     "https://help.salesforce.com/s/articleView?id=000381973&type=1",
+					ModuleDependencies: &ModuleDependencies{
+						ModuleSalesforceAccountEngagement:     {},
+						ModuleSalesforceAccountEngagementDemo: {},
+					},
+				},
 			},
 		},
 	})
diff --git a/providers/salesforce/connector.go b/providers/salesforce/connector.go
index 7c750718..dd0ffdd8 100644
--- a/providers/salesforce/connector.go
+++ b/providers/salesforce/connector.go
@@ -61,7 +61,7 @@ func NewConnector(opts ...Option) (conn *Connector, outErr error) {
 
 	conn.moduleInfo = conn.providerInfo.ReadModuleInfo(conn.moduleID)
 
-	conn.setBaseURL(conn.providerInfo.BaseURL)
+	conn.setCRMBaseURL(conn.providerInfo.BaseURL)
 	conn.Client.HTTPClient.ErrorHandler = interpreter.ErrorHandler{
 		JSON: &interpreter.DirectFaultyResponder{Callback: conn.interpretJSONError},
 		XML:  &interpreter.DirectFaultyResponder{Callback: conn.interpretXMLError},
@@ -72,10 +72,15 @@ func NewConnector(opts ...Option) (conn *Connector, outErr error) {
 	// Read/Write/ListObjectMetadata will delegate to this adapter.
 	moduleID := params.Module.Selection.ID
 	if isPardotModule(moduleID) {
-		conn.pardotAdapter, err = pardot.NewAdapter(conn.Client, conn.moduleInfo, params.Metadata.Map)
+		conn.pardotAdapter, err = pardot.NewAdapter(conn.Client, conn.moduleInfo,
+			params.Value(pardot.MetadataKeyBusinessUnitID),
+		)
 		if err != nil {
 			return nil, err
 		}
+
+		// Override error handler. Pardot has different format from Standard Salesforce.
+		conn.Client.HTTPClient.ErrorHandler = pardot.ErrorHandlerFunc
 	}
 
 	return conn, nil
@@ -117,9 +122,20 @@ func (c *Connector) getURIPartSobjectsDescribe(objectName string) (*urlbuilder.U
 	return urlbuilder.New(uriSobjects, objectName, "describe")
 }
 
+func (c *Connector) setCRMBaseURL(newURL string) {
+	c.BaseURL = newURL
+	c.Client.HTTPClient.Base = newURL
+}
+
+// Warning: this method is to be used by mock tests.
+// The code is in the intermediate state and improved URL storage and handling is anticipated.
 func (c *Connector) setBaseURL(newURL string) {
 	c.BaseURL = newURL
 	c.Client.HTTPClient.Base = newURL
+
+	if c.pardotAdapter != nil {
+		c.pardotAdapter.BaseURL = newURL
+	}
 }
 
 func (c *Connector) isPardotModule() bool {
diff --git a/providers/salesforce/internal/pardot/adapter.go b/providers/salesforce/internal/pardot/adapter.go
index d35547ec..9831ebe2 100644
--- a/providers/salesforce/internal/pardot/adapter.go
+++ b/providers/salesforce/internal/pardot/adapter.go
@@ -2,19 +2,28 @@ package pardot
 
 import (
 	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/common/urlbuilder"
 	"github.com/amp-labs/connectors/providers"
 )
 
+const MetadataKeyBusinessUnitID = "businessUnitId"
+
 type Adapter struct {
-	Client  *common.JSONHTTPClient
-	BaseURL string
+	Client         *common.JSONHTTPClient
+	BaseURL        string
+	BusinessUnitID string
 }
 
 func NewAdapter(
-	client *common.JSONHTTPClient, info *providers.ModuleInfo, metadata map[string]string,
+	client *common.JSONHTTPClient, info *providers.ModuleInfo, businessUnitID string,
 ) (*Adapter, error) {
 	return &Adapter{
-		Client:  client,
-		BaseURL: info.BaseURL,
+		Client:         client,
+		BaseURL:        info.BaseURL,
+		BusinessUnitID: businessUnitID,
 	}, nil
 }
+
+func (a *Adapter) getURL(objectName string) (*urlbuilder.URL, error) {
+	return urlbuilder.New(a.BaseURL, "api/v5/objects", objectName)
+}
diff --git a/providers/salesforce/internal/pardot/errors.go b/providers/salesforce/internal/pardot/errors.go
new file mode 100644
index 00000000..456311b3
--- /dev/null
+++ b/providers/salesforce/internal/pardot/errors.go
@@ -0,0 +1,32 @@
+package pardot
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/amp-labs/connectors/common/interpreter"
+)
+
+func ErrorHandlerFunc(rsp *http.Response, body []byte) error {
+	return interpreter.ErrorHandler{
+		JSON: interpreter.NewFaultyResponder(errorFormats, nil),
+	}.Handle(rsp, body)
+}
+
+var errorFormats = interpreter.NewFormatSwitch( // nolint:gochecknoglobals
+	[]interpreter.FormatTemplate{
+		{
+			MustKeys: nil,
+			Template: func() interpreter.ErrorDescriptor { return &ResponseError{} },
+		},
+	}...,
+)
+
+type ResponseError struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
+func (r ResponseError) CombineErr(base error) error {
+	return fmt.Errorf("%w: %v", base, r.Message)
+}
diff --git a/providers/salesforce/internal/pardot/operations.go b/providers/salesforce/internal/pardot/operations.go
index 240ad65a..d5d4d057 100644
--- a/providers/salesforce/internal/pardot/operations.go
+++ b/providers/salesforce/internal/pardot/operations.go
@@ -2,13 +2,72 @@ package pardot
 
 import (
 	"context"
+	"strings"
 
 	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/common/urlbuilder"
+	"github.com/amp-labs/connectors/internal/datautils"
+	"github.com/amp-labs/connectors/internal/jsonquery"
+	"github.com/spyzhov/ajson"
 )
 
+// DefaultPageSize is a relative limit. In fact the largest page size allowed is 100,000.
+// Reference:
+// https://developer.salesforce.com/docs/marketing/pardot/guide/version5overview.html#pagination
+const DefaultPageSize = "1000"
+
 func (a *Adapter) Read(ctx context.Context, params common.ReadParams) (*common.ReadResult, error) {
-	// TODO needs implementation.
-	return nil, common.ErrNotImplemented
+	url, err := a.buildReadURL(params)
+	if err != nil {
+		return nil, err
+	}
+
+	rsp, err := a.Client.Get(ctx, url.String(), common.Header{
+		Key:   "Pardot-Business-Unit-Id",
+		Value: a.BusinessUnitID,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return common.ParseResult(
+		rsp,
+		common.ExtractOptionalRecordsFromPath("values"),
+		func(node *ajson.Node) (string, error) {
+			return jsonquery.New(node).StrWithDefault("nextPageUrl", "")
+		},
+		common.GetMarshaledData,
+		params.Fields,
+	)
+}
+
+func (a *Adapter) buildReadURL(params common.ReadParams) (*urlbuilder.URL, error) {
+	if len(params.NextPage) != 0 {
+		return urlbuilder.New(params.NextPage.String())
+	}
+
+	objectNameLower := strings.ToLower(params.ObjectName)
+
+	url, err := a.getURL(objectNameLower)
+	if err != nil {
+		return nil, err
+	}
+
+	url.WithQueryParam("fields", strings.Join(params.Fields.List(), ","))
+	url.WithQueryParam("limit", DefaultPageSize)
+
+	if !params.Since.IsZero() {
+		if query, ok := incrementalQuery[objectNameLower]; ok {
+			url.WithQueryParam(query, datautils.Time.FormatRFC3339WithOffset(params.Since))
+		}
+	}
+
+	return url, nil
+}
+
+// incrementalQuery is a registry of object name to the query parameter used for performing incremental reading.
+var incrementalQuery = map[string]string{ // nolint:gochecknoglobals
+	"emails": "sentAtAfterOrEqualTo",
 }
 
 func (a *Adapter) Write(ctx context.Context, params common.WriteParams) (*common.WriteResult, error) {
diff --git a/providers/salesforce/params.go b/providers/salesforce/params.go
index 66daaa95..97b7c48d 100644
--- a/providers/salesforce/params.go
+++ b/providers/salesforce/params.go
@@ -8,6 +8,7 @@ import (
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/common/paramsbuilder"
 	"github.com/amp-labs/connectors/providers"
+	"github.com/amp-labs/connectors/providers/salesforce/internal/pardot"
 	"golang.org/x/oauth2"
 )
 
@@ -39,7 +40,15 @@ func newParams(opts []Option) (*common.ConnectorParams, error) { // nolint:unuse
 
 func (p parameters) ValidateParams() error {
 	if isPardotModule(p.Module.Selection.ID) {
-		return p.Client.ValidateParams()
+		// Check that business unit id is present.
+		p.Metadata.WithMetadata(p.Metadata.Map, []string{
+			pardot.MetadataKeyBusinessUnitID,
+		})
+
+		return errors.Join(
+			p.Client.ValidateParams(),
+			p.Metadata.ValidateParams(),
+		)
 	}
 
 	return errors.Join(


Test patch:
diff --git a/providers/salesforce/read_test.go b/providers/salesforce/read_test.go
index 78a7c62c..74383e09 100644
--- a/providers/salesforce/read_test.go
+++ b/providers/salesforce/read_test.go
@@ -4,12 +4,14 @@ import (
 	"errors"
 	"net/http"
 	"testing"
+	"time"
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/internal/jsonquery"
 	"github.com/amp-labs/connectors/providers"
 	"github.com/amp-labs/connectors/test/utils/mockutils"
+	"github.com/amp-labs/connectors/test/utils/mockutils/mockcond"
 	"github.com/amp-labs/connectors/test/utils/mockutils/mockserver"
 	"github.com/amp-labs/connectors/test/utils/testroutines"
 	"github.com/amp-labs/connectors/test/utils/testutils"
@@ -139,6 +141,170 @@ func TestRead(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
 	}
 }
 
+func TestReadPardot(t *testing.T) { //nolint:funlen,gocognit,cyclop,maintidx
+	t.Parallel()
+
+	responseMissingHeader := testutils.DataFromFile(t, "pardot/read/emails/err-missing-header.json")
+	responseMissingQuery := testutils.DataFromFile(t, "pardot/read/emails/err-missing-query.json")
+	responseEmailsFirstPage := testutils.DataFromFile(t, "pardot/read/emails/1-first-page.json")
+	responseEmailsEmptyPage := testutils.DataFromFile(t, "pardot/read/emails/2-empty-page.json")
+
+	pardotHeader := http.Header{
+		"Pardot-Business-Unit-Id": []string{"test-business-unit-id"},
+	}
+
+	tests := []testroutines.Read{
+		{
+			Name:         "Read object must be included",
+			Server:       mockserver.Dummy(),
+			ExpectedErrs: []error{common.ErrMissingObjects},
+		},
+		{
+			Name:         "At least one field is requested",
+			Input:        common.ReadParams{ObjectName: "emails"},
+			Server:       mockserver.Dummy(),
+			ExpectedErrs: []error{common.ErrMissingFields},
+		},
+		{
+			Name:  "Error Missing Header message is understood",
+			Input: common.ReadParams{ObjectName: "emails", Fields: connectors.Fields("id")},
+			Server: mockserver.Fixed{
+				Setup:  mockserver.ContentJSON(),
+				Always: mockserver.Response(http.StatusBadRequest, responseMissingHeader),
+			}.Server(),
+			ExpectedErrs: []error{
+				common.ErrBadRequest,
+				errors.New("A required header is missing: Pardot-Business-Unit-Id header not found on request."), // nolint:goerr113
+			},
+		},
+		{
+			Name:  "Error Missing Query message is understood",
+			Input: common.ReadParams{ObjectName: "emails", Fields: connectors.Fields("id")},
+			Server: mockserver.Fixed{
+				Setup:  mockserver.ContentJSON(),
+				Always: mockserver.Response(http.StatusBadRequest, responseMissingQuery),
+			}.Server(),
+			ExpectedErrs: []error{
+				common.ErrBadRequest,
+				errors.New("One or more required parameters are missing: fields"), // nolint:goerr113
+			},
+		},
+		{
+			Name: "Read emails first page",
+			Input: common.ReadParams{
+				ObjectName: "eMaILs",
+				Fields:     connectors.Fields("name", "subject"),
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/api/v5/objects/emails"),
+					mockcond.QueryParam("limit", "1000"),
+					mockcond.Or{
+						mockcond.QueryParam("fields", "name,subject"),
+						mockcond.QueryParam("fields", "subject,name"),
+					},
+					mockcond.Header(pardotHeader),
+				},
+				Then: mockserver.Response(http.StatusOK, responseEmailsFirstPage),
+			}.Server(),
+			Comparator: testroutines.ComparatorSubsetRead,
+			Expected: &common.ReadResult{
+				Rows: 2,
+				Data: []common.ReadResultRow{{
+					Fields: map[string]any{
+						"name":    "Sending first email ever",
+						"subject": "Few Moments Later",
+					},
+					Raw: map[string]any{
+						"id":         float64(34277860),
+						"clientType": "Web",
+						"updatedAt":  "2025-05-16T11:27:08-07:00",
+					},
+				}, {
+					Fields: map[string]any{
+						"name":    "Second email, on the roll",
+						"subject": "Second email, unbelievable",
+					},
+					Raw: map[string]any{
+						"id":         float64(34277863),
+						"clientType": "Web",
+						"updatedAt":  "2025-05-16T11:29:03-07:00",
+					},
+				}},
+				NextPage: "https://pi.demo.pardot.com/api/v5/objects/emails?fields=id,name,subject,clientType,prospectId,listEmailId,updatedAt&nextPageToken=eyJvcmRlckJ5IjoiIiwiZmlsdGVycyI6W10sImxpbWl0IjoxLCJyZXN1bWVWYWx1ZSI6eyJpZCI6MzQyNzc4NjN9LCJwYWdlIjoyLCJyZWNDb3VudCI6MiwiZXhwaXJlVGltZSI6IjIwMjUtMDUtMTZUMjA6Mzc6MTMtMDc6MDAiLCJkZWxldGVkIjpudWxsfQ==", // nolint:lll
+				Done:     false,
+			},
+			ExpectedErrs: nil,
+		},
+		{
+			Name: "Incremental read of emails last empty page",
+			Input: common.ReadParams{
+				ObjectName: "eMaILs",
+				Fields:     connectors.Fields("name"),
+				Since: time.Date(2024, 9, 19, 4, 30, 45, 600,
+					time.FixedZone("UTC-8", -8*60*60)),
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/api/v5/objects/emails"),
+					mockcond.QueryParam("limit", "1000"),
+					mockcond.QueryParam("fields", "name"),
+					mockcond.QueryParam("sentAtAfterOrEqualTo", "2024-09-19T04:30:45-08:00"),
+					mockcond.Header(pardotHeader),
+				},
+				Then: mockserver.Response(http.StatusOK, responseEmailsEmptyPage),
+			}.Server(),
+			Expected: &common.ReadResult{
+				Rows:     0,
+				Data:     []common.ReadResultRow{},
+				NextPage: "",
+				Done:     true,
+			},
+			ExpectedErrs: nil,
+		},
+		{
+			Name: "Read Emails using next page token",
+			Input: common.ReadParams{
+				ObjectName: "eMaILs",
+				Fields:     connectors.Fields("name"),
+				Since: time.Date(2024, 9, 19, 4, 30, 45, 600,
+					time.FixedZone("UTC-8", -8*60*60)),
+				NextPage: testroutines.URLTestServer + "/api/v5/objects/emails?fields=id,name,subject,clientType,prospectId,listEmailId,updatedAt&nextPageToken=eyJvcmRlckJ5IjoiIiwiZmlsdGVycyI6W10sImxpbWl0IjoxLCJyZXN1bWVWYWx1ZSI6eyJpZCI6MzQyNzc4NjN9LCJwYWdlIjoyLCJyZWNDb3VudCI6MiwiZXhwaXJlVGltZSI6IjIwMjUtMDUtMTZUMjA6Mzc6MTMtMDc6MDAiLCJkZWxldGVkIjpudWxsfQ==", // nolint:lll
+			},
+			Server: mockserver.Conditional{
+				Setup: mockserver.ContentJSON(),
+				If: mockcond.And{
+					mockcond.Path("/api/v5/objects/emails"),
+					// Provider API doesn't allow these query parameters alongside with nextPageToken.
+					mockcond.QueryParamsMissing("sentAtAfterOrEqualTo", "limit"),
+					mockcond.Header(pardotHeader),
+				},
+				Then: mockserver.Response(http.StatusOK, responseEmailsEmptyPage),
+			}.Server(),
+			Expected: &common.ReadResult{
+				Rows:     0,
+				Data:     []common.ReadResultRow{},
+				NextPage: "",
+				Done:     true,
+			},
+			ExpectedErrs: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		// nolint:varnamelen
+		t.Run(tt.Name, func(t *testing.T) {
+			t.Parallel()
+
+			tt.Run(t, func() (connectors.ReadConnector, error) {
+				return constructTestConnectorAccountEngagement(tt.Server.URL)
+			})
+		})
+	}
+}
+
 func constructTestConnector(serverURL string) (*Connector, error) {
 	return constructTestConnectorGeneral(serverURL, providers.ModuleSalesforceCRM)
 }
@@ -152,6 +318,10 @@ func constructTestConnectorGeneral(serverURL string, module common.ModuleID) (*C
 		WithAuthenticatedClient(mockutils.NewClient()),
 		WithWorkspace("test-workspace"),
 		WithModule(module),
+		WithMetadata(map[string]string{
+			"isDemo":         "true",
+			"businessUnitId": "test-business-unit-id",
+		}),
 	)
 	if err != nil {
 		return nil, err
diff --git a/providers/salesforce/test/pardot/read/emails/1-first-page.json b/providers/salesforce/test/pardot/read/emails/1-first-page.json
new file mode 100644
index 00000000..1925a860
--- /dev/null
+++ b/providers/salesforce/test/pardot/read/emails/1-first-page.json
@@ -0,0 +1,24 @@
+{
+  "nextPageToken": "eyJvcmRlckJ5IjoiIiwiZmlsdGVycyI6W10sImxpbWl0IjoxLCJyZXN1bWVWYWx1ZSI6eyJpZCI6MzQyNzc4NjN9LCJwYWdlIjoyLCJyZWNDb3VudCI6MiwiZXhwaXJlVGltZSI6IjIwMjUtMDUtMTZUMjA6Mzc6MTMtMDc6MDAiLCJkZWxldGVkIjpudWxsfQ==",
+  "nextPageUrl": "https://pi.demo.pardot.com/api/v5/objects/emails?fields=id,name,subject,clientType,prospectId,listEmailId,updatedAt&nextPageToken=eyJvcmRlckJ5IjoiIiwiZmlsdGVycyI6W10sImxpbWl0IjoxLCJyZXN1bWVWYWx1ZSI6eyJpZCI6MzQyNzc4NjN9LCJwYWdlIjoyLCJyZWNDb3VudCI6MiwiZXhwaXJlVGltZSI6IjIwMjUtMDUtMTZUMjA6Mzc6MTMtMDc6MDAiLCJkZWxldGVkIjpudWxsfQ==",
+  "values": [
+    {
+      "id": 34277860,
+      "clientType": "Web",
+      "listEmailId": null,
+      "name": "Sending first email ever",
+      "prospectId": null,
+      "subject": "Few Moments Later",
+      "updatedAt": "2025-05-16T11:27:08-07:00"
+    },
+    {
+      "id": 34277863,
+      "clientType": "Web",
+      "listEmailId": null,
+      "name": "Second email, on the roll",
+      "prospectId": null,
+      "subject": "Second email, unbelievable",
+      "updatedAt": "2025-05-16T11:29:03-07:00"
+    }
+  ]
+}
diff --git a/providers/salesforce/test/pardot/read/emails/2-empty-page.json b/providers/salesforce/test/pardot/read/emails/2-empty-page.json
new file mode 100644
index 00000000..4ae60d43
--- /dev/null
+++ b/providers/salesforce/test/pardot/read/emails/2-empty-page.json
@@ -0,0 +1,5 @@
+{
+  "nextPageToken": null,
+  "nextPageUrl": null,
+  "values": []
+}
diff --git a/providers/salesforce/test/pardot/read/emails/err-missing-header.json b/providers/salesforce/test/pardot/read/emails/err-missing-header.json
new file mode 100644
index 00000000..849d9d74
--- /dev/null
+++ b/providers/salesforce/test/pardot/read/emails/err-missing-header.json
@@ -0,0 +1,4 @@
+{
+  "code": 181,
+  "message": "A required header is missing: Pardot-Business-Unit-Id header not found on request."
+}
diff --git a/providers/salesforce/test/pardot/read/emails/err-missing-query.json b/providers/salesforce/test/pardot/read/emails/err-missing-query.json
new file mode 100644
index 00000000..94fd3d54
--- /dev/null
+++ b/providers/salesforce/test/pardot/read/emails/err-missing-query.json
@@ -0,0 +1,4 @@
+{
+  "code": 13,
+  "message": "One or more required parameters are missing: fields"
+}
diff --git a/test/salesforce/connector.go b/test/salesforce/connector.go
index fa033b85..413944a7 100644
--- a/test/salesforce/connector.go
+++ b/test/salesforce/connector.go
@@ -14,6 +14,12 @@ import (
 	"golang.org/x/oauth2"
 )
 
+var fieldBusinessUnitID = credscanning.Field{
+	Name:      "businessUnitId",
+	PathJSON:  "metadata.businessUnitId",
+	SuffixENV: "BUSINESS_UNIT_ID",
+}
+
 func GetSalesforceConnector(ctx context.Context) *salesforce.Connector {
 	return getSalesforceConnector(ctx, providers.ModuleSalesforceCRM)
 }
@@ -29,6 +35,9 @@ func getSalesforceConnector(ctx context.Context, module common.ModuleID) *salesf
 		salesforce.WithClient(ctx, http.DefaultClient, getConfig(reader), reader.GetOauthToken()),
 		salesforce.WithWorkspace(reader.Get(credscanning.Fields.Workspace)),
 		salesforce.WithModule(module),
+		salesforce.WithMetadata(map[string]string{
+			"businessUnitId": reader.Get(fieldBusinessUnitID),
+		}),
 	)
 	if err != nil {
 		testUtils.Fail("error creating connector", "error", err)
@@ -59,7 +68,9 @@ func GetSalesforceAccessToken() string {
 
 func getSalesforceJSONReader() *credscanning.ProviderCredentials {
 	filePath := credscanning.LoadPath(providers.Salesforce)
-	reader := utils.MustCreateProvCredJSON(filePath, true)
+	reader := utils.MustCreateProvCredJSON(filePath, true,
+		fieldBusinessUnitID,
+	)
 
 	return reader
 }
diff --git a/test/salesforce/pardot/read/emails/main.go b/test/salesforce/pardot/read/emails/main.go
new file mode 100644
index 00000000..e0136321
--- /dev/null
+++ b/test/salesforce/pardot/read/emails/main.go
@@ -0,0 +1,39 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/common"
+	connTest "github.com/amp-labs/connectors/test/salesforce"
+	"github.com/amp-labs/connectors/test/utils"
+)
+
+func main() {
+	// Handle Ctrl-C gracefully.
+	ctx, done := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer done()
+
+	// Set up slog logging.
+	utils.SetupLogging()
+
+	conn := connTest.GetSalesforceAccountEngagementConnector(ctx)
+
+	res, err := conn.Read(ctx, common.ReadParams{
+		ObjectName: "EmaILs",
+		Fields: connectors.Fields(
+			"id", "name", "subject", "clientType",
+			"prospectId", "listEmailId", "updatedAt",
+			"senderOptions.type"),
+	})
+	if err != nil {
+		utils.Fail("error reading from provider", "error", err)
+	}
+
+	slog.Info("Reading")
+	utils.DumpJSON(res, os.Stdout)
+}