Added tests, fixed shared module lookup, data sync status fixed

2020-11-25 13:48:07 +01:00
parent 5566621720
commit ab3da2906b
8 changed files with 536 additions and 58 deletions
@@ -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{}
+}
@@ -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
 		}
@@ -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")
+}
@@ -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}
@@ -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})

@@ -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(),
 			}

@@ -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,
+		},
+	}
+}
@@ -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),
+	}
+}