payments/internal/connectors/engine/workflow/fetch_conversions_test.go at main

mirror of https://github.com/hanzo-fi/payments.git synced 2026-10-30 20:31:20 +00:00

Files

T

zeekay 7d727874d5 chore: rebrand Formance Payments -> Hanzo Payments (hanzo-fi)

- go.mod module path + all self-imports: github.com/formancehq/payments -> github.com/hanzo-fi/payments
  (1590 .go files / 3152 refs; 20 modules: root, pkg/client, pkg/domain, 18 ce/plugins).
  External dep kept: formancehq/go-libs.
- plugins: each ce/plugins/* consumed a *published* pkg/domain; added local
  'replace .../pkg/domain => ../../../pkg/domain' (consistent with root) so every
  standalone plugin module builds in-monorepo. compile-plugin.sh MOD var rebranded too.
- branding: README, LICENSE (dual MIT+EE; upstream copyright retained + Hanzo added),
  goreleaser (release images -> registry.hanzo.ai/hanzo-fi, OCI labels, homepage),
  proto go_package, renovate, speakeasy gen.yaml, connector-template *.gotpl.
- CI: removed Numary/Formance workflows + composite action; added canonical hanzo.yml +
  cicd.yml (hanzoai/ci, gated HANZO_CI_ENABLED) + secretless ci.yml (go build/vet).
- kept as data/compat keys: wire capability enums ALLOW_FORMANCE_*, generated SDK type
  client.Formance, domain field FormanceRedirectURL, OTEL_SERVICE_NAME, upstream base image.
- verified: go build ./... (default + -tags ee) + all 20 nested modules green.

2026-07-05 13:08:55 -07:00

383 lines

14 KiB

Go

Raw Permalink Blame History

 package workflow
 import (
 	"context"
 	"errors"
 	"fmt"
 	"math/big"
 	"time"
 	"github.com/hanzo-fi/payments/internal/connectors/engine/activities"
 	"github.com/hanzo-fi/payments/pkg/domain/models"
 	"github.com/stretchr/testify/mock"
 	"go.temporal.io/sdk/temporal"
 )
 func (s *UnitTestSuite) Test_FetchNextConversions_WithoutInstance_Success() {
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   fmt.Sprintf("%s-%s", models.CAPABILITY_FETCH_CONVERSIONS.String(), "1"),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{},
 			NewState:    []byte(`{}`),
 			HasMore:     false,
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.StorageStatesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID: s.connectorID,
 		FromPayload: &FromPayload{
 			ID:      "1",
 			Payload: []byte(`{}`),
 		},
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err := s.env.GetWorkflowError()
 	s.NoError(err)
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_Success() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.Equal("test", instance.ScheduleID)
 		s.Equal(s.connectorID, instance.ConnectorID)
 		s.False(instance.Terminated)
 		return nil
 	})
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   models.CAPABILITY_FETCH_CONVERSIONS.String(),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{},
 			NewState:    []byte(`{}`),
 			HasMore:     false,
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.StorageStatesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.Equal("test", instance.ScheduleID)
 		s.Equal(s.connectorID, instance.ConnectorID)
 		s.True(instance.Terminated)
 		return nil
 	})
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.NoError(err)
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_HasMoreLoop_Success() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.Equal("test", instance.ScheduleID)
 		s.Equal(s.connectorID, instance.ConnectorID)
 		s.False(instance.Terminated)
 		return nil
 	})
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   models.CAPABILITY_FETCH_CONVERSIONS.String(),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	// First page: HasMore = true
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{},
 			NewState:    []byte(`{"cursor":"page2"}`),
 			HasMore:     true,
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.StorageStatesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	// Second page: HasMore = false
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{},
 			NewState:    []byte(`{}`),
 			HasMore:     false,
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.StorageStatesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.Equal("test", instance.ScheduleID)
 		s.Equal(s.connectorID, instance.ConnectorID)
 		s.True(instance.Terminated)
 		return nil
 	})
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.NoError(err)
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_StorageInstancesStore_Error() {
 	expectedErr := errors.New("error-test")
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(
 		temporal.NewNonRetryableApplicationError("error-test", "STORAGE", expectedErr),
 	)
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.Error(err)
 	s.ErrorContains(err, expectedErr.Error())
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_StorageStatesGet_Error() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	expectedErr := errors.New("error-test")
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		nil,
 		temporal.NewNonRetryableApplicationError("error-test", "STORAGE", expectedErr),
 	)
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.True(instance.Terminated)
 		s.NotNil(instance.Error)
 		return nil
 	})
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.Error(err)
 	s.ErrorContains(err, expectedErr.Error())
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_PluginFetchNextConversions_Error() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   models.CAPABILITY_FETCH_CONVERSIONS.String(),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	expectedErr := errors.New("error-test")
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		nil,
 		temporal.NewNonRetryableApplicationError("error-test", "PLUGIN", expectedErr),
 	)
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.True(instance.Terminated)
 		s.NotNil(instance.Error)
 		return nil
 	})
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.Error(err)
 	s.ErrorContains(err, expectedErr.Error())
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_StorageConversionsUpsert_Error() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   models.CAPABILITY_FETCH_CONVERSIONS.String(),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{
 				{
 					Reference:        "conv-1",
 					CreatedAt:        time.Now().UTC(),
 					SourceAsset:      "USD/2",
 					DestinationAsset: "USDC/6",
 					SourceAmount:     big.NewInt(10000),
 					Status:           models.CONVERSION_STATUS_COMPLETED,
 					Raw:              []byte(`{}`),
 				},
 			},
 			NewState: []byte(`{}`),
 			HasMore:  false,
 		},
 		nil,
 	)
 	expectedErr := errors.New("error-test")
 	s.env.OnActivity(activities.StorageConversionsUpsertActivity, mock.Anything, mock.Anything).Once().Return(
 		temporal.NewNonRetryableApplicationError("error-test", "STORAGE", expectedErr),
 	)
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.True(instance.Terminated)
 		s.NotNil(instance.Error)
 		return nil
 	})
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.Error(err)
 	s.ErrorContains(err, expectedErr.Error())
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_StorageStatesStore_Error() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   models.CAPABILITY_FETCH_CONVERSIONS.String(),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{},
 			NewState:    []byte(`{}`),
 			HasMore:     false,
 		},
 		nil,
 	)
 	expectedErr := errors.New("error-test")
 	s.env.OnActivity(activities.StorageStatesStoreActivity, mock.Anything, mock.Anything).Once().Return(
 		temporal.NewNonRetryableApplicationError("error-test", "STORAGE", expectedErr),
 	)
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(func(ctx context.Context, instance models.Instance) error {
 		s.True(instance.Terminated)
 		s.NotNil(instance.Error)
 		return nil
 	})
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.Error(err)
 	s.ErrorContains(err, expectedErr.Error())
 }
 func (s *UnitTestSuite) Test_FetchNextConversions_StorageInstancesUpdate_Error() {
 	s.env.OnActivity(activities.StorageInstancesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	s.env.OnActivity(activities.StorageStatesGetActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.State{
 			ID: models.StateID{
 				Reference:   models.CAPABILITY_FETCH_CONVERSIONS.String(),
 				ConnectorID: s.connectorID,
 			},
 			ConnectorID: s.connectorID,
 			State:       []byte(`{}`),
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.PluginFetchNextConversionsActivity, mock.Anything, mock.Anything).Once().Return(
 		&models.FetchNextConversionsResponse{
 			Conversions: []models.PSPConversion{},
 			NewState:    []byte(`{}`),
 			HasMore:     false,
 		},
 		nil,
 	)
 	s.env.OnActivity(activities.StorageStatesStoreActivity, mock.Anything, mock.Anything).Once().Return(nil)
 	expectedErr := errors.New("error-test")
 	s.env.OnActivity(activities.StorageInstancesUpdateActivity, mock.Anything, mock.Anything).Once().Return(
 		temporal.NewNonRetryableApplicationError("error-test", "STORAGE", expectedErr),
 	)
 	err := s.env.SetTypedSearchAttributesOnStart(temporal.NewSearchAttributes(temporal.NewSearchAttributeKeyKeyword(SearchAttributeScheduleID).ValueSet("test")))
 	s.NoError(err)
 	s.env.ExecuteWorkflow(RunFetchNextConversions, FetchNextConversions{
 		ConnectorID:  s.connectorID,
 		FromPayload:  nil,
 		Periodically: false,
 	}, []models.ConnectorTaskTree{})
 	s.True(s.env.IsWorkflowCompleted())
 	err = s.env.GetWorkflowError()
 	s.Error(err)
 	s.ErrorContains(err, expectedErr.Error())
 }

383 lines 14 KiB Go Raw Permalink Blame History

383 lines

14 KiB

Go

Raw Permalink Blame History