Repository: amp-labs/connectors
Pull request #73 (2024-02-26 23:48:13)

[ENG-522]Update testcases credentials, create reusable schema,
This PR updates salesforce test cases to implement credential registry. 

Salesforce Test

```
jaehyunlim@Jaehyuns-MacBook-Pro connectors % go run test/salesforce/bulkwrite/main.go
Testing Bulkwrite...

=============================================
Testing Bulkwrite
=============================================

Upload complete.
{
    "state": "UploadComplete",
    "jobId": "750Dp0000090X8WIAU"
}
Write Result
{
    "id": "750Dp0000090X8WIAU",
    "object": "Touchpoint__c",
    "createdById": "005Dp000003Cd1SIAS",
    "externalIdFieldName": "external_id__c",
    "state": "Failed",
    "operation": "upsert",
    "columnDelimiter": "COMMA",
    "lineEnding": "LF",
    "numberRecordsFailed": 0,
    "numberRecordsProcessed": 0,
    "apexProcessingTime": 0,
    "apiActiveProcessingTime": 0,
    "apiVersion": 59,
    "concurrencyMode": "Parallel",
    "contentType": "CSV",
    "createdDate": "2024-03-19T23:30:28.000+0000",
    "jobType": "V2Ingest",
    "retries": 0,
    "systemModstamp": "2024-03-19T23:30:29.000+0000",
    "totalProcessingTime": 0
}


=============================================
Testing SuccessResults
=============================================

Upload complete.
{
    "state": "UploadComplete",
    "jobId": "750Dp0000090X8lIAE"
}
Write Result
{
    "jobId": "750Dp0000090X8lIAE",
    "state": "Failed",
    "jobInfo": {
        "id": "750Dp0000090X8lIAE",
        "object": "Touchpoint__c",
        "createdById": "005Dp000003Cd1SIAS",
        "externalIdFieldName": "external_id__c",
        "state": "Failed",
        "operation": "upsert",
        "columnDelimiter": "COMMA",
        "lineEnding": "LF",
        "numberRecordsFailed": 0,
        "numberRecordsProcessed": 0,
        "apexProcessingTime": 0,
        "apiActiveProcessingTime": 0,
        "apiVersion": 59,
        "concurrencyMode": "Parallel",
        "contentType": "CSV",
        "createdDate": "2024-03-19T23:30:28.000+0000",
        "jobType": "V2Ingest",
        "retries": 0,
        "systemModstamp": "2024-03-19T23:30:29.000+0000",
        "totalProcessingTime": 0
    },
    "message": "No records processed successfully. This is likely due the CSV being empty or issues with CSV column names."
}


=============================================
Testing Partial Failure
=============================================

Upload complete.
{
    "state": "UploadComplete",
    "jobId": "750Dp0000090X8gIAE"
}
Write Result
{
    "jobId": "750Dp0000090X8gIAE",
    "state": "Failed",
    "jobInfo": {
        "id": "750Dp0000090X8gIAE",
        "object": "Touchpoint__c",
        "createdById": "005Dp000003Cd1SIAS",
        "externalIdFieldName": "external_id__c",
        "state": "Failed",
        "operation": "upsert",
        "columnDelimiter": "COMMA",
        "lineEnding": "LF",
        "numberRecordsFailed": 0,
        "numberRecordsProcessed": 0,
        "apexProcessingTime": 0,
        "apiActiveProcessingTime": 0,
        "apiVersion": 59,
        "concurrencyMode": "Parallel",
        "contentType": "CSV",
        "createdDate": "2024-03-19T23:30:28.000+0000",
        "jobType": "V2Ingest",
        "retries": 0,
        "systemModstamp": "2024-03-19T23:30:29.000+0000",
        "totalProcessingTime": 0
    },
    "message": "No records processed successfully. This is likely due the CSV being empty or issues with CSV column names."
}


=============================================
Testing Complete Failure
=============================================

Upload complete.
{
    "state": "UploadComplete",
    "jobId": "750Dp0000090X8bIAE"
}
Write Result
{
    "jobId": "750Dp0000090X8bIAE",
    "state": "Failed",
    "jobInfo": {
        "id": "750Dp0000090X8bIAE",
        "object": "Touchpoint__c",
        "createdById": "005Dp000003Cd1SIAS",
        "externalIdFieldName": "external_id__c",
        "state": "Failed",
        "operation": "upsert",
        "columnDelimiter": "COMMA",
        "lineEnding": "LF",
        "numberRecordsFailed": 0,
        "numberRecordsProcessed": 0,
        "apexProcessingTime": 0,
        "apiActiveProcessingTime": 0,
        "apiVersion": 59,
        "concurrencyMode": "Parallel",
        "contentType": "CSV",
        "createdDate": "2024-03-19T23:30:28.000+0000",
        "jobType": "V2Ingest",
        "retries": 0,
        "systemModstamp": "2024-03-19T23:30:29.000+0000",
        "totalProcessingTime": 0
    },
    "message": "No records processed successfully. This is likely due the CSV being empty or issues with CSV column names."
}

```

Hubspot test

```
jaehyunlim@Jaehyuns-MacBook-Pro connectors % go run test/hubspot/write/main.go
{
  "success": true,
  "recordId": "5069569450",
  "data": {
    "company": "Numedii",
    "createdate": "2024-03-19T23:11:00.536Z",
    "email": "jalynbeatty@lubowitz.name",
    "firstname": "Monserrat",
    "hs_all_contact_vids": "5069569450",
    "hs_calculated_phone_number": "+16619403402",
    "hs_calculated_phone_number_country_code": "US",
    "hs_email_domain": "lubowitz.name",
    "hs_is_contact": "true",
    "hs_is_unworked": "true",
    "hs_lifecyclestage_lead_date": "2024-03-19T23:11:00.536Z",
    "hs_object_id": "5069569450",
    "hs_object_source": "INTEGRATION",
    "hs_object_source_id": "2317233",
    "hs_object_source_label": "INTEGRATION",
    "hs_pipeline": "contacts-lifecycle-pipeline",
    "hs_searchable_calculated_phone_number": "6619403402",
    "lastmodifieddate": "2024-03-19T23:11:00.536Z",
    "lastname": "Hermann",
    "lifecyclestage": "lead",
    "phone": "6619403402",
    "website": "http://www.nationalstrategic.com/empower/repurpose/distributed/web-readiness"
  }
}Wrote contact
Reading contacts..
{
  "rows": 0,
  "data": [],
  "done": true
}
```

jlimatampersand: Current dependencies on/for this PR:
* **#73** <a href="https://app.graphite.dev/github/pr/amp-labs/connectors/73?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`

This [stack of pull requests](https://stacking.dev/?utm_source=stack-comment) is managed by [Graphite](https://app.graphite.dev/github/pr/amp-labs/connectors/73?utm_source=stack-comment).

Patch:
diff --git a/.gitignore b/.gitignore
index a80f8102..f0950b63 100644
--- a/.gitignore
+++ b/.gitignore
@@ -120,6 +120,5 @@ go.work
 
 playground/
 
-creds.json
-.DS_Store
-**/.DS_Store
+*creds.json
+**/.DS_Store
\ No newline at end of file
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 00000000..631d999c
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,91 @@
+# Contributing to Connectors
+
+When you want to contribute to Connectors, we highly suggest you include tests.
+
+To test connectors, you'll need your own account and instance with the provider you're testing (like Salesforce or Hubspot). You'll also need to set up credentials (using oAuth2.0) to link the connectors to the providers.
+
+Since each provider has different credential parameters and everyone has different environments, we've made a `CredentialRegistry` and `Reader` interface to make credential loading consistent. Please use these. We currently support `JSONReader`, `EnvVarReader`, and `ValueReader`.
+
+For examples on how to do this, check out the tests in `test/salesforce` and `test/hubspot`.
+
+Example:
+
+`salesforceCred1.json`
+```json
+{
+    "id": "someID",
+    "projectId": "someProjectId",
+    "providerApp": {
+        "id": "someId",
+        "externalRef": "externalRef:p0:e0:pr0",
+        "provider": "salesforce",
+        "clientId": "YOUR_CLIENT_ID1",
+        "clientSecret": "YOUR_CLIENT_SECRET1",
+        "scopes": null,
+        "projectId": "someProjectId"
+    },
+}
+```
+
+`salesforceCred2.json`
+```json
+{
+    "id": "someId2",
+    "providerApp": {
+        "id": "someId2",
+        "externalRef": "externalRef:p0:e0:pr0",
+        "provider": "salesforce",
+        "clientId": "YOUR_CLIENT_ID2",
+        "clientSecret": "YOUR_CLIENT_SECRET2",
+        "scopes": null,
+        "projectId": "someProjectId2"
+    },
+}
+```
+
+```go
+	credentialsRegistry := utils.NewCredentialsRegistry()
+    readers := []utils.Reader{
+		&utils.EnvReader{
+			EnvName: "ACCESS_TOKEN",  // This reads an environment variable named ACCESS_TOKEN
+			CredKey: utils.AccessToken, // or some string key to get key
+		},
+		&utils.EnvReader{
+			EnvName: "REFRESH_TOKEN", // This reads an environment variable named REFRESH_TOKEN
+			CredKey: utils.RefreshToken,
+		},
+		&utils.JSONReader{
+			FilePath: "salesforceCred1.json", // This reads from a file
+			JSONPath: "$.providerApp.clientId",
+			CredKey:  utils.ClientId,
+		},
+		&utils.JSONReader{
+			FilePath: "salesforceCred2.json", // This reads from a file
+			JSONPath: "$.providerApp.clientSecret",
+			CredKey:  utils.ClientSecret,
+		},
+       
+	}
+	credentialsRegistry.AddReaders(readers...)
+    credentialsRegistry.AddReader(
+        &ValueReader{
+            Val:     "some-salesforce-instance-subdomain", // Hard coded value
+            CredKey: "workspaceRef",
+        },
+    )
+	salesforceWorkspace := credentialsRegistry.MustString("workspaceRef")
+
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
+	ctx := context.Background()
+
+	sfc, err := connectors.Salesforce(
+		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
+		salesforce.WithWorkspace(salesforceWorkspace))
+	if err != nil {
+		slog.Error("Error creating Salesforce connector", "error", err)
+
+		return
+	}
+
+```
\ No newline at end of file
diff --git a/utils/credentialRegistry.go b/utils/credentialRegistry.go
new file mode 100644
index 00000000..cd133d22
--- /dev/null
+++ b/utils/credentialRegistry.go
@@ -0,0 +1,442 @@
+// nolint: ireturn
+package utils
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"log/slog"
+	"os"
+
+	"github.com/go-playground/validator"
+	"github.com/spyzhov/ajson"
+)
+
+var (
+	validate                = validator.New() //nolint:gochecknoglobals
+	ErrKeyNotFound          = errors.New("key not found")
+	ErrWrongType            = errors.New("wrong type")
+	ErrEnvVarNotSet         = errors.New("environment variable not set")
+	ErrEmptyFilePathPathVar = errors.New("empty value for file path")
+	ErrJSONPathNotFound     = errors.New("empty value at json path")
+	ErrReaderNotFound       = errors.New("Reader not found")
+	ErrCredentialNotFound   = errors.New("credential not found")
+)
+
+type CredentialsRegistry map[string]Reader
+
+type Reader interface {
+	Value() (any, error)
+	Key() (string, error)
+}
+
+type EnvReaderParam struct {
+	EnvName string `json:"envVar" validate:"required"`
+}
+
+type EnvReader struct {
+	EnvName string `json:"params" validate:"required"`
+	CredKey string `json:"string" validate:"required"`
+}
+
+func (r *EnvReader) Value() (any, error) {
+	value := os.Getenv(r.EnvName)
+	if value == "" {
+		return "", fmt.Errorf("%w: %s", ErrEnvVarNotSet, r.EnvName)
+	}
+
+	return value, nil
+}
+
+func (r *EnvReader) Key() (string, error) {
+	if r.CredKey == "" {
+		return "", fmt.Errorf("%w: %s", ErrKeyNotFound, r.EnvName)
+	}
+
+	return r.CredKey, nil
+}
+
+type ValueReader struct {
+	Val     any    `json:"val"    validate:"required"`
+	CredKey string `json:"string" validate:"required"`
+}
+
+func (r *ValueReader) Value() (any, error) {
+	if r.Val == nil {
+		return "", fmt.Errorf("%w: %s", ErrCredentialNotFound, r.CredKey)
+	}
+
+	return r.Val, nil
+}
+
+func (r *ValueReader) Key() (string, error) {
+	if r.CredKey == "" {
+		return "", fmt.Errorf("%w: %s", ErrKeyNotFound, r.Val)
+	}
+
+	return r.CredKey, nil
+}
+
+type JSONReader struct {
+	FilePath string `json:"filePath" validate:"required"`
+	JSONPath string `json:"jsonPath" validate:"required"`
+	CredKey  string `json:"string"   validate:"required"`
+}
+
+func (r *JSONReader) Key() (string, error) {
+	if r.CredKey == "" {
+		return "", fmt.Errorf("%w: %s", ErrKeyNotFound, r.FilePath)
+	}
+
+	return r.CredKey, nil
+}
+
+func (r *JSONReader) Value() (any, error) {
+	data, err := os.ReadFile(r.FilePath)
+	if err != nil {
+		slog.Error("Error opening creds.json", "error", err)
+
+		return nil, err
+	}
+
+	credsMap, err := ajson.Unmarshal(data)
+	if err != nil {
+		return nil, err
+	}
+
+	list, err := credsMap.JSONPath(r.JSONPath)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(list) == 0 || list[0] == nil {
+		return nil, fmt.Errorf("%w: %s", ErrJSONPathNotFound, r.JSONPath)
+	}
+
+	return list[0].Value()
+}
+
+func (c CredentialsRegistry) AddReader(reader Reader) error {
+	if err := validate.Struct(reader); err != nil {
+		return err
+	}
+
+	key, err := reader.Key()
+	if err != nil {
+		return err
+	}
+
+	c[key] = reader
+
+	return nil
+}
+
+func (c CredentialsRegistry) AddReaders(readers ...Reader) error {
+	for _, reader := range readers {
+		if err := validate.Struct(reader); err != nil {
+			return fmt.Errorf("%w: %v", err, reader)
+		}
+
+		if err := c.AddReader(reader); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c CredentialsRegistry) Get(key string) (any, error) {
+	reader, ok := c[key]
+	if !ok {
+		return nil, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return reader.Value()
+}
+
+func (c CredentialsRegistry) GetString(key string) (string, error) {
+	reader, ok := c[key]
+	if !ok {
+		return "", fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[string](reader)
+}
+
+func (c CredentialsRegistry) GetBool(key string) (bool, error) {
+	reader, ok := c[key]
+	if !ok {
+		return false, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[bool](reader)
+}
+
+func (c CredentialsRegistry) GetInt(key string) (int, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[int](reader)
+}
+
+func (c CredentialsRegistry) GetFloat64(key string) (float64, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[float64](reader)
+}
+
+func (c CredentialsRegistry) GetFloat32(key string) (float32, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[float32](reader)
+}
+
+func (c CredentialsRegistry) GetInt64(key string) (int64, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[int64](reader)
+}
+
+func (c CredentialsRegistry) GetInt32(key string) (int32, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[int32](reader)
+}
+
+func (c CredentialsRegistry) GetInt16(key string) (int16, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[int16](reader)
+}
+
+func (c CredentialsRegistry) GetInt8(key string) (int8, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[int8](reader)
+}
+
+func (c CredentialsRegistry) GetUint64(key string) (uint64, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[uint64](reader)
+}
+
+func (c CredentialsRegistry) GetUint32(key string) (uint32, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[uint32](reader)
+}
+
+func (c CredentialsRegistry) GetUint16(key string) (uint16, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[uint16](reader)
+}
+
+func (c CredentialsRegistry) GetUint8(key string) (uint8, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[uint8](reader)
+}
+
+func (c CredentialsRegistry) GetUint(key string) (uint, error) {
+	reader, ok := c[key]
+	if !ok {
+		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[uint](reader)
+}
+
+func (c CredentialsRegistry) GetMap(key string) (map[string]*ajson.Node, error) {
+	reader, ok := c[key]
+	if !ok {
+		return nil, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	}
+
+	return getFromReader[map[string]*ajson.Node](reader)
+}
+
+func (c CredentialsRegistry) MustString(credKey string) string {
+	str, err := c.GetString(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return str
+}
+
+func (c CredentialsRegistry) MustBool(credKey string) bool {
+	b, err := c.GetBool(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return b
+}
+
+func (c CredentialsRegistry) MustInt(credKey string) int {
+	i, err := c.GetInt(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustFloat64(credKey string) float64 {
+	f, err := c.GetFloat64(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return f
+}
+
+func (c CredentialsRegistry) MustFloat32(credKey string) float32 {
+	f, err := c.GetFloat32(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return f
+}
+
+func (c CredentialsRegistry) MustInt64(credKey string) int64 {
+	i, err := c.GetInt64(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustInt32(credKey string) int32 {
+	i, err := c.GetInt32(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustInt16(credKey string) int16 {
+	i, err := c.GetInt16(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustInt8(credKey string) int8 {
+	i, err := c.GetInt8(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustUint64(credKey string) uint64 {
+	i, err := c.GetUint64(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustUint32(credKey string) uint32 {
+	i, err := c.GetUint32(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustUint16(credKey string) uint16 {
+	i, err := c.GetUint16(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustUint8(credKey string) uint8 {
+	i, err := c.GetUint8(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func (c CredentialsRegistry) MustUint(credKey string) uint {
+	i, err := c.GetUint(credKey)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return i
+}
+
+func NewCredentialsRegistry() CredentialsRegistry {
+	return make(CredentialsRegistry)
+}
+
+func getFromReader[T any](reader Reader) (T, error) {
+	var of T
+
+	val, err := reader.Value()
+	if err != nil {
+		return of, err
+	}
+
+	return get[T](val)
+}
+
+func get[T any](val any) (T, error) {
+	var of T
+
+	v, ok := (val).(T)
+	if !ok {
+		return of, fmt.Errorf("%w. expected %T, got %T", ErrWrongType, of, val)
+	}
+
+	return v, nil
+}
diff --git a/utils/credentials.go b/utils/credentials.go
index 4b7887f3..fe0be17b 100644
--- a/utils/credentials.go
+++ b/utils/credentials.go
@@ -1,411 +1,91 @@
-// nolint: ireturn
 package utils
 
 import (
-	"errors"
 	"fmt"
-	"log"
-	"log/slog"
-	"os"
+	"time"
 
-	"github.com/spyzhov/ajson"
+	"golang.org/x/oauth2"
 )
 
+//nolint:gochecknoglobals
 var (
-	ErrWrongType            = errors.New("wrong type")
-	ErrEnvVarNotSet         = errors.New("environment variable not set")
-	ErrEmptyStringEnvVar    = errors.New("empty value for environment variable")
-	ErrEmptyFilePathPathVar = errors.New("empty value for file path")
-	ErrJSONPathNotFound     = errors.New("empty value at json path")
-	ErrReaderNotFound       = errors.New("Reader not found")
-	ErrCredentialNotFound   = errors.New("credential not found")
+	AccessToken  = "accessToken"
+	RefreshToken = "refreshToken"
+	ClientId     = "clientId"
+	ClientSecret = "clientSecret"
+	WorkspaceRef = "workspaceRef"
+	Provider     = "provider"
 )
 
-type CredentialsRegistry map[string]Reader
+func SalesforceOAuthConfigFromRegistry(registry CredentialsRegistry) *oauth2.Config {
+	clientId := registry.MustString(ClientId)
+	clientSecret := registry.MustString(ClientSecret)
+	salesforceWorkspace := registry.MustString(WorkspaceRef)
 
-type Reader interface {
-	Value() (any, error)
-	Key() (string, error)
-}
-
-type EnvReaderParam struct {
-	EnvName string `json:"envVar" validate:"required"`
-}
-
-type EnvReader struct {
-	EnvName string `json:"params" validate:"required"`
-	CredKey string `json:"string" validate:"required"`
-}
-
-func (r *EnvReader) Value() (any, error) {
-	value := os.Getenv(r.EnvName)
-	if value == "" {
-		return "", fmt.Errorf("%w: %s", ErrEnvVarNotSet, r.EnvName)
-	}
-
-	return value, nil
-}
-
-func (r *EnvReader) Key() (string, error) {
-	if r.CredKey == "" {
-		return "", fmt.Errorf("%w: %s", ErrEmptyStringEnvVar, r.EnvName)
+	cfg := &oauth2.Config{
+		ClientID:     clientId,
+		ClientSecret: clientSecret,
+		Endpoint: oauth2.Endpoint{
+			AuthURL:   fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/authorize", salesforceWorkspace),
+			TokenURL:  fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/token", salesforceWorkspace),
+			AuthStyle: oauth2.AuthStyleInParams,
+		},
 	}
 
-	return r.CredKey, nil
+	return cfg
 }
 
-type JSONReader struct {
-	FilePath string `json:"filePath" validate:"required"`
-	JSONPath string `json:"jsonPath" validate:"required"`
-	CredKey  string `json:"string"   validate:"required"`
-}
-
-func (r *JSONReader) Key() (string, error) {
-	if r.CredKey == "" {
-		return "", fmt.Errorf("%w: %s", ErrEmptyFilePathPathVar, r.FilePath)
-	}
-
-	return r.CredKey, nil
-}
-
-func (r *JSONReader) Value() (any, error) {
-	data, err := os.ReadFile(r.FilePath)
-	if err != nil {
-		slog.Error("Error opening creds.json", "error", err)
-
-		return nil, err
-	}
-
-	credsMap, err := ajson.Unmarshal(data)
-	if err != nil {
-		return nil, err
-	}
-
-	list, err := credsMap.JSONPath(r.JSONPath)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(list) == 0 || list[0] == nil {
-		return nil, fmt.Errorf("%w: %s", ErrJSONPathNotFound, r.JSONPath)
-	}
-
-	return list[0].Value()
-}
-
-func (c CredentialsRegistry) AddReader(reader Reader) error {
-	key, err := reader.Key()
-	if err != nil {
-		return err
-	}
-
-	c[key] = reader
-
-	return nil
-}
-
-func (c CredentialsRegistry) AddReaders(readers ...Reader) error {
-	for _, Reader := range readers {
-		if err := c.AddReader(Reader); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-func (c CredentialsRegistry) Get(key string) (any, error) {
-	reader, ok := c[key]
-	if !ok {
-		return nil, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return reader.Value()
-}
+func SalesforceOauthTokenFromRegistry(registry CredentialsRegistry) *oauth2.Token {
+	accessToken := registry.MustString(AccessToken)
+	refreshToken := registry.MustString(RefreshToken)
 
-func (c CredentialsRegistry) GetString(key string) (string, error) {
-	reader, ok := c[key]
-	if !ok {
-		return "", fmt.Errorf("%w: %s", ErrReaderNotFound, key)
+	tok := &oauth2.Token{
+		AccessToken:  accessToken,
+		RefreshToken: refreshToken,
+		TokenType:    "bearer",
+		Expiry:       time.Now().Add(-1 * time.Hour),
 	}
 
-	return getFromReader[string](reader)
-}
-
-func (c CredentialsRegistry) GetBool(key string) (bool, error) {
-	reader, ok := c[key]
-	if !ok {
-		return false, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[bool](reader)
-}
-
-func (c CredentialsRegistry) GetInt(key string) (int, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[int](reader)
-}
-
-func (c CredentialsRegistry) GetFloat64(key string) (float64, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[float64](reader)
-}
-
-func (c CredentialsRegistry) GetFloat32(key string) (float32, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[float32](reader)
-}
-
-func (c CredentialsRegistry) GetInt64(key string) (int64, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[int64](reader)
-}
-
-func (c CredentialsRegistry) GetInt32(key string) (int32, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[int32](reader)
-}
-
-func (c CredentialsRegistry) GetInt16(key string) (int16, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[int16](reader)
-}
-
-func (c CredentialsRegistry) GetInt8(key string) (int8, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[int8](reader)
-}
-
-func (c CredentialsRegistry) GetUint64(key string) (uint64, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[uint64](reader)
-}
-
-func (c CredentialsRegistry) GetUint32(key string) (uint32, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[uint32](reader)
-}
-
-func (c CredentialsRegistry) GetUint16(key string) (uint16, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[uint16](reader)
-}
-
-func (c CredentialsRegistry) GetUint8(key string) (uint8, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[uint8](reader)
-}
-
-func (c CredentialsRegistry) GetUint(key string) (uint, error) {
-	reader, ok := c[key]
-	if !ok {
-		return 0, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[uint](reader)
-}
-
-func (c CredentialsRegistry) GetMap(key string) (map[string]*ajson.Node, error) {
-	reader, ok := c[key]
-	if !ok {
-		return nil, fmt.Errorf("%w: %s", ErrReaderNotFound, key)
-	}
-
-	return getFromReader[map[string]*ajson.Node](reader)
-}
-
-func (c CredentialsRegistry) MustString(credKey string) string {
-	str, err := c.GetString(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return str
-}
-
-func (c CredentialsRegistry) MustBool(credKey string) bool {
-	b, err := c.GetBool(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return b
-}
-
-func (c CredentialsRegistry) MustInt(credKey string) int {
-	i, err := c.GetInt(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustFloat64(credKey string) float64 {
-	f, err := c.GetFloat64(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return f
-}
-
-func (c CredentialsRegistry) MustFloat32(credKey string) float32 {
-	f, err := c.GetFloat32(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return f
-}
-
-func (c CredentialsRegistry) MustInt64(credKey string) int64 {
-	i, err := c.GetInt64(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustInt32(credKey string) int32 {
-	i, err := c.GetInt32(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustInt16(credKey string) int16 {
-	i, err := c.GetInt16(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustInt8(credKey string) int8 {
-	i, err := c.GetInt8(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustUint64(credKey string) uint64 {
-	i, err := c.GetUint64(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustUint32(credKey string) uint32 {
-	i, err := c.GetUint32(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustUint16(credKey string) uint16 {
-	i, err := c.GetUint16(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustUint8(credKey string) uint8 {
-	i, err := c.GetUint8(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func (c CredentialsRegistry) MustUint(credKey string) uint {
-	i, err := c.GetUint(credKey)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	return i
-}
-
-func NewCredentialsRegistry() CredentialsRegistry {
-	return make(CredentialsRegistry)
+	return tok
 }
 
-func getFromReader[T any](reader Reader) (T, error) {
-	var of T
+func HubspotOauthTokenFromRegistry(registry CredentialsRegistry) *oauth2.Token {
+	accessToken := registry.MustString(AccessToken)
+	refreshToken := registry.MustString(RefreshToken)
 
-	val, err := reader.Value()
-	if err != nil {
-		return of, err
+	tok := &oauth2.Token{
+		AccessToken:  accessToken,
+		RefreshToken: refreshToken,
+		TokenType:    "bearer",
+		Expiry:       time.Now().Add(-1 * time.Hour), // just pretend it's expired already, whatever, it'll fetch a new one.
 	}
 
-	return get[T](val)
+	return tok
 }
 
-func get[T any](val any) (T, error) {
-	var of T
+func HubspotOAuthConfigFromRegistry(registry CredentialsRegistry) *oauth2.Config {
+	clientId := registry.MustString(ClientId)
+	clientSecret := registry.MustString(ClientSecret)
 
-	v, ok := (val).(T)
-	if !ok {
-		return of, fmt.Errorf("%w. expected %T, got %T", ErrWrongType, of, val)
+	cfg := &oauth2.Config{
+		ClientID:     clientId,
+		ClientSecret: clientSecret,
+		RedirectURL:  "http://localhost:8080/callbacks/v1/oauth",
+		Endpoint: oauth2.Endpoint{
+			AuthURL:   "https://app.hubspot.com/oauth/authorize",
+			TokenURL:  "https://api.hubapi.com/oauth/v1/token",
+			AuthStyle: oauth2.AuthStyleInParams,
+		},
+		Scopes: []string{
+			"crm.objects.contacts.read",
+			"crm.objects.contacts.write",
+			"crm.objects.deals.read",
+			"crm.objects.line_items.read",
+			"oauth",
+			"crm.objects.companies.read",
+			"tickets",
+		},
 	}
 
-	return v, nil
+	return cfg
 }


Test patch:
diff --git a/test/creds.go b/test/creds.go
deleted file mode 100644
index 6833cc9c..00000000
--- a/test/creds.go
+++ /dev/null
@@ -1,75 +0,0 @@
-package test
-
-import (
-	"log/slog"
-	"os"
-
-	"github.com/spyzhov/ajson"
-)
-
-type Creds struct {
-	ClientId     string `json:"clientId"`
-	ClientSecret string `json:"clientSecret"`
-	AccessToken  string `json:"accessToken"`
-	RefreshToken string `json:"refreshToken"`
-	Workspace    string `json:"workspace"`
-	Provider     string `json:"provider"`
-}
-
-func GetCreds(path string) (*Creds, error) {
-	if path == "" {
-		path = "../creds.json"
-	}
-
-	data, err := os.ReadFile(path)
-	if err != nil {
-		slog.Error("Error opening creds.json", "error", err)
-		return nil, err
-	}
-
-	credsMap, err := ajson.Unmarshal(data)
-	if err != nil {
-		return nil, err
-	}
-
-	clientId, err := credsMap.JSONPath("$.providerApp.clientId")
-	if err != nil {
-		return nil, err
-	}
-
-	clientSecret, err := credsMap.JSONPath("$.providerApp.clientSecret")
-	if err != nil {
-		return nil, err
-	}
-
-	accessToken, err := credsMap.JSONPath("$.AccessToken.Token")
-	if err != nil {
-		return nil, err
-	}
-
-	refreshToken, err := credsMap.JSONPath("$.RefreshToken.Token")
-	if err != nil {
-		return nil, err
-	}
-
-	workspace, err := credsMap.JSONPath("$.providerWorkspaceRef")
-	if err != nil {
-		return nil, err
-	}
-
-	provider, err := credsMap.JSONPath("$.providerApp.provider")
-	if err != nil {
-		return nil, err
-	}
-
-	creds := &Creds{
-		ClientId:     clientId[0].MustString(),
-		ClientSecret: clientSecret[0].MustString(),
-		AccessToken:  accessToken[0].MustString(),
-		RefreshToken: refreshToken[0].MustString(),
-		Workspace:    workspace[0].MustString(),
-		Provider:     provider[0].MustString(),
-	}
-
-	return creds, nil
-}
diff --git a/test/hubspot/write.go b/test/hubspot/write.go
new file mode 100644
index 00000000..b014eb7f
--- /dev/null
+++ b/test/hubspot/write.go
@@ -0,0 +1,57 @@
+package hubspot
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/hubspot"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
+)
+
+// GetHubspotConnector returns a Hubspot connector.
+func GetHubspotConnector(ctx context.Context, filePath string) *hubspot.Connector {
+	registry := utils.NewCredentialsRegistry()
+
+	readers := []utils.Reader{
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.CLIENT_ID",
+			CredKey:  "clientId",
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.CLIENT_SECRET",
+			CredKey:  "clientSecret",
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.REFRESH_TOKEN",
+			CredKey:  "refreshToken",
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.ACCESS_TOKEN",
+			CredKey:  "accessToken",
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.PROVIDER",
+			CredKey:  "provider",
+		},
+	}
+	registry.AddReaders(readers...)
+
+	cfg := utils.HubspotOAuthConfigFromRegistry(registry)
+	tok := utils.HubspotOauthTokenFromRegistry(registry)
+
+	conn, err := connectors.Hubspot(
+		hubspot.WithClient(ctx, http.DefaultClient, cfg, tok),
+		hubspot.WithModule(hubspot.ModuleCRM))
+	if err != nil {
+		testUtils.Fail("error creating hubspot connector", "error", err)
+	}
+
+	return conn
+}
diff --git a/test/hubspot/main.go b/test/hubspot/write/main.go
similarity index 61%
rename from test/hubspot/main.go
rename to test/hubspot/write/main.go
index 556058c7..3e29dacc 100644
--- a/test/hubspot/main.go
+++ b/test/hubspot/write/main.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/amp-labs/connectors/common"
+	hsTest "github.com/amp-labs/connectors/test/hubspot"
 	"github.com/amp-labs/connectors/test/utils"
 	"github.com/brianvoe/gofakeit/v6"
 )
@@ -23,25 +24,36 @@ type Contact struct {
 	Firstname string `json:"firstname"`
 }
 
-// Running instructions
-//
-// Step 1: prepare creds.json file
-//
-//	e.g. {
-//	  "CLIENT_ID": "<client id goes here>",
-//	  "CLIENT_SECRET": "<client secret goes here>",
-//	  "ACCESS_TOKEN": "<access token goes here>",
-//	  "REFRESH_TOKEN": "<refresh token goes here>"
-//	}
-//
-// In 1password, you can find a Hubspot creds.json file in the "Shared" vault.
-// Look for the title "Hubspot Sample OAuth Credentials".
-// The 1password item has an attached file called "creds.json" that contains the JSON.
-//
-// Step 2: run the following commands
-//
-//	$> export CREDENTIALS_FILE=creds.json
-//	$> go run test/hubspot/main.go
+/*
+	Running instructions
+
+	Step 1: prepare "hubspot-creds.json" file
+
+	Create a file called "hubspot-creds.json" in the root of the project with the following contents
+
+		e.g. {
+		"CLIENT_ID": "<client id goes here>",
+		"CLIENT_SECRET": "<client secret goes here>",
+		"ACCESS_TOKEN": "<access token goes here>",
+		"REFRESH_TOKEN": "<refresh token goes here>"
+		}
+
+	or export to an environment variable HUBSPOT_CRED_FILE by following command
+
+	$> export HUBSPOT_CRED_FILE=./hubspot-creds.json # or the path to your hubspot-creds.json file
+
+
+	In 1password, you can find a Hubspot creds.json file in the "Shared" vault.
+	Look for the title "Hubspot Sample OAuth Credentials".
+	The 1password item has an attached file called "creds.json" that contains the JSON.
+
+	Step 2: run the following command
+
+		$> go run test/hubspot/write/main.go
+
+
+*/
+
 func main() {
 	// Handle Ctrl-C gracefully.
 	ctx, done := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -50,8 +62,13 @@ func main() {
 	// Set up slog logging.
 	utils.SetupLogging()
 
+	filePath := os.Getenv("HUBSPOT_CRED_FILE")
+	if filePath == "" {
+		filePath = "./hubspot-creds.json"
+	}
+
 	// Get the Hubspot connector.
-	hsConn := utils.GetHubspotConnector(ctx, "creds.json")
+	hsConn := hsTest.GetHubspotConnector(ctx, filePath)
 	defer utils.Close(hsConn)
 
 	// Write an artificial contact to Hubspot.
diff --git a/test/metadata/main.go b/test/metadata/main.go
index f002d487..5efde32a 100644
--- a/test/metadata/main.go
+++ b/test/metadata/main.go
@@ -6,55 +6,42 @@ import (
 	"log/slog"
 	"net/http"
 	"os"
-	"time"
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/common"
 	"github.com/amp-labs/connectors/salesforce"
-	"github.com/amp-labs/connectors/test"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
 	"github.com/subchen/go-xmldom"
-	"golang.org/x/oauth2"
 )
 
 func main() {
-	creds, err := test.GetCreds("../creds.json")
-	if err != nil {
-		slog.Error("Error getting creds", "error", err)
-		os.Exit(1)
-	}
-
-	clientId := creds.ClientId
-	clientSecret := creds.ClientSecret
-	accessToken := creds.AccessToken
-	refreshToken := creds.RefreshToken
-
-	salesforceWorkspace := creds.Workspace
-
-	cfg := &oauth2.Config{
-		ClientID:     clientId,
-		ClientSecret: clientSecret,
-		Endpoint: oauth2.Endpoint{
-			AuthURL:   fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/authorize", salesforceWorkspace),
-			TokenURL:  fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/token", salesforceWorkspace),
-			AuthStyle: oauth2.AuthStyleInParams,
-		},
+	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	})
+
+	logger := slog.New(handler)
+	slog.SetDefault(logger)
+
+	// assumes that this code is being run from the root of the project
+	// go run test/salesforce/bulkwrite/main.go
+	filePath := os.Getenv("SALESFORCE_CRED_FILE_PATH")
+	if filePath == "" {
+		filePath = "./salesforce-creds.json"
 	}
 
-	tok := &oauth2.Token{
-		AccessToken:  accessToken,
-		RefreshToken: refreshToken,
-		TokenType:    "bearer",
-		Expiry:       time.Now().Add(-1 * time.Hour),
-	}
+	ampConnectionSchemaReader := testUtils.JSONFileReaders(filePath)
+	credentialsRegistry := utils.NewCredentialsRegistry()
+	credentialsRegistry.AddReaders(ampConnectionSchemaReader...)
+	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
 
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
 	ctx := context.Background()
 
 	sfc, err := connectors.Salesforce(
-		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok,
-			salesforce.GetTokenUpdater(tok), // this is necessary to update token
-		),
-		salesforce.WithWorkspace(salesforceWorkspace),
-	)
+		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
+		salesforce.WithWorkspace(salesforceWorkspace))
 	if err != nil {
 		slog.Error("Error creating Salesforce connector", "error", err)
 
diff --git a/test/metadata/testCreateCustomObject.json b/test/metadata/testCreateCustomObject.json
deleted file mode 100644
index aed3fdfd..00000000
--- a/test/metadata/testCreateCustomObject.json
+++ /dev/null
@@ -1,75 +0,0 @@
-{
-    "xmlName": "metadata",
-    "attributes": [
-        {
-            "key": "xsi:type",
-            "value": "CustomObject"
-        }
-    ],
-    "children": [
-        {
-            "xmlName": "fullName",
-            "attributes": null,
-            "children": [
-                "TestObject15__c"
-            ],
-            "selfClosing": false
-        },
-        {
-            "xmlName": "label",
-            "attributes": null,
-            "children": [
-                "Test Object 15"
-            ],
-            "selfClosing": false
-        },
-        {
-            "xmlName": "pluralLabel",
-            "attributes": null,
-            "children": [
-                "Test Objects 15"
-            ],
-            "selfClosing": false
-        },
-        {
-            "xmlName": "nameField",
-            "attributes": null,
-            "children": [
-                {
-                    "xmlName": "type",
-                    "attributes": null,
-                    "children": [
-                        "Text"
-                    ],
-                    "selfClosing": false
-                },
-                {
-                    "xmlName": "label",
-                    "attributes": null,
-                    "children": [
-                        "Test Object Name"
-                    ],
-                    "selfClosing": false
-                }
-            ],
-            "selfClosing": false
-        },
-        {
-            "xmlName": "deploymentStatus",
-            "attributes": null,
-            "children": [
-                "Deployed"
-            ],
-            "selfClosing": false
-        },
-        {
-            "xmlName": "sharingModel",
-            "attributes": null,
-            "children": [
-                "ReadWrite"
-            ],
-            "selfClosing": false
-        }
-    ],
-    "selfClosing": false
-}
diff --git a/test/bulkwrite/main.go b/test/salesforce/bulkwrite/main.go
similarity index 76%
rename from test/bulkwrite/main.go
rename to test/salesforce/bulkwrite/main.go
index 70fcd4aa..bfca2ddc 100644
--- a/test/bulkwrite/main.go
+++ b/test/salesforce/bulkwrite/main.go
@@ -12,8 +12,8 @@ import (
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/salesforce"
-	"github.com/amp-labs/connectors/test"
-	"golang.org/x/oauth2"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
 )
 
 const (
@@ -23,45 +23,29 @@ const (
 func main() { //nolint:funlen
 	fmt.Println("Testing Bulkwrite...")
 
-	creds, err := test.GetCreds("../../creds.json")
-	if err != nil {
-		slog.Error("Error getting creds", "error", err)
-		os.Exit(1)
-	}
-
-	clientId := creds.ClientId
-	clientSecret := creds.ClientSecret
-	accessToken := creds.AccessToken
-	refreshToken := creds.RefreshToken
-
-	salesforceWorkspace := creds.Workspace
-
 	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
 		Level: slog.LevelDebug,
 	})
+
 	logger := slog.New(handler)
 	slog.SetDefault(logger)
 
-	cfg := &oauth2.Config{
-		ClientID:     clientId,
-		ClientSecret: clientSecret,
-		Endpoint: oauth2.Endpoint{
-			AuthURL:   fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/authorize", salesforceWorkspace),
-			TokenURL:  fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/token", salesforceWorkspace),
-			AuthStyle: oauth2.AuthStyleInParams,
-		},
+	// assumes that this code is being run from the root of the project
+	// go run test/salesforce/bulkwrite/main.go
+	filePath := os.Getenv("SALESFORCE_CRED_FILE_PATH")
+	if filePath == "" {
+		filePath = "./salesforce-creds.json"
 	}
 
-	tok := &oauth2.Token{
-		AccessToken:  accessToken,
-		RefreshToken: refreshToken,
-		TokenType:    "bearer",
-		Expiry:       time.Now().Add(-1 * time.Hour), // just pretend it's expired already, whatever, it'll fetch a new one.
-	}
+	ampConnectionSchemaReader := testUtils.JSONFileReaders(filePath)
+	credentialsRegistry := utils.NewCredentialsRegistry()
+	credentialsRegistry.AddReaders(ampConnectionSchemaReader...)
+	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
 
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
 	ctx := context.Background()
 
-	// Create a new Salesforce connector, with a token provider that uses the sfdx CLI to fetch an access token.
 	sfc, err := connectors.Salesforce(
 		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
 		salesforce.WithWorkspace(salesforceWorkspace))
@@ -145,22 +129,22 @@ func testBulkWrite(ctx context.Context, sfc *salesforce.Connector, filePath stri
 
 var testList = []testRunner{
 	{
-		filePath:  "./touchpoints_20231130.csv",
+		filePath:  "./test/salesforce/bulkwrite/touchpoints_20231130.csv",
 		testTitle: "Testing Bulkwrite",
 		fn:        testBulkWrite,
 	},
 	{
-		filePath:  "./touchpoints_20231130.csv",
+		filePath:  "./test/salesforce/bulkwrite/touchpoints_20231130.csv",
 		testTitle: "Testing SuccessResults",
 		fn:        testGetJobResultsForFile,
 	},
 	{
-		filePath:  "./touchpoints_partial_failure_20231228.csv",
+		filePath:  "./test/salesforce/bulkwrite/touchpoints_partial_failure_20231228.csv",
 		testTitle: "Testing Partial Failure",
 		fn:        testGetJobResultsForFile,
 	},
 	{
-		filePath:  "./touchpoints_complete_failure_20231228.csv",
+		filePath:  "./test/salesforce/bulkwrite/touchpoints_complete_failure_20231228.csv",
 		testTitle: "Testing Complete Failure",
 		fn:        testGetJobResultsForFile,
 	},
diff --git a/test/bulkwrite/touchpoints_20231130.csv b/test/salesforce/bulkwrite/touchpoints_20231130.csv
similarity index 100%
rename from test/bulkwrite/touchpoints_20231130.csv
rename to test/salesforce/bulkwrite/touchpoints_20231130.csv
diff --git a/test/bulkwrite/touchpoints_complete_failure_20231228.csv b/test/salesforce/bulkwrite/touchpoints_complete_failure_20231228.csv
similarity index 100%
rename from test/bulkwrite/touchpoints_complete_failure_20231228.csv
rename to test/salesforce/bulkwrite/touchpoints_complete_failure_20231228.csv
diff --git a/test/bulkwrite/touchpoints_partial_failure_20231228.csv b/test/salesforce/bulkwrite/touchpoints_partial_failure_20231228.csv
similarity index 100%
rename from test/bulkwrite/touchpoints_partial_failure_20231228.csv
rename to test/salesforce/bulkwrite/touchpoints_partial_failure_20231228.csv
diff --git a/test/salesforce/event-flow/main.go b/test/salesforce/event-flow/main.go
index c7d38856..e8360367 100644
--- a/test/salesforce/event-flow/main.go
+++ b/test/salesforce/event-flow/main.go
@@ -12,51 +12,34 @@ import (
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/salesforce"
-	"github.com/amp-labs/connectors/test"
-	"golang.org/x/oauth2"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
 )
 
 func main() {
-	// assumed that the creds.json file is in the root of the project
-	creds, err := test.GetCreds("./creds.json")
-	if err != nil {
-		slog.Error("Error getting creds", "error", err)
-		os.Exit(1)
-	}
-
-	clientId := creds.ClientId
-	clientSecret := creds.ClientSecret
-	accessToken := creds.AccessToken
-	refreshToken := creds.RefreshToken
-
-	salesforceWorkspace := creds.Workspace
-
 	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
 		Level: slog.LevelDebug,
 	})
+
 	logger := slog.New(handler)
 	slog.SetDefault(logger)
 
-	cfg := &oauth2.Config{
-		ClientID:     clientId,
-		ClientSecret: clientSecret,
-		Endpoint: oauth2.Endpoint{
-			AuthURL:   fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/authorize", salesforceWorkspace),
-			TokenURL:  fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/token", salesforceWorkspace),
-			AuthStyle: oauth2.AuthStyleInParams,
-		},
+	// assumes that this code is being run from the root of the project
+	// go run test/salesforce/bulkwrite/main.go
+	filePath := os.Getenv("SALESFORCE_CRED_FILE_PATH")
+	if filePath == "" {
+		filePath = "./salesforce-creds.json"
 	}
 
-	tok := &oauth2.Token{
-		AccessToken:  accessToken,
-		RefreshToken: refreshToken,
-		TokenType:    "bearer",
-		Expiry:       time.Now().Add(-1 * time.Hour), // just pretend it's expired already, whatever, it'll fetch a new one.
-	}
+	ampConnectionSchemaReader := testUtils.JSONFileReaders(filePath)
+	credentialsRegistry := utils.NewCredentialsRegistry()
+	credentialsRegistry.AddReaders(ampConnectionSchemaReader...)
+	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
 
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
 	ctx := context.Background()
 
-	// Create a new Salesforce connector, with a token provider that uses the sfdx CLI to fetch an access token.
 	sfc, err := connectors.Salesforce(
 		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
 		salesforce.WithWorkspace(salesforceWorkspace))
diff --git a/test/salesforce/metadata/main.go b/test/salesforce/metadata/main.go
new file mode 100644
index 00000000..941123b5
--- /dev/null
+++ b/test/salesforce/metadata/main.go
@@ -0,0 +1,86 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/salesforce"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
+	"github.com/subchen/go-xmldom"
+)
+
+func main() {
+	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	})
+
+	logger := slog.New(handler)
+	slog.SetDefault(logger)
+
+	// assumes that this code is being run from the root of the project
+	// go run test/salesforce/bulkwrite/main.go
+	filePath := os.Getenv("SALESFORCE_CRED_FILE_PATH")
+	if filePath == "" {
+		filePath = "./salesforce-creds.json"
+	}
+
+	ampConnectionSchemaReader := testUtils.JSONFileReaders(filePath)
+	credentialsRegistry := utils.NewCredentialsRegistry()
+	credentialsRegistry.AddReaders(ampConnectionSchemaReader...)
+	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
+
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
+	ctx := context.Background()
+
+	sfc, err := connectors.Salesforce(
+		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok,
+			salesforce.GetTokenUpdater(tok), // this is necessary to update token
+		),
+		salesforce.WithWorkspace(salesforceWorkspace),
+	)
+	if err != nil {
+		slog.Error("Error creating Salesforce connector", "error", err)
+
+		return
+	}
+
+	defer func() {
+		_ = sfc.Close()
+	}()
+
+	example, err := xmldom.ParseXML(`<createMetadata><metadata xsi:type="CustomObject"><fullName>TestObject15__c</fullName><label>Test Object 15</label><pluralLabel>Test Objects 15</pluralLabel><nameField><type>Text</type><label>Test Object Name</label></nameField><deploymentStatus>Deployed</deploymentStatus><sharingModel>ReadWrite</sharingModel></metadata><metadata xsi:type="CustomField"><fullName>TestObject13__c.Comments__c</fullName><label>Comments</label><type>LongTextArea</type><length>500</length><inlineHelpText>This field contains help text for this object</inlineHelpText><description>Add your comments about this object here</description><visibleLines>30</visibleLines><required>false</required><trackFeedHistory>false</trackFeedHistory><trackHistory>false</trackHistory></metadata></createMetadata>`)
+
+	if err != nil {
+		slog.Error("err parsing", "error", err)
+		os.Exit(1)
+	}
+
+	node := example.Root
+
+	// xmldom.ParseXML has known issue that namespace in attribute is not correctly parsed
+	// ex) xsi:type="CustomObject" is parsed as xsi="CustomObject"
+	// so we need to manually change the attribute name
+	// We are using this ParseXML only in this test runner to generate XML
+	// so we can safely implement this package with below modification
+	metadataList := node.FindByName("metadata")
+	for _, metadata := range metadataList {
+		for _, attr := range metadata.Attributes {
+			if attr.Name == "type" {
+				attr.Name = "xsi:type"
+			}
+		}
+	}
+
+	res, err := sfc.CreateMetadata(ctx, node, tok)
+	if err != nil {
+		slog.Error("err", "err", err)
+	}
+
+	fmt.Println("Field Operation Result", res)
+}
diff --git a/test/salesforce/read-write.go b/test/salesforce/read-write/read-write.go
similarity index 77%
rename from test/salesforce/read-write.go
rename to test/salesforce/read-write/read-write.go
index 971ffa8a..a8aaca0f 100644
--- a/test/salesforce/read-write.go
+++ b/test/salesforce/read-write/read-write.go
@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"encoding/json"
-	"flag"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -12,8 +11,9 @@ import (
 
 	"github.com/amp-labs/connectors"
 	"github.com/amp-labs/connectors/salesforce"
+	testUtils "github.com/amp-labs/connectors/test/utils"
+	"github.com/amp-labs/connectors/utils"
 	"github.com/joho/godotenv"
-	"golang.org/x/oauth2"
 )
 
 // Set the appropriate environment variables in a .env file, then run:
@@ -26,50 +26,36 @@ func main() {
 }
 
 func mainFn() int { //nolint:funlen
-	if err := godotenv.Load(); err != nil {
-		slog.Error("Error loading .env file", "error", err)
-
-		return 1
-	}
-
-	salesforceWorkspace := os.Getenv("SALESFORCE_WORKSPACE")
-	clientId := os.Getenv("SALESFORCE_CLIENT_ID")
-	clientSecret := os.Getenv("SALESFORCE_CLIENT_SECRET")
-	accessToken := os.Getenv("SALESFORCE_ACCESS_TOKEN")
-	refreshToken := os.Getenv("SALESFORCE_REFRESH_TOKEN")
-
-	workspace := flag.String("workspace", salesforceWorkspace, "Salesforce workspace")
-	flag.Parse()
-
 	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
 		Level: slog.LevelDebug,
 	})
+
 	logger := slog.New(handler)
 	slog.SetDefault(logger)
 
-	cfg := &oauth2.Config{
-		ClientID:     clientId,
-		ClientSecret: clientSecret,
-		Endpoint: oauth2.Endpoint{
-			AuthURL:   fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/authorize", *workspace),
-			TokenURL:  fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/token", *workspace),
-			AuthStyle: oauth2.AuthStyleInParams,
-		},
-	}
+	if err := godotenv.Load(); err != nil {
+		slog.Error("Error loading .env file", "error", err)
 
-	tok := &oauth2.Token{
-		AccessToken:  accessToken,
-		RefreshToken: refreshToken,
-		TokenType:    "bearer",
-		Expiry:       time.Now().Add(-1 * time.Hour), // just pretend it's expired already, whatever, it'll fetch a new one.
+		return 1
 	}
 
+	salesforceEnvVarPrefix := "SALESFORCE_"
+	credentialsRegistry := utils.NewCredentialsRegistry()
+
+	envSchema := testUtils.EnvVarsReaders(salesforceEnvVarPrefix)
+	credentialsRegistry.AddReaders(
+		envSchema...,
+	)
+
+	salesforceWorkspace := credentialsRegistry.MustString(utils.WorkspaceRef)
+
+	cfg := utils.SalesforceOAuthConfigFromRegistry(credentialsRegistry)
+	tok := utils.SalesforceOauthTokenFromRegistry(credentialsRegistry)
 	ctx := context.Background()
 
-	// Create a new Salesforce connector, with a token provider that uses the sfdx CLI to fetch an access token.
 	sfc, err := connectors.Salesforce(
 		salesforce.WithClient(ctx, http.DefaultClient, cfg, tok),
-		salesforce.WithWorkspace(*workspace))
+		salesforce.WithWorkspace(salesforceWorkspace))
 	if err != nil {
 		slog.Error("Error creating Salesforce connector", "error", err)
 
diff --git a/test/utils/creds.go b/test/utils/creds.go
index 4f3a3341..ade8ee9a 100644
--- a/test/utils/creds.go
+++ b/test/utils/creds.go
@@ -1,23 +1,70 @@
 package utils
 
 import (
-	"os"
+	"fmt"
 
-	"github.com/spyzhov/ajson"
+	"github.com/amp-labs/connectors/utils"
 )
 
-// Credentials returns the credentials from the creds.json file.
-func Credentials(defaultFileName string) (*ajson.Node, error) {
-	fileName := defaultFileName
-
-	if fn, ok := os.LookupEnv("CREDENTIALS_FILE"); ok {
-		fileName = fn
+// preset values from JSON file from schema equal to response from reference: https://docs.withampersand.com/reference/getconnection
+func JSONFileReaders(filePath string) []utils.Reader {
+	schema := []utils.Reader{
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.AccessToken.Token",
+			CredKey:  utils.AccessToken,
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.RefreshToken.Token",
+			CredKey:  utils.RefreshToken,
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.providerApp.clientId",
+			CredKey:  utils.ClientId,
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.providerApp.clientSecret",
+			CredKey:  utils.ClientSecret,
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.providerWorkspaceRef",
+			CredKey:  utils.WorkspaceRef,
+		},
+		&utils.JSONReader{
+			FilePath: filePath,
+			JSONPath: "$.providerApp.provider",
+			CredKey:  utils.Provider,
+		},
 	}
 
-	byteValue, err := os.ReadFile(fileName)
-	if err != nil {
-		return nil, err
-	}
+	return schema
+}
 
-	return ajson.Unmarshal(byteValue)
+func EnvVarsReaders(prefix string) []utils.Reader {
+	return []utils.Reader{
+		&utils.EnvReader{
+			EnvName: fmt.Sprintf("%sACCESS_TOKEN", prefix),
+			CredKey: utils.AccessToken,
+		},
+		&utils.EnvReader{
+			EnvName: fmt.Sprintf("%sREFRESH_TOKEN", prefix),
+			CredKey: utils.RefreshToken,
+		},
+		&utils.EnvReader{
+			EnvName: fmt.Sprintf("%sCLIENT_ID", prefix),
+			CredKey: utils.ClientId,
+		},
+		&utils.EnvReader{
+			EnvName: fmt.Sprintf("%sCLIENT_SECRET", prefix),
+			CredKey: utils.ClientSecret,
+		},
+		&utils.EnvReader{
+			EnvName: fmt.Sprintf("%sWORKSPACE_REF", prefix),
+			CredKey: utils.WorkspaceRef,
+		},
+	}
 }
diff --git a/test/utils/hubspot.go b/test/utils/hubspot.go
deleted file mode 100644
index b069ba9c..00000000
--- a/test/utils/hubspot.go
+++ /dev/null
@@ -1,85 +0,0 @@
-package utils
-
-import (
-	"context"
-	"net/http"
-	"time"
-
-	"github.com/amp-labs/connectors"
-	"github.com/amp-labs/connectors/hubspot"
-	"github.com/spyzhov/ajson"
-	"golang.org/x/oauth2"
-)
-
-// GetHubspotConnector returns a Hubspot connector.
-func GetHubspotConnector(ctx context.Context, defaultFileName string) *hubspot.Connector {
-	creds, err := Credentials(defaultFileName)
-	if err != nil {
-		Fail("error getting credentials", "error", err)
-	}
-
-	cfg, tok := GetOAuthInfo(creds)
-
-	conn, err := connectors.Hubspot(
-		hubspot.WithClient(ctx, http.DefaultClient, cfg, tok),
-		hubspot.WithModule(hubspot.ModuleCRM))
-	if err != nil {
-		Fail("error creating hubspot connector", "error", err)
-	}
-
-	return conn
-}
-
-// GetOAuthInfo returns the OAuth2 config and token from the creds.json file.
-func GetOAuthInfo(creds *ajson.Node) (*oauth2.Config, *oauth2.Token) {
-	config := creds.MustObject()
-
-	clientId, found := config["CLIENT_ID"]
-	if !found {
-		Fail("CLIENT_ID not found in creds")
-	}
-
-	clientSecret, found := config["CLIENT_SECRET"]
-
-	if !found {
-		Fail("CLIENT_SECRET not found in creds")
-	}
-
-	// Treat this as optional, since we can get a new one if needed.
-	accessToken, _ := config["ACCESS_TOKEN"]
-
-	refreshToken, found := config["REFRESH_TOKEN"]
-
-	if !found {
-		Fail("REFRESH_TOKEN not found in creds")
-	}
-
-	cfg := &oauth2.Config{
-		ClientID:     clientId.MustString(),
-		ClientSecret: clientSecret.MustString(),
-		RedirectURL:  "http://localhost:8080/callbacks/v1/oauth",
-		Endpoint: oauth2.Endpoint{
-			AuthURL:   "https://app.hubspot.com/oauth/authorize",
-			TokenURL:  "https://api.hubapi.com/oauth/v1/token",
-			AuthStyle: oauth2.AuthStyleInParams,
-		},
-		Scopes: []string{
-			"crm.objects.contacts.read",
-			"crm.objects.contacts.write",
-			"crm.objects.deals.read",
-			"crm.objects.line_items.read",
-			"oauth",
-			"crm.objects.companies.read",
-			"tickets",
-		},
-	}
-
-	tok := &oauth2.Token{
-		AccessToken:  accessToken.MustString(),
-		RefreshToken: refreshToken.MustString(),
-		TokenType:    "bearer",
-		Expiry:       time.Now().Add(-1 * time.Hour), // just pretend it's expired already, whatever, it'll fetch a new one.
-	}
-
-	return cfg, tok
-}
diff --git a/utils/credentials_test.go b/utils/credentialRegistry_test.go
similarity index 85%
rename from utils/credentials_test.go
rename to utils/credentialRegistry_test.go
index bda4c888..10d7ba4e 100644
--- a/utils/credentials_test.go
+++ b/utils/credentialRegistry_test.go
@@ -39,6 +39,10 @@ var testPreset = []Reader{
 		JSONPath: "$['providers'][0]['number']",
 		CredKey:  "ProviderNumber",
 	},
+	&ValueReader{
+		Val:     "myValue",
+		CredKey: "MyValue",
+	},
 }
 
 func TestCredentialOptions(t *testing.T) {
@@ -77,4 +81,19 @@ func TestCredentialOptions(t *testing.T) {
 	useToken, err := opts.GetBool("UseToken")
 	require.NoError(t, err)
 	require.True(t, useToken)
+
+	myValue, err := opts.GetString("MyValue")
+	require.NoError(t, err)
+	require.Equal(t, "myValue", myValue)
+
+	require.Error(t,
+		opts.AddReader(
+			&ValueReader{},
+		),
+	)
+	require.Error(t,
+		opts.AddReaders(
+			[]Reader{&ValueReader{}}...,
+		),
+	)
 }