Repository: amp-labs/connectors
Pull request #66 (2024-02-12 21:48:07)

[ENG-668] Add generic basic connector creation
* This PR allows a caller to construct a connector given there is some basic configuration for the request provider inside the provider catalog. 
* Updates readme
* Disallows creation of a basic connector for salesforce/hubspot using these methods as they currently don't use the catalog to set their internal state

codesee-maps[bot]: <details open>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=66&src=https%3A%2F%2Fgithub.com%2Famp-labs%2Fconnectors">Click on the image</a> for a new way to code review</b>
</summary>

####

[![Review these changes using an interactive CodeSee Map](https://s3.us-east-2.amazonaws.com/maps.codesee.io/images/github/amp-labs/connectors/66/029c5221/e1ffcf74f86117b641cc0aa0dd9d3bc1fc21c7f0.svg)](https://app.codesee.io/r/reviews?pr=66&src=https%3A%2F%2Fgithub.com%2Famp-labs%2Fconnectors)

#### Legend
<img src="https://app.codesee.io/Legend_v3.png" alt="CodeSee Map legend" width="220" height="156">
</details>

Patch:
diff --git a/README.md b/README.md
index cfbe0c59..58c16b76 100644
--- a/README.md
+++ b/README.md
@@ -68,6 +68,11 @@ func main() {
   }
 }
 ```
+
+## Provider catalog
+
+The provider catalog is a YAML file that contains information about all SaaS providers that Ampersand supports. It contains information such as the base URL, and the operations that the provider supports. This is currently stored in the `providers/providers.yaml` file.
+
 ## Ways to initialize a Connector
 
 There are 3 ways to initialize a Connector:
@@ -93,3 +98,31 @@ client, err := connectors.Salesforce.New(
 ```go
 client, err := connectors.New("salesforce", map[string]any{"workspace": "salesforce-instance-name"})
 ```
+
+## Basic connectors
+
+Basic connectors allow you to proxy through requests to a SaaS provider via Ampersand. 
+
+#### Adding a new provider
+
+To add a new basic connector that allows proxying through the ampersand platform, you need to add a new entry to the `providers.yaml` file. The entry should have some required fields which are tagged with `validate: required` in the `providers/types.go` file.
+
+
+### Initialization
+
+**Note**: If your provider requires variables to be replaced in the catalog (providers.yaml), use the `WithCatalogSubstitutions` option to replace placeholders with actual values. 
+If you don't, the provider info will be incomplete and the connector will not work. 
+
+For example, a provider may use `{{workspace}}` in an option (maybe in the base URL) which needs to be replaced with an actual customer instance name. In that case, you would initialize the connector like this:
+
+```go
+conn, err := basic.NewConnector(
+    providers.SomeProvider,
+    basic.WithClient(context.Background(), http.DefaultClient, cfg, tok),
+    
+    // WithCatalogSubstitutions allows you to replace placeholders in the catalog (providers.yaml) with actual values.
+    basic.WithCatalogSubstitutions(map[string]string{
+		"workspace": "customer-instance-ref"
+	}),
+)
+```
diff --git a/basic.go b/basic.go
deleted file mode 100644
index 4371531c..00000000
--- a/basic.go
+++ /dev/null
@@ -1,22 +0,0 @@
-package connectors
-
-import (
-	"fmt"
-	"io"
-
-	"github.com/amp-labs/connectors/common"
-	"github.com/amp-labs/connectors/providers"
-)
-
-// BasicConnector is an interface that can be used to implement a connector with
-// basic configuration about the provider.
-type BasicConnector interface {
-	fmt.Stringer
-	io.Closer
-
-	// HTTPClient returns the underlying HTTP client. This is useful for proxy requests.
-	HTTPClient() *common.HTTPClient
-
-	// Provider returns the connector provider.
-	Provider() providers.Provider
-}
diff --git a/basic/client.go b/basic/client.go
new file mode 100644
index 00000000..4bab3a3a
--- /dev/null
+++ b/basic/client.go
@@ -0,0 +1,13 @@
+package basic
+
+import (
+	"github.com/amp-labs/connectors/common"
+)
+
+func (c *Connector) HTTPClient() *common.HTTPClient {
+	return c.Client.HTTPClient
+}
+
+func (c *Connector) Close() error {
+	return nil
+}
diff --git a/basic/connector.go b/basic/connector.go
new file mode 100644
index 00000000..156f4beb
--- /dev/null
+++ b/basic/connector.go
@@ -0,0 +1,68 @@
+package basic
+
+import (
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/providers"
+)
+
+// Connector is a Basic connector.
+type Connector struct {
+	ProviderInfo *providers.ProviderInfo
+	Client       *common.JSONHTTPClient
+	provider     providers.Provider
+}
+
+// NewConnector returns a new Basic connector.
+func NewConnector(
+	provider providers.Provider,
+	opts ...Option,
+) (conn *Connector, outErr error) {
+	defer func() {
+		if re := recover(); re != nil {
+			tmp, ok := re.(error)
+			if !ok {
+				panic(re)
+			}
+
+			outErr = tmp
+			conn = nil
+		}
+	}()
+
+	// Set up basic params
+	params := &basicParams{}
+	params.provider = provider
+
+	// Apply options & verify
+	for _, opt := range opts {
+		opt(params)
+	}
+
+	var err error
+
+	params, err = params.prepare()
+	if err != nil {
+		return nil, err
+	}
+
+	// Create connector
+	conn = &Connector{
+		provider: params.provider,
+		Client:   params.client,
+	}
+
+	// Read provider info & replace catalog variables with given substitutions, if any
+	providerInfo, err := providers.ReadConfig(conn.provider, &params.substitutions)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set provider info & http client options
+	conn.ProviderInfo = providerInfo
+	conn.Client.HTTPClient.ErrorHandler = conn.interpretError
+
+	// Set base URL
+	conn.Client.HTTPClient.Base = conn.ProviderInfo.BaseURL
+
+	return conn, nil
+}
diff --git a/basic/errors.go b/basic/errors.go
new file mode 100644
index 00000000..b14c12ae
--- /dev/null
+++ b/basic/errors.go
@@ -0,0 +1,11 @@
+package basic
+
+import (
+	"net/http"
+
+	"github.com/amp-labs/connectors/common"
+)
+
+func (c *Connector) interpretError(res *http.Response, body []byte) error {
+	return common.InterpretError(res, body)
+}
diff --git a/basic/params.go b/basic/params.go
new file mode 100644
index 00000000..c0b9ac1e
--- /dev/null
+++ b/basic/params.go
@@ -0,0 +1,80 @@
+package basic
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/amp-labs/connectors/providers"
+	"golang.org/x/oauth2"
+)
+
+var (
+	// ErrMissingClient is returned when a connector is created without a client.
+	ErrMissingClient = errors.New("missing client")
+
+	// ErrMissingProvider is returned when a connector is created without a provider.
+	ErrMissingProvider = errors.New("missing provider")
+)
+
+type Option func(*basicParams)
+
+type basicParams struct {
+	provider      providers.Provider
+	client        *common.JSONHTTPClient
+	substitutions map[string]string
+}
+
+func (p *basicParams) prepare() (*basicParams, error) {
+	if p.provider == "" {
+		return nil, ErrMissingProvider
+	}
+
+	if p.client == nil {
+		return nil, ErrMissingClient
+	}
+
+	return p, nil
+}
+
+// WithCatalogSubstitutions sets the provider values to use while making substitutions &
+// reading from providers.yaml. If the provider values are not set, the connector
+// will use error out.
+func WithCatalogSubstitutions(substitutions map[string]string) Option {
+	return func(params *basicParams) {
+		params.substitutions = substitutions
+	}
+}
+
+// WithClient sets the http client to use for the connector.
+func WithClient(ctx context.Context, client *http.Client, config *oauth2.Config, token *oauth2.Token,
+	opts ...common.OAuthOption,
+) Option {
+	return func(params *basicParams) {
+		options := []common.OAuthOption{
+			common.WithClient(client),
+			common.WithOAuthConfig(config),
+			common.WithOAuthToken(token),
+		}
+
+		oauthClient, err := common.NewOAuthHTTPClient(ctx, append(options, opts...)...)
+		if err != nil {
+			panic(err) // caught in NewConnector
+		}
+
+		WithAuthenticatedClient(oauthClient)(params)
+	}
+}
+
+// WithAuthenticatedClient sets the http client to use for the connector. Its usage is optional.
+func WithAuthenticatedClient(client common.AuthenticatedHTTPClient) Option {
+	return func(params *basicParams) {
+		params.client = &common.JSONHTTPClient{
+			HTTPClient: &common.HTTPClient{
+				Client:       client,
+				ErrorHandler: common.InterpretError,
+			},
+		}
+	}
+}
diff --git a/basic/provider.go b/basic/provider.go
new file mode 100644
index 00000000..6f33a72d
--- /dev/null
+++ b/basic/provider.go
@@ -0,0 +1,9 @@
+package basic
+
+import (
+	"github.com/amp-labs/connectors/providers"
+)
+
+func (c *Connector) Provider() providers.Provider {
+	return c.provider
+}
diff --git a/basic/string.go b/basic/string.go
new file mode 100644
index 00000000..6d102344
--- /dev/null
+++ b/basic/string.go
@@ -0,0 +1,9 @@
+package basic
+
+import (
+	"fmt"
+)
+
+func (c *Connector) String() string {
+	return fmt.Sprintf("%s.Connector", c.Provider())
+}
diff --git a/connectors.go b/connectors.go
index 83da9db4..a5419f42 100644
--- a/connectors.go
+++ b/connectors.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"strings"
 
 	"github.com/amp-labs/connectors/common"
@@ -12,6 +13,19 @@ import (
 	"github.com/amp-labs/connectors/salesforce"
 )
 
+// BasicConnector is an interface that can be used to implement a connector with
+// basic configuration about the provider.
+type BasicConnector interface {
+	fmt.Stringer
+	io.Closer
+
+	// HTTPClient returns the underlying HTTP client. This is useful for proxy requests.
+	HTTPClient() *common.HTTPClient
+
+	// Provider returns the connector provider.
+	Provider() providers.Provider
+}
+
 // Connector is an interface that all connectors must implement.
 type Connector interface {
 	BasicConnector
diff --git a/providers/catalog.go b/providers/catalog.go
index be54be20..f6801499 100644
--- a/providers/catalog.go
+++ b/providers/catalog.go
@@ -9,6 +9,7 @@ import (
 	"runtime"
 	"strings"
 
+	"github.com/go-playground/validator"
 	"gopkg.in/yaml.v3"
 )
 
@@ -35,6 +36,12 @@ func ReadConfig(provider Provider, substitutions *map[string]string) (*ProviderI
 		return nil, ErrProviderCatalogNotFound
 	}
 
+	// Validate the provider configuration
+	v := validator.New()
+	if err := v.Struct(&providerConfig); err != nil {
+		return nil, err
+	}
+
 	// Apply substitutions to the provider configuration values which contain variables in the form of {{var}}.
 	err = substituteStruct(&providerConfig, substitutions)
 	if err != nil {
diff --git a/providers/providers.yaml b/providers/providers.yaml
index 6ff3c1cf..9fd1bde0 100644
--- a/providers/providers.yaml
+++ b/providers/providers.yaml
@@ -36,7 +36,7 @@ providers:
       write: true
 
   linkedIn:
-    baseUrl: https://api.linkedin.com
+    baseUrl: https://api.linkedin.com/v2
     authType: oauth2
     support:
       bulkWrite: false
diff --git a/providers/types.go b/providers/types.go
index b6526ee8..701d8ff0 100644
--- a/providers/types.go
+++ b/providers/types.go
@@ -9,10 +9,10 @@ type Catalog struct {
 // in the configuration. The substitution is only done on string fields. If you want to use pointers in the struct,
 // you might have to update the code to handle it.
 type ProviderInfo struct {
-	Support   ConnectorSupport `yaml:"support"`
-	AuthType  AuthType         `yaml:"authType"`
+	Support   ConnectorSupport `validate:"required" yaml:"support"`
+	AuthType  AuthType         `validate:"required" yaml:"authType"`
+	BaseURL   string           `validate:"required" yaml:"baseUrl"`
 	OauthOpts OauthOpts        `yaml:"oauthOpts"`
-	BaseURL   string           `yaml:"baseUrl"`
 }
 
 type ConnectorSupport struct {
diff --git a/salesforce/events.go b/salesforce/events.go
new file mode 100644
index 00000000..d97445f2
--- /dev/null
+++ b/salesforce/events.go
@@ -0,0 +1,274 @@
+package salesforce
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/amp-labs/connectors/common"
+	"github.com/spyzhov/ajson"
+)
+
+// nolint: revive
+const (
+	AllowedManagedPackageNamespaces NamedCredentialParameterType = "AllowedManagedPackageNamespaces"
+	ClientCertificate               NamedCredentialParameterType = "ClientCertificate"
+	HttpHeader                      NamedCredentialParameterType = "HttpHeader"
+	OutboundNetworkConnection       NamedCredentialParameterType = "OutboundNetworkConnection"
+	Url                             NamedCredentialParameterType = "Url"
+)
+
+type SFAPIResponseBody struct {
+	Id       string        `json:"id"`
+	Success  bool          `json:"success"`
+	Errors   []interface{} `json:"errors"`
+	Infos    []interface{} `json:"infos"`
+	Warnings []interface{} `json:"warnings"`
+}
+
+// nolint:tagliatelle
+type EventChannel struct {
+	Id       string                `json:"Id,omitempty"`
+	FullName string                `json:"FullName"`
+	Metadata *EventChannelMetadata `json:"Metadata"`
+}
+
+type EventChannelMetadata struct {
+	ChannelType string `json:"channelType"`
+	Label       string `json:"label"`
+}
+
+// nolint:tagliatelle
+type EventRelayConfig struct {
+	Id                      string                    `json:"Id,omitempty"`
+	FullName                string                    `json:"FullName,omitempty"`
+	Metadata                *EventRelayConfigMetadata `json:"Metadata,omitempty"`
+	DeveloperName           string                    `json:"DeveloperName,omitempty"`
+	DestinationResourceName string                    `json:"DestinationResourceName,omitempty"`
+	EventChannel            string                    `json:"EventChannel,omitempty"`
+}
+
+type EventRelayConfigMetadata struct {
+	DestinationResourceName string `json:"destinationResourceName,omitempty"`
+	EventChannel            string `json:"eventChannel,omitempty"`
+	State                   string `json:"state,omitempty"`
+}
+
+// nolint:tagliatelle
+type EventChannelMember struct {
+	Id       string                      `json:"Id,omitempty"`
+	FullName string                      `json:"FullName"`
+	Metadata *EventChannelMemberMetadata `json:"Metadata"`
+}
+
+type EventChannelMemberMetadata struct {
+	EventChannel   string `json:"eventChannel"`
+	SelectedEntity string `json:"selectedEntity,omitempty"`
+}
+
+type NamedCredentialParameterType string
+
+// nolint: tagliatelle
+type ToolingApiBaseParams struct {
+	DeveloperName   string `json:"DeveloperName,omitempty"`
+	Language        string `json:"Language,omitempty"`
+	ManageableState string `json:"ManageableState,omitempty"`
+	MasterLabel     string `json:"MasterLabel,omitempty"`
+	NamespacePrefix string `json:"NamespacePrefix,omitempty"`
+}
+
+type CalloutOptions struct {
+	AllowMergeFieldsInBody      bool `json:"allowMergeFieldsInBody"`
+	AllowMergeFieldsInHeader    bool `json:"allowMergeFieldsInHeader"`
+	GenerateAuthorizationHeader bool `json:"generateAuthorizationHeader"`
+}
+
+type NamedCredentialParameter struct {
+	Certificate               string `json:"certificate"`
+	Description               string `json:"description"`
+	ExternalCredential        string `json:"externalCredential"`
+	OutboundNetworkConnection string `json:"outboundNetworkConnection"`
+	ParameterName             string `json:"parameterName"`
+	ParameterType             string `json:"parameterType"`
+	ParameterValue            string `json:"parameterValue"`
+	SequenceNumber            int    `json:"sequenceNumber"`
+}
+
+// nolint: lll
+type NamedCredentialMetadata struct {
+	AllowMergeFieldsInBody      bool                        `json:"allowMergeFieldsInBody,omitempty"`
+	AllowMergeFieldsInHeader    bool                        `json:"allowMergeFieldsInHeader,omitempty"`
+	GenerateAuthorizationHeader bool                        `json:"generateAuthorizationHeader,omitempty"`
+	FullName                    string                      `json:"fullName,omitempty"                    validate:"required"`
+	Label                       string                      `json:"label,omitempty"`
+	NamedCredentialParameters   []*NamedCredentialParameter `json:"namedCredentialParameters,omitempty"`
+	NamedCredentialType         string                      `json:"namedCredentialType,omitempty"`
+
+	// Below are deprecated fields, but still in use in SF
+	AuthProvider             string `json:"authProvider,omitempty"`
+	AuthTokenEndpointUrl     string `json:"authTokenEndpointUrl,omitempty"` // nolint: revive
+	AwsAccessKey             string `json:"awsAccessKey,omitempty"`
+	AwsAccessSecret          string `json:"awsAccessSecret,omitempty"`
+	AwsRegion                string `json:"awsRegion,omitempty"`
+	AwsService               string `json:"awsService,omitempty"`
+	Certificate              string `json:"certificate,omitempty"`
+	Endpoint                 string `json:"endpoint,omitempty"`
+	JwtAudience              string `json:"jwtAudience,omitempty"`
+	JwtFormulaSubject        string `json:"jwtFormulaSubject,omitempty"`
+	JwtIssuer                string `json:"jwtIssuer,omitempty"`
+	JwtSigningCertificate    string `json:"jwtSigningCertificate,omitempty"`
+	JwtTextSubject           string `json:"jwtTextSubject,omitempty"`
+	JwtValidityPeriodSeconds int    `json:"jwtValidityPeriodSeconds,omitempty"`
+	OauthRefreshToken        string `json:"oauthRefreshToken,omitempty"`
+	OauthScope               string `json:"oauthScope,omitempty"`
+	OauthToken               string `json:"oauthToken,omitempty"`
+	Password                 string `json:"password,omitempty"`
+	PrincipalType            string `json:"principalType,omitempty"`
+	Protocol                 string `json:"protocol,omitempty"`
+	Username                 string `json:"username,omitempty"`
+}
+
+// nolint:tagliatelle
+type NamedCredential struct {
+	FullName string                   `json:"FullName"`
+	Metadata *NamedCredentialMetadata `json:"Metadata"`
+
+	// below exist in response, but not in request
+	Id string `json:"Id,omitempty"`
+}
+
+func (n *NamedCredential) DestinationResourceName() string {
+	return fmt.Sprint("callout:", n.FullName)
+}
+
+// nolint: lll
+// https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_platformeventchannel.htm
+func (c *Connector) CreateEventChannel(ctx context.Context, channel *EventChannel) (*EventChannel, error) {
+	res, err := c.postToSFAPI(ctx, channel, "tooling/sobjects/PlatformEventChannel", "PlatformEventChannel")
+	if err != nil {
+		return nil, err
+	}
+
+	channel.Id = res.Id
+
+	return channel, nil
+}
+
+// nolint: lll
+// https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_platformeventchannelmember.htm
+func (c *Connector) CreateEventChannelMember(
+	ctx context.Context,
+	member *EventChannelMember,
+) (*EventChannelMember, error) {
+	res, err := c.postToSFAPI(ctx, member, "tooling/sobjects/PlatformEventChannelMember", "EventChannelMember")
+	if err != nil {
+		return nil, err
+	}
+
+	member.Id = res.Id
+
+	return member, nil
+}
+
+// nolint: lll
+// https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_eventrelayconfig.htm?q=EventRelayConfig
+func (c *Connector) CreateEventRelayConfig(
+	ctx context.Context,
+	cfg *EventRelayConfig,
+) (*EventRelayConfig, error) {
+	res, err := c.postToSFAPI(ctx, cfg, "/tooling/sobjects/EventRelayConfig", "EventRelayConfig")
+	if err != nil {
+		return nil, err
+	}
+
+	cfg.Id = res.Id
+
+	return cfg, nil
+}
+
+// nolint: lll
+// https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_eventrelayconfig.htm?q=EventRelayConfig
+func (c *Connector) RunEventRelay(ctx context.Context, cfg *EventRelayConfig) error {
+	location, err := joinURLPath("tooling/sobjects/EventRelayConfig", cfg.Id)
+	if err != nil {
+		return err
+	}
+
+	config := &EventRelayConfig{
+		FullName: cfg.FullName,
+		Metadata: &EventRelayConfigMetadata{
+			State: "RUN",
+		},
+	}
+
+	_, err = c.patch(ctx, location, config) // patch returns no content with 204. If it fails, it will return an error.
+	if err != nil {
+		slog.Error("Run EventRelayConfig", "error", err)
+
+		return err
+	}
+
+	return nil
+}
+
+// nolint: lll
+// https://developer.salesforce.com/docs/atlas.en-us.chatterapi.meta/chatterapi/connect_responses_organization.htm?q=organization
+func (c *Connector) getOrganization(ctx context.Context) (map[string]*ajson.Node, error) {
+	resp, err := c.get(ctx, "connect/organization")
+	if err != nil {
+		return nil, err
+	}
+
+	return resp.Body.GetObject()
+}
+
+func (c *Connector) GetOrganizationId(ctx context.Context) (string, error) {
+	org, err := c.getOrganization(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return org["orgId"].MustString(), nil
+}
+
+func GetRemoteResource(orgId, channelId string) string {
+	return fmt.Sprintf("aws.partner/salesforce.com/%s/%s", orgId, channelId)
+}
+
+// nolint: lll
+// https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_namedcredential.htm
+func (c *Connector) CreateNamedCredential(ctx context.Context, creds *NamedCredential) (*NamedCredential, error) {
+	res, err := c.postToSFAPI(ctx, creds, "/tooling/sobjects/NamedCredential", "NamedCredential")
+	if err != nil {
+		return nil, err
+	}
+
+	creds.Id = res.Id
+
+	return creds, nil
+}
+
+type Credential interface {
+	DestinationResourceName() string
+}
+
+func (c *Connector) postToSFAPI(ctx context.Context, body any, path string, entity string) (*SFAPIResponseBody, error) {
+	location, err := joinURLPath(c.BaseURL, path)
+	if err != nil {
+		return nil, err
+	}
+
+	fmt.Println("location: ", location)
+
+	resp, err := c.post(ctx, location, body)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := common.UnmarshalJSON[SFAPIResponseBody](resp)
+	if res.Warnings != nil && len(res.Warnings) > 0 {
+		slog.Warn(entity, "warnings", res.Warnings)
+	}
+
+	return res, err
+}
diff --git a/utils/credentials.go b/utils/credentials.go
new file mode 100644
index 00000000..10b0bb4a
--- /dev/null
+++ b/utils/credentials.go
@@ -0,0 +1,402 @@
+// nolint: ireturn
+package utils
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"log/slog"
+	"os"
+
+	"github.com/spyzhov/ajson"
+)
+
+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")
+)
+
+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", ErrEmptyStringEnvVar, r.EnvName)
+	}
+
+	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", 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 (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) 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
+}


Test patch:
diff --git a/providers/catalog_test.go b/providers/catalog_test.go
index 6f726b18..cea8de43 100644
--- a/providers/catalog_test.go
+++ b/providers/catalog_test.go
@@ -79,7 +79,7 @@ func TestReadConfig(t *testing.T) { //nolint:funlen
 				},
 				AuthType:  AuthTypeOAuth2,
 				OauthOpts: OauthOpts{},
-				BaseURL:   "https://api.linkedin.com",
+				BaseURL:   "https://api.linkedin.com/v2",
 			},
 			expectedErr: nil,
 		},
@@ -101,27 +101,26 @@ func TestReadConfig(t *testing.T) { //nolint:funlen
 			t.Parallel()
 
 			config, err := ReadConfig(tc.provider, &tc.substitutions)
-			t.Logf("Test case: %s", tc.description)
 
 			if !errors.Is(err, tc.expectedErr) {
-				t.Errorf("Expected error: %v, but got: %v", tc.expectedErr, err)
+				t.Errorf("[%s] Expected error: %v, but got: %v", tc.description, tc.expectedErr, err)
 			}
 
 			if tc.expectedErr == nil && config != nil {
 				if config.Support != tc.expected.Support {
-					t.Errorf("Expected support: %v, but got: %v", tc.expected.Support, config.Support)
+					t.Errorf("[%s] Expected support: %v, but got: %v", tc.description, tc.expected.Support, config.Support)
 				}
 
 				if config.AuthType != tc.expected.AuthType {
-					t.Errorf("Expected auth: %v, but got: %v", tc.expected.AuthType, config.AuthType)
+					t.Errorf("[%s] Expected auth: %v, but got: %v", tc.description, tc.expected.AuthType, config.AuthType)
 				}
 
 				if config.OauthOpts != tc.expected.OauthOpts {
-					t.Errorf("Expected auth options: %v, but got: %v", tc.expected.OauthOpts, config.OauthOpts)
+					t.Errorf("[%s] Expected auth options: %v, but got: %v", tc.description, tc.expected.OauthOpts, config.OauthOpts)
 				}
 
 				if config.BaseURL != tc.expected.BaseURL {
-					t.Errorf("Expected base URL: %s, but got: %s", tc.expected.BaseURL, config.BaseURL)
+					t.Errorf("[%s] Expected base URL: %s, but got: %s", tc.description, tc.expected.BaseURL, config.BaseURL)
 				}
 			}
 		})
diff --git a/test/salesforce/event-flow/main.go b/test/salesforce/event-flow/main.go
new file mode 100644
index 00000000..aba0eb6f
--- /dev/null
+++ b/test/salesforce/event-flow/main.go
@@ -0,0 +1,366 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/amp-labs/connectors"
+	"github.com/amp-labs/connectors/salesforce"
+	"github.com/amp-labs/connectors/test"
+	"golang.org/x/oauth2"
+)
+
+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
+
+	salesforceSubdomain := creds.Subdomain
+
+	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", salesforceSubdomain),
+			TokenURL:  fmt.Sprintf("https://%s.my.salesforce.com/services/oauth2/token", salesforceSubdomain),
+			AuthStyle: oauth2.AuthStyleInParams,
+		},
+	}
+
+	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.
+	}
+
+	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.WithSubdomain(salesforceSubdomain))
+	if err != nil {
+		slog.Error("Error creating Salesforce connector", "error", err)
+
+		return
+	}
+
+	defer func() {
+		_ = sfc.Close()
+	}()
+
+	uniqueString := strconv.Itoa(int(time.Now().UnixMilli()))
+
+	namedCred, err := TestCredentialsLegacy(sfc, ctx, uniqueString)
+	if err != nil {
+		return
+	}
+
+	testCDC(sfc, ctx, namedCred, uniqueString)
+
+	peName := "Employee__e" // this should be created from UI
+	testPE(sfc, ctx, namedCred, peName, uniqueString)
+}
+
+func testCDC(sfc *salesforce.Connector, ctx context.Context, creds salesforce.Credential, uniqueString string) {
+	fmt.Println("-----------------Testing Change Data Capture-----------------")
+
+	cdcChannel, err := TestCDCChannel(sfc, ctx, "TestCDCChannel"+uniqueString)
+	if err != nil {
+		return
+	}
+
+	evtCfg, err := TestEventRelayConfig(sfc, ctx, creds, cdcChannel, "CDC_"+uniqueString)
+	if err != nil {
+		return
+	}
+
+	objectName := "Account"
+
+	_, err = testChangeDataCaptureChannelMembership(sfc, ctx, cdcChannel.FullName, objectName)
+	if err != nil {
+		return
+	}
+
+	if sfc.RunEventRelay(ctx, evtCfg) != nil {
+		slog.Error("Error running event relay", "error", err)
+		return
+	}
+
+	slog.Info("Event relay config updated", "state", "RUN")
+
+	orgId, err := sfc.GetOrganizationId(ctx)
+	if err != nil {
+		slog.Error("Failed to get orgId", "error", err)
+		return
+	}
+
+	remoteResource := salesforce.GetRemoteResource(orgId, cdcChannel.FullName)
+	printWithField("RemoteResource", "resource", remoteResource)
+}
+
+func testPE(sfc *salesforce.Connector, ctx context.Context, creds salesforce.Credential, eventName string, uniqueString string) {
+	fmt.Println("-----------------Testing Platform Event-----------------")
+
+	peChannel, err := TestPlatformEventChannel(sfc, ctx, "TestPEChannel"+uniqueString)
+	if err != nil {
+		return
+	}
+
+	evtCfg, err := TestEventRelayConfig(sfc, ctx, creds, peChannel, "PE_"+uniqueString)
+	if err != nil {
+		return
+	}
+
+	objectName := "Account"
+
+	_, err = testPlatformEventChannelMembership(sfc, ctx, eventName, peChannel.FullName, objectName, uniqueString)
+	if err != nil {
+		return
+	}
+
+	if sfc.RunEventRelay(ctx, evtCfg) != nil {
+		return
+	}
+
+	slog.Info("Event relay config updated", "state", "RUN")
+
+	orgId, err := sfc.GetOrganizationId(ctx)
+	if err != nil {
+		slog.Error("Failed to get orgId", "error", err)
+		return
+	}
+
+	remoteResource := salesforce.GetRemoteResource(orgId, peChannel.FullName)
+	printWithField("RemoteResource", "resource", remoteResource)
+}
+
+func testPlatformEventChannelMembership(sfc *salesforce.Connector, ctx context.Context, peName string, channelName string, objectName string, uniqueString string) (*salesforce.EventChannelMember, error) {
+	rawPEName := getRawPEName(objectName)
+
+	rawChannelName := getRawChannelNameFromChannel(channelName) // TODO FIXME
+
+	member := &salesforce.EventChannelMember{
+		FullName: getPEChannelMembershipName(rawChannelName, rawPEName),
+		Metadata: &salesforce.EventChannelMemberMetadata{
+			EventChannel:   getChannelName(rawChannelName),
+			SelectedEntity: peName,
+		},
+	}
+
+	newChannelMember, err := sfc.CreateEventChannelMember(ctx, member)
+	if err != nil {
+		slog.Error("Error event channel member", "error", err)
+
+		return nil, err
+	}
+
+	printWithField("Event channel membership created", "member", newChannelMember)
+
+	return newChannelMember, nil
+}
+
+func testChangeDataCaptureChannelMembership(sfc *salesforce.Connector, ctx context.Context, channelName string, objecName string) (*salesforce.EventChannelMember, error) {
+	eventName := getCDCEventName(objecName)
+	rawChannelName := getRawChannelNameFromChannel(channelName)
+
+	member := &salesforce.EventChannelMember{
+		FullName: getCDCChannelMembershipName(rawChannelName, eventName),
+		Metadata: &salesforce.EventChannelMemberMetadata{
+			EventChannel:   getChannelName(rawChannelName),
+			SelectedEntity: eventName,
+		},
+	}
+
+	newChannelMember, err := sfc.CreateEventChannelMember(ctx, member)
+	if err != nil {
+		slog.Error("Error event channel member", "error", err)
+
+		return nil, err
+	}
+
+	printWithField("Event channel membership created", "member", newChannelMember)
+
+	return newChannelMember, nil
+}
+
+func TestCDCChannel(sfc *salesforce.Connector, ctx context.Context, channelName string) (*salesforce.EventChannel, error) {
+	channel := &salesforce.EventChannel{
+		FullName: getChannelName(channelName),
+		Metadata: &salesforce.EventChannelMetadata{
+			ChannelType: "data",
+			Label:       "Test change data capture Channel",
+		},
+	}
+
+	newChannel, err := sfc.CreateEventChannel(ctx, channel)
+	if err != nil {
+		slog.Error("Error creating data channel", "error", err)
+		return nil, err
+	}
+
+	printWithField("Data Event channel created", "body", newChannel)
+
+	return newChannel, nil
+}
+
+func TestPlatformEventChannel(sfc *salesforce.Connector, ctx context.Context, channelName string) (*salesforce.EventChannel, error) {
+	channel := &salesforce.EventChannel{
+		FullName: getChannelName(channelName),
+		Metadata: &salesforce.EventChannelMetadata{
+			ChannelType: "event",
+			Label:       "Test Event Channel",
+		},
+	}
+
+	newChannel, err := sfc.CreateEventChannel(ctx, channel)
+	if err != nil {
+		slog.Error("Error creating event channel", "error", err)
+
+		return nil, err
+	}
+
+	printWithField("Platform Event channel created", "body", newChannel)
+
+	return newChannel, nil
+}
+
+func TestEventRelayConfig(sfc *salesforce.Connector, ctx context.Context, cred salesforce.Credential, channel *salesforce.EventChannel, uniqueString string) (*salesforce.EventRelayConfig, error) {
+	evtCfg := &salesforce.EventRelayConfig{
+		FullName: "TestEventRelayConfig" + uniqueString,
+		Metadata: &salesforce.EventRelayConfigMetadata{
+			DestinationResourceName: cred.DestinationResourceName(),
+			EventChannel:            channel.FullName,
+		},
+	}
+
+	newEvtCfg, err := sfc.CreateEventRelayConfig(ctx, evtCfg)
+	if err != nil {
+		slog.Error("Error event relay config", "error", err)
+
+		return nil, err
+	}
+
+	printWithField("Event relay config created", "body", newEvtCfg)
+	printWithField("Event relay config metadata", "metadata", evtCfg.Metadata)
+
+	return newEvtCfg, nil
+}
+
+func testOrganizationId(sfc *salesforce.Connector, ctx context.Context) (string, error) {
+	fmt.Println("-----------------Testing Organization Id-----------------")
+
+	orgId, err := sfc.GetOrganizationId(ctx)
+	if err != nil {
+		slog.Error("Error querying org", "error", err)
+		return "", err
+	}
+
+	printWithField("Org Id created", "id", orgId)
+
+	return orgId, nil
+}
+
+func TestCredentialsLegacy(sfc *salesforce.Connector, ctx context.Context, uniqueString string) (*salesforce.NamedCredential, error) {
+	fmt.Println("-----------------Testing Named Credential Legacy-----------------")
+
+	namedCred := &salesforce.NamedCredential{
+		FullName: "TestNamedCredentialLegacy" + uniqueString,
+		Metadata: &salesforce.NamedCredentialMetadata{
+			GenerateAuthorizationHeader: true,
+			Label:                       "TestNamedCredentialLegacy" + uniqueString,
+
+			// below are legacy fields
+			Endpoint:      "arn:aws:us-east-2:381491976069",
+			PrincipalType: "NamedUser",
+			Protocol:      "NoAuthentication",
+		},
+	}
+
+	newNamedCred, err := sfc.CreateNamedCredential(ctx, namedCred)
+	if err != nil {
+		slog.Error("Error named cred", "error", err)
+		return nil, err
+	}
+
+	printWithField("Named credential created", "body", newNamedCred)
+
+	return newNamedCred, nil
+}
+
+func printWithField(header string, key string, v interface{}) {
+	slog.Info(header, key, fmt.Sprintf("%+v", v))
+}
+
+func isCustomObject(objName string) bool {
+	return strings.HasSuffix(objName, "__c")
+}
+
+func getRawObjectName(objName string) string {
+	return removeSuffix(objName, 3)
+}
+
+func getCDCEventName(objName string) string {
+	if isCustomObject(objName) {
+		return fmt.Sprintf("%s__ChangeEvent", getRawObjectName(objName))
+	}
+
+	return fmt.Sprintf("%sChangeEvent", objName)
+}
+
+func getChannelName(rawChannelName string) string {
+	return fmt.Sprintf("%s__chn", rawChannelName)
+}
+
+func removeSuffix(objName string, suffixLength int) string {
+	return objName[:len(objName)-suffixLength]
+}
+
+func getRawChannelNameFromChannel(channelName string) string {
+	if strings.HasSuffix(channelName, "__chn") {
+		return removeSuffix(channelName, 5)
+	}
+
+	return channelName
+}
+
+func getRawChannelNameFromObject(objectName string) string {
+	if strings.HasSuffix(objectName, "__e") {
+		return removeSuffix(objectName, 3)
+	}
+
+	return objectName
+}
+
+func getPEChannelMembershipName(channelName, eventName string) string {
+	return fmt.Sprintf("%s%sChannelEvent_e", channelName, eventName)
+}
+
+func getCDCChannelMembershipName(rawChannelName string, eventName string) string {
+	return fmt.Sprintf("%s_chn_%s", rawChannelName, eventName)
+}
+
+func getRawPEName(peName string) string {
+	return removeSuffix(peName, 3)
+}
diff --git a/utils/credentials_test.go b/utils/credentials_test.go
new file mode 100644
index 00000000..bda4c888
--- /dev/null
+++ b/utils/credentials_test.go
@@ -0,0 +1,80 @@
+//nolint:gochecknoglobals
+package utils
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+var testAccessTokenReader = &JSONReader{
+	FilePath: "testcred1.json",
+	JSONPath: "$['accessToken']",
+	CredKey:  "AccessToken",
+}
+
+var testPreset = []Reader{
+	&EnvReader{
+		EnvName: "TEST_ENV_CLIENT_ID",
+		CredKey: "ClientId",
+	},
+	&JSONReader{
+		FilePath: "testcred1.json",
+		JSONPath: "$['useToken']",
+		CredKey:  "UseToken",
+	},
+	&JSONReader{
+		FilePath: "testcred2.json",
+		JSONPath: "$['refreshToken']",
+		CredKey:  "RefreshToken",
+	},
+	&JSONReader{
+		FilePath: "testcred2.json",
+		JSONPath: "$['providers'][0]['name']",
+		CredKey:  "Provider",
+	},
+	&JSONReader{
+		FilePath: "testcred2.json",
+		JSONPath: "$['providers'][0]['number']",
+		CredKey:  "ProviderNumber",
+	},
+}
+
+func TestCredentialOptions(t *testing.T) {
+	t.Parallel()
+
+	if os.Setenv("TEST_ENV_CLIENT_ID", "clientId") != nil {
+		t.Fatal("Error setting environment variable")
+	}
+
+	opts := NewCredentialsRegistry()
+
+	require.NoError(t, opts.AddReader(testAccessTokenReader))
+
+	require.NoError(t, opts.AddReaders(testPreset...))
+
+	refreshToken, err := opts.GetString("RefreshToken")
+	require.NoError(t, err)
+	require.Equal(t, "refreshToken", refreshToken)
+
+	accessToken, err := opts.GetString("AccessToken")
+	require.NoError(t, err)
+	require.Equal(t, "accessToken", accessToken)
+
+	clientId, err := opts.GetString("ClientId")
+	require.NoError(t, err)
+	require.Equal(t, "clientId", clientId)
+
+	provider, err := opts.GetString("Provider")
+	require.NoError(t, err)
+	require.Equal(t, "provider", provider)
+
+	providerNumber, err := opts.GetFloat64("ProviderNumber")
+	require.NoError(t, err)
+	require.Equal(t, 3, int(providerNumber))
+
+	useToken, err := opts.GetBool("UseToken")
+	require.NoError(t, err)
+	require.True(t, useToken)
+}
diff --git a/utils/testcred1.json b/utils/testcred1.json
new file mode 100644
index 00000000..2dfeac54
--- /dev/null
+++ b/utils/testcred1.json
@@ -0,0 +1,4 @@
+{
+    "accessToken": "accessToken",
+    "useToken": true
+}
diff --git a/utils/testcred2.json b/utils/testcred2.json
new file mode 100644
index 00000000..c942e4f8
--- /dev/null
+++ b/utils/testcred2.json
@@ -0,0 +1,9 @@
+{
+    "refreshToken": "refreshToken",
+    "providers": [
+        {
+            "name": "provider",
+            "number": 3
+        }
+    ]
+}