diff --git a/federation/service/processer_data_test.go b/federation/service/processer_data_test.go new file mode 100644 index 000000000..ddac8583a --- /dev/null +++ b/federation/service/processer_data_test.go @@ -0,0 +1,178 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "testing" + + cs "github.com/cortezaproject/corteza-server/compose/service" + ct "github.com/cortezaproject/corteza-server/compose/types" + "github.com/cortezaproject/corteza-server/federation/types" + ss "github.com/cortezaproject/corteza-server/system/service" + st "github.com/cortezaproject/corteza-server/system/types" + "github.com/stretchr/testify/require" +) + +type ( + testSyncService struct { + Sync + } + + testSharedModuleService struct { + SharedModuleService + } + testRecordServicePersistSuccess struct { + cs.RecordService + } + testRecordServicePersistError struct { + cs.RecordService + } + testUserService struct { + ss.UserService + } + testRoleService struct { + ss.RoleService + } +) + +func TestProcesserData_persist(t *testing.T) { + + var ( + tcc = []struct { + name string + payload string + mappings string + persisted int + err string + values *ct.RecordValueSet + s *Sync + }{ + { + "successful persist on valid mapping", + `{"response": {"set": [{"recordID":"1","values":[{"name":"Facebook","value":"foobar"}]}]}}`, + `[{"origin":{"kind":"String","name":"Description","label":"Description","isMulti":false},"destination":{"kind":"String","name":"Name","label":"Description","isMulti":false}},{"origin":{"kind":"Url","name":"Facebook","label":"Facebook","isMulti":false},"destination":{"kind":"Url","name":"Fb","label":"Facebook","isMulti":false}}]`, + 1, + "", + &ct.RecordValueSet{&ct.RecordValue{Name: "Fb", Value: ""}}, + NewSync( + &Syncer{}, + &Mapper{}, + &testSharedModuleService{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "persist error on valid mapping", + `{"response": {"set": [{"recordID":"1","values":[{"name":"Facebook","value":"foobar"}]}]}}`, + `[{"origin":{"kind":"String","name":"Description","label":"Description","isMulti":false},"destination":{"kind":"String","name":"Name","label":"Description","isMulti":false}},{"origin":{"kind":"Url","name":"Facebook","label":"Facebook","isMulti":false},"destination":{"kind":"Url","name":"Fb","label":"Facebook","isMulti":false}}]`, + 0, + "", + &ct.RecordValueSet{&ct.RecordValue{Name: "Fb", Value: ""}}, + NewSync( + &Syncer{}, + &Mapper{}, + &testSharedModuleService{}, + &testRecordServicePersistError{}, + &testUserService{}, + &testRoleService{}), + }, + { + "persist one record on mixed mappings", + `{"response": {"set": [{"recordID":"1","values":[{"name":"THIS_FIELD_NAME_IS_NOT_ON_ORIGIN","value":"this value will not be set"}]}, {"recordID":"1","values":[{"name":"Facebook","value":"this value WILL be set"}]}]}}`, + `[{"origin":{"kind":"String","name":"Description","label":"Description","isMulti":false},"destination":{"kind":"String","name":"Name","label":"Description","isMulti":false}},{"origin":{"kind":"Url","name":"Facebook","label":"Facebook","isMulti":false},"destination":{"kind":"Url","name":"Fb","label":"Facebook","isMulti":false}}]`, + 2, + "", + &ct.RecordValueSet{&ct.RecordValue{Name: "Fb", Value: ""}}, + NewSync( + &Syncer{}, + &Mapper{}, + &testSharedModuleService{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "no records from payload", + `{"response": {"set": []}}`, + `[{"origin":{"kind":"String","name":"Description","label":"Description","isMulti":false},"destination":{"kind":"String","name":"Name","label":"Description","isMulti":false}},{"origin":{"kind":"Url","name":"Facebook","label":"Facebook","isMulti":false},"destination":{"kind":"Url","name":"Fb","label":"Facebook","isMulti":false}}]`, + 0, + "", + &ct.RecordValueSet{&ct.RecordValue{Name: "Fb", Value: ""}}, + NewSync( + &Syncer{}, + &Mapper{}, + &testSharedModuleService{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "validation error, no persist", + `{"respon`, + `[{"origin":{"kind":"String","name":"Description","label":"Description","isMulti":false},"destination":{"kind":"String","name":"Name","label":"Description","isMulti":false}},{"origin":{"kind":"Url","name":"Facebook","label":"Facebook","isMulti":false},"destination":{"kind":"Url","name":"Fb","label":"Facebook","isMulti":false}}]`, + 0, + "unexpected end of JSON input", + &ct.RecordValueSet{&ct.RecordValue{Name: "Fb", Value: ""}}, + NewSync( + &Syncer{}, + &Mapper{}, + &testSharedModuleService{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + } + ) + + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + var ( + ctx = context.Background() + req = require.New(t) + ) + + mm := &types.ModuleFieldMappingSet{} + json.Unmarshal([]byte(tc.mappings), mm) + + dp := &dataProcesser{ + ID: 1, + ComposeModuleID: 1, + ComposeNamespaceID: 1, + NodeBaseURL: "", + ModuleMappings: mm, + ModuleMappingValues: tc.values, + SyncService: tc.s, + Node: &types.Node{}, + User: &st.User{}, + } + + out, err := dp.Process(ctx, []byte(tc.payload)) + + if tc.err != "" { + req.Equal(tc.err, err.Error()) + } else { + req.NoError(err) + } + + req.Equal(tc.persisted, out.(dataProcesserResponse).Processed) + }) + } +} + +func (s testRecordServicePersistSuccess) Create(record *ct.Record) (*ct.Record, error) { + return nil, nil +} + +func (s testRecordServicePersistSuccess) With(_ context.Context) cs.RecordService { + return &testRecordServicePersistSuccess{} +} + +func (s testRecordServicePersistError) Create(record *ct.Record) (*ct.Record, error) { + return nil, errors.New("mocked error") +} + +func (s testRecordServicePersistError) With(_ context.Context) cs.RecordService { + return &testRecordServicePersistError{} +} diff --git a/federation/service/processer_structure.go b/federation/service/processer_structure.go index 04f1a8246..456361ccd 100644 --- a/federation/service/processer_structure.go +++ b/federation/service/processer_structure.go @@ -57,7 +57,7 @@ func (dp *structureProcesser) Process(ctx context.Context, payload []byte) (Proc if err != nil { return structureProcesserResponse{ - ModuleID: existing.ExternalFederationModuleID, + ModuleID: new.ExternalFederationModuleID, Processed: processed, }, err } diff --git a/federation/service/processer_structure_test.go b/federation/service/processer_structure_test.go new file mode 100644 index 000000000..bbc111ef9 --- /dev/null +++ b/federation/service/processer_structure_test.go @@ -0,0 +1,232 @@ +package service + +import ( + "context" + "errors" + "testing" + + cs "github.com/cortezaproject/corteza-server/compose/service" + "github.com/cortezaproject/corteza-server/federation/types" + ss "github.com/cortezaproject/corteza-server/system/service" + st "github.com/cortezaproject/corteza-server/system/types" + "github.com/stretchr/testify/require" +) + +type ( + testSyncStructureService struct{ Sync } + testSyncStructureServiceCanUpdateFieldsError struct{ Sync } + + testStructureSharedModuleService struct{ SharedModuleService } + testStructureSharedModuleServiceError struct{ SharedModuleService } + testStructureSharedModuleServiceCreateNewModule struct{ SharedModuleService } + testStructureSharedModuleServiceCreateNewModuleErr struct{ SharedModuleService } + testStructureSharedModuleServiceUpdateModule struct{ SharedModuleService } + testStructureSharedModuleServiceUpdateModuleErr struct{ SharedModuleService } + + testStructureRecordServicePersistSuccess struct{ cs.RecordService } + testStructureUserService struct{ ss.UserService } + testStructureRoleService struct{ ss.RoleService } +) + +func TestProcesserStructure_persist(t *testing.T) { + + var ( + tcc = []struct { + name string + payload string + persisted int + err string + s *Sync + }{ + { + "no modules from payload", + `{"response": {"set": []}}`, + 0, + "", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleService{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "validation error, no persist", + `{"respon`, + 0, + "unexpected end of JSON input", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleService{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "lookup existing module error", + `{"response":{"set": [{"moduleID": "11", "nodeID": "21", "composeModuleID": "31", "composeNamespaceID":"32","handle":"handle","name":"name","fields":[]}]}}`, + 0, + "db error", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleServiceError{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "create new module", + `{"response":{"set": [{"moduleID": "11", "nodeID": "21", "composeModuleID": "31", "composeNamespaceID":"32","handle":"handle","name":"name","fields":[]}]}}`, + 1, + "", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleServiceCreateNewModule{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "create new module error", + `{"response":{"set": [{"moduleID": "11", "nodeID": "21", "composeModuleID": "31", "composeNamespaceID":"32","handle":"handle","name":"name","fields":[]}]}}`, + 0, + "could not create new module", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleServiceCreateNewModuleErr{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "could not update module fields error", + `{"response":{"set": [{"moduleID": "11", "nodeID": "21", "composeModuleID": "31", "composeNamespaceID":"32","handle":"handle","name":"name","fields":[]}]}}`, + 0, + "module structure changed", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleServiceUpdateModule{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "update module fields success", + `{"response":{"set": [{"moduleID": "11", "nodeID": "21", "composeModuleID": "31", "composeNamespaceID":"32","handle":"handle","name":"name","fields":[{"kind":"String","label":"label","isMulti":false,"name":"existingField"}]}]}}`, + 1, + "", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleServiceUpdateModule{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + { + "update module error", + `{"response":{"set": [{"moduleID": "11", "nodeID": "21", "composeModuleID": "31", "composeNamespaceID":"32","handle":"handle","name":"name","fields":[]}]}}`, + 0, + "could not update module", + NewSync( + &Syncer{}, + &Mapper{}, + &testStructureSharedModuleServiceUpdateModuleErr{}, + &testRecordServicePersistSuccess{}, + &testUserService{}, + &testRoleService{}), + }, + } + ) + + for _, tc := range tcc { + t.Run(tc.name, func(t *testing.T) { + var ( + ctx = context.Background() + req = require.New(t) + ) + + dp := &structureProcesser{ + SyncService: tc.s, + Node: &types.Node{}, + User: &st.User{}, + } + + out, err := dp.Process(ctx, []byte(tc.payload)) + + if tc.err != "" { + if err != nil { + req.Equal(tc.err, err.Error()) + } else { + req.Fail("err should not be nil") + } + } else { + req.NoError(err) + } + + req.Equal(tc.persisted, out.(structureProcesserResponse).Processed) + }) + } +} + +func (s testStructureSharedModuleServiceError) Find(ctx context.Context, filter types.SharedModuleFilter) (types.SharedModuleSet, types.SharedModuleFilter, error) { + return types.SharedModuleSet{}, types.SharedModuleFilter{}, errors.New("db error") +} + +func (s testStructureSharedModuleServiceCreateNewModule) Find(ctx context.Context, filter types.SharedModuleFilter) (types.SharedModuleSet, types.SharedModuleFilter, error) { + return types.SharedModuleSet{}, types.SharedModuleFilter{}, nil +} + +func (s testStructureSharedModuleServiceCreateNewModule) Create(ctx context.Context, new *types.SharedModule) (*types.SharedModule, error) { + return &types.SharedModule{}, nil +} + +func (s testStructureSharedModuleServiceCreateNewModuleErr) Find(ctx context.Context, filter types.SharedModuleFilter) (types.SharedModuleSet, types.SharedModuleFilter, error) { + return types.SharedModuleSet{}, types.SharedModuleFilter{}, nil +} + +func (s testStructureSharedModuleServiceCreateNewModuleErr) Create(ctx context.Context, new *types.SharedModule) (*types.SharedModule, error) { + return nil, errors.New("could not create new module") +} + +func (s testStructureSharedModuleServiceUpdateModule) Find(ctx context.Context, filter types.SharedModuleFilter) (types.SharedModuleSet, types.SharedModuleFilter, error) { + return types.SharedModuleSet{ + &types.SharedModule{ + ID: 11, + NodeID: 21, + Handle: "handle", + Name: "name", + Fields: types.ModuleFieldSet{ + &types.ModuleField{ + Kind: "String", + Label: "label", + IsMulti: false, + Name: "existingField", + }, + }, + }}, + types.SharedModuleFilter{}, + nil +} + +func (s testStructureSharedModuleServiceUpdateModule) Update(ctx context.Context, updated *types.SharedModule) (*types.SharedModule, error) { + return &types.SharedModule{}, nil +} + +func (s testStructureSharedModuleServiceUpdateModuleErr) Find(ctx context.Context, filter types.SharedModuleFilter) (types.SharedModuleSet, types.SharedModuleFilter, error) { + return types.SharedModuleSet{&types.SharedModule{ + Fields: types.ModuleFieldSet{}, + }}, + types.SharedModuleFilter{}, + nil +} + +func (s testStructureSharedModuleServiceUpdateModuleErr) Update(ctx context.Context, updated *types.SharedModule) (*types.SharedModule, error) { + return nil, errors.New("could not update module") +} diff --git a/federation/service/shared_module.go b/federation/service/shared_module.go index 9dade14b3..9abe0a345 100644 --- a/federation/service/shared_module.go +++ b/federation/service/shared_module.go @@ -128,45 +128,6 @@ func (svc sharedModule) uniqueCheck(ctx context.Context, m *types.SharedModule) return nil } -// func (svc sharedModule) DeleteByID(ctx context.Context, nodeID, moduleID uint64) error { -// return trim1st(svc.updater(ctx, nodeID, moduleID, ModuleActionDelete, svc.handleDelete)) -// } - -// func (svc sharedModule) updater(ctx context.Context, nodeID, moduleID uint64, action func(...*moduleActionProps) *moduleAction, fn moduleUpdateHandler) (*types.SharedModule, error) { -// var ( -// moduleChanged, fieldsChanged bool - -// n *types.Node -// m *types.SharedModule -// // m, old *types.SharedModule -// aProps = &moduleActionProps{module: &types.SharedModule{ID: moduleID, NodeID: nodeID}} -// err error -// ) - -// err = store.Tx(ctx, svc.store, func(ctx context.Context, s store.Storer) (err error) { -// if m, err = svc.store.LookupFederationSharedModuleByID(ctx, moduleID); err != nil { -// return err -// } - -// // TODO - handle node id also -// if moduleChanged, fieldsChanged, err = fn(ctx, n, m); err != nil { -// return err -// } - -// return err -// }) - -// return m, svc.recordAction(ctx, aProps, action, err) -// } - -// func (svc sharedModule) handleDelete(ctx context.Context, n *types.Node, m *types.SharedModule) (bool, bool, error) { -// if err := store.DeleteFederationSharedModuleByID(ctx, svc.store, m.ID); err != nil { -// return false, false, err -// } - -// return false, false, nil -// } - func (svc sharedModule) Find(ctx context.Context, filter types.SharedModuleFilter) (set types.SharedModuleSet, f types.SharedModuleFilter, err error) { var ( aProps = &sharedModuleActionProps{filter: &filter} diff --git a/federation/service/sync.go b/federation/service/sync.go index 30613887b..4840552a1 100644 --- a/federation/service/sync.go +++ b/federation/service/sync.go @@ -11,7 +11,6 @@ import ( ct "github.com/cortezaproject/corteza-server/compose/types" "github.com/cortezaproject/corteza-server/federation/types" "github.com/cortezaproject/corteza-server/pkg/filter" - "github.com/cortezaproject/corteza-server/system/service" ss "github.com/cortezaproject/corteza-server/system/service" st "github.com/cortezaproject/corteza-server/system/types" ) @@ -45,18 +44,20 @@ func NewSync(s *Syncer, m *Mapper, sm SharedModuleService, cs cs.RecordService, // is possible func (s *Sync) CanUpdateSharedModule(ctx context.Context, new *types.SharedModule, existing *types.SharedModule) (bool, error) { // check for mapped fields - fstr, err := json.Marshal(new.Fields) - f2str, err := json.Marshal(existing.Fields) + var ( + fstr, f2str []byte + err error + ) - if err != nil { + if fstr, err = json.Marshal(new.Fields); err != nil { return false, err } - if string(fstr) == string(f2str) { - return true, nil + if f2str, err = json.Marshal(existing.Fields); err != nil { + return false, err } - return false, nil + return string(fstr) == string(f2str), nil } // ProcessPayload passes the payload to the syncer lib @@ -83,7 +84,7 @@ func (s *Sync) CreateRecord(ctx context.Context, rec *ct.Record) (*ct.Record, er func (s *Sync) LookupSharedModule(ctx context.Context, new *types.SharedModule) (*types.SharedModule, error) { var sm *types.SharedModule - list, _, err := service.DefaultStore.SearchFederationSharedModules(ctx, types.SharedModuleFilter{ + list, _, err := s.sharedModuleService.Find(ctx, types.SharedModuleFilter{ NodeID: new.NodeID, ExternalFederationModuleID: new.ExternalFederationModuleID}) diff --git a/federation/service/sync_worker_data.go b/federation/service/sync_worker_data.go index 3c21ecc52..ffaf60229 100644 --- a/federation/service/sync_worker_data.go +++ b/federation/service/sync_worker_data.go @@ -245,7 +245,7 @@ func (w *syncWorkerData) Watch(ctx context.Context, delay time.Duration, limit i NodeID: meta.Node.ID, ModuleID: meta.ID, SyncStatus: syncStatus, - SyncType: types.NodeSyncTypeStructure, + SyncType: types.NodeSyncTypeData, TimeOfAction: time.Now().UTC(), } diff --git a/federation/service/syncer.go b/federation/service/syncer.go index 65fde578f..e68064d09 100644 --- a/federation/service/syncer.go +++ b/federation/service/syncer.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "time" @@ -12,7 +13,9 @@ import ( ) type ( - Syncer struct{} + Syncer struct { + client http.Client + } Url struct { Url types.SyncerURI @@ -44,10 +47,6 @@ func (h *Syncer) Queue(url Url, out chan Url) { } func (h *Syncer) Fetch(ctx context.Context, url string) (io.Reader, error) { - client := http.Client{ - Timeout: time.Duration(3) * time.Second, - } - req, err := http.NewRequest("GET", url, nil) if err != nil { return nil, err @@ -57,13 +56,13 @@ func (h *Syncer) Fetch(ctx context.Context, url string) (io.Reader, error) { req.Header.Add("Authorization", `Bearer `+authToken.(string)) } - resp, err := client.Do(req) + resp, err := h.client.Do(req) if err != nil { return nil, err } if resp.StatusCode != 200 { - return nil, errors.New("404") + return nil, errors.New(fmt.Sprintf("invalid return status: %d", resp.StatusCode)) } return resp.Body, nil @@ -92,3 +91,11 @@ func (h *Syncer) ParseHeader(ctx context.Context, payload []byte) (aux AuxRespon err = json.Unmarshal(payload, &aux) return } + +func NewSyncer() *Syncer { + return &Syncer{ + client: http.Client{ + Timeout: time.Duration(3) * time.Second, + }, + } +} diff --git a/federation/service/syncer_test.go b/federation/service/syncer_test.go index d65860e19..467e1278b 100644 --- a/federation/service/syncer_test.go +++ b/federation/service/syncer_test.go @@ -1,7 +1,10 @@ package service import ( + "bytes" "context" + "io/ioutil" + "net/http" "testing" "github.com/cortezaproject/corteza-server/federation/types" @@ -9,10 +12,12 @@ import ( ) type ( + RoundTripFunc func(req *http.Request) *http.Response + testProcesser struct{} ) -func (p *testProcesser) Process(ctx context.Context, payload []byte) (int, error) { +func (p *testProcesser) Process(ctx context.Context, payload []byte) (ProcesserResponse, error) { return 0, nil } @@ -43,7 +48,7 @@ func TestSyncer_process(t *testing.T) { func TestSyncer_parseHeader(t *testing.T) { var ( req = require.New(t) - syncer = &Syncer{} + syncer = NewSyncer() ctx = context.Background() ) @@ -60,3 +65,97 @@ func TestSyncer_parseHeader(t *testing.T) { req.EqualError(err, "unexpected end of JSON input") req.Equal(n, 0) } + +func TestSyncer_response(t *testing.T) { + var ( + req = require.New(t) + ) + + tests := []struct { + name string + url string + expect string + syncer *Syncer + ctx context.Context + }{ + { + name: "non 200 request", + url: "http://example.ltd", + syncer: &Syncer{ + client: *NewHttpClient(func(req *http.Request) *http.Response { + return &http.Response{ + StatusCode: 500, + Body: ioutil.NopCloser(bytes.NewBufferString(`{"error":{"message":"common error response"}}`)), + Header: make(http.Header), + } + }), + }, + ctx: context.Background(), + expect: "invalid return status: 500", + }, + { + name: "invalid hostname", + url: "http://examp le.ltd", + syncer: &Syncer{ + client: *NewHttpClient(func(req *http.Request) *http.Response { + return &http.Response{ + StatusCode: 200, + Body: ioutil.NopCloser(bytes.NewBufferString(`{"error":{"message":"common error response"}}`)), + Header: make(http.Header), + } + }), + }, + ctx: context.Background(), + expect: "parse \"http://examp le.ltd\": invalid character \" \" in host name", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := tt.syncer.Fetch(tt.ctx, tt.url) + req.EqualError(err, tt.expect) + }) + } + +} + +func TestSyncer_authTokenCorrectlySet(t *testing.T) { + var ( + req = require.New(t) + expectedAuthToken = "TEST_JWT_TOKEN" + actualAuthToken = "" + ) + + ctx := context.WithValue( + context.Background(), + FederationUserToken, + expectedAuthToken) + + syncer := &Syncer{ + client: *NewHttpClient(func(r *http.Request) *http.Response { + + actualAuthToken = r.Header.Get("Authorization") + + return &http.Response{ + StatusCode: 200, + Body: ioutil.NopCloser(bytes.NewBufferString("OK")), + Header: make(http.Header), + } + }), + } + + _, err := syncer.Fetch(ctx, "http://example.ltd") + + req.Equal("Bearer "+expectedAuthToken, actualAuthToken) + req.NoError(err) +} + +func (f RoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req), nil +} + +func NewHttpClient(rt RoundTripFunc) *http.Client { + return &http.Client{ + Transport: RoundTripFunc(rt), + } +}