Before either importing the synapse for the first time or making updates to it later on, we strongly recommend to first test it locally. Testing can be done using the pytest framework and mocking the platform requests that each method would receive.
Importing dependancies
In the test file we first recommend to include the following scripts and libraries (bellow examples are taken from the sample file provided with the Pipedrive Synapse):
import pytest import flask import json import main from syncari.models import *
Helper Methods
To ease the configuration of each method we recommend to set up helper methods that return mock request data, that Syncari’s platform would provide as arguments in methods:
- __execute_request() -> executes the synapse.
- __route() -> wrapper on the __execute_request() to add logging and assert that the synapse always outputs a response.
- __get_entity() -> to mock a describe request that provides an entity schema for the entity when a CRUD method is triggered.
- get_connection() -> creates the framework persisting connection object which includes authentication and metadata.
- get_raw_connection() -> mocks test data that the user would input in the UI.
def __execute_request(synapse_request):
with flask.Flask(__name__).test_request_context(data=synapse_request):
return main.execute(flask.request)
def __route(synapse_request):
resp = __execute_request(synapse_request)
print(resp)
print(type(resp))
assert resp is not None
return json.loads(resp
def __get_entity(entity_name):
synapse_request = Request(type=RequestType.DESCRIBE,
connection=get_connection(),
body=DescribeRequest(entities=[entity_name])).json()
synapse_response = __route(synapse_request)
print("response is: " + str(synapse_response))
return json.loads(synapse_response[0])
def get_connection():
synapse_request = Request(type=RequestType.TEST,
connection=get_raw_connection(),
body=None).json()
resp = __execute_request(synapse_request)
print(type(resp))
print(resp)
return Connection.parse_raw(resp)
def get_raw_connection():
authConfig=AuthConfig(accessToken="")
connection = Connection(id='1', name='name', endpoint='https://syncariinc-sandbox.pipedrive.com/v1/',
authConfig=authConfig, idFieldName='idfield', watermarkFieldName='watermarkfield',
createdAtFieldName='createdfield', updatedAtFieldName='updatedfield', oAuthRedirectUrl='http://redirect.com', extraField='extraField')
return connection
Testing Connection and Describe Method
Once we have the helper methods set we can test the mandatory methods which include - synapse_info, test and describe method. When it comes to testing describe methods we recommend writing two tests - one for all entities and one for a single one. This is to ensure the synapse will function for both cases: resyncing and initial setup as well as invocations during the live sync of pipelines. Describe tests are especially recommended when you are using a seeded schema and are making changes to it, since it ensures that future updates doesn’t break the existing synapse.
def test_connect_request():
connection = get_connection()
assert connection is not None
assert connection.authConfig.accessToken is not None
def test_synapse_info_request():
synapse_request = Request(type=RequestType.SYNAPSE_INFO,
connection=get_connection(),
body=None).json()
synapse_response = __route(synapse_request)
assert synapse_response is not None
assert synapse_response['name'] == 'pipedrive'
def test_describe_request():
entity = __get_entity('person')
assert entity['apiName'] == 'person'
attribute_names = [attribute['apiName'] for attribute in entity['attributes']]
assert 'id' in attribute_names
def test_describe_all_request():
synapse_request = Request(type=RequestType.DESCRIBE,
connection=get_connection(),
body=DescribeRequest(entities=[])).json()
entities = __route(synapse_request)
# currently 12 entities are supported
assert len(entities) == 12
for ent in entities:
entity = json.loads(ent)
assert entity['apiName'] is not None
attribute_names = [attribute['apiName'] for attribute in entity['attributes']]
Testing Read Method
Once we have the required methods working we can move on to CRUD methods. Here we recommend to have at least one test that covers all entities on the read method (if used) - this could be on either incremental sync or a historical resync example:
def test_read_request(): objects = ['stage','pipeline','role','activity','activityType','person','user','organization','deal','product'] #,'lead','leadLabel'] for object in objects: synapse_request = Request(type=RequestType.READ, connection=get_connection(), body=SyncRequest( entity=__get_entity(object), watermark=Watermark(start=0, end=0, isResync=True))).json() synapse_response = __route(synapse_request) assert synapse_response['watermark'] is not None assert len(synapse_response['data']) = 1 for resp_dict in synapse_response['data']: assert resp_dict['name'] == object assert resp_dict['id'] is not None assert resp_dict['values']['id'] is not None assert resp_dict['lastModified'] 0 assert resp_dict['createdAt'] 0
Testing CRUD Methods
For the rest of the CRUD methods you can implement them one of the two ways - you can have separate methods as we have a separate Ge By ID method that tries to fetch a single record or have them grouped together, like Create, Update, Delete.
def test_crud_request():
entity = __get_entity('person')
watermark = Watermark(start=0, end=0)
data = [Record(name='person', id='', syncariEntityId='',deleted=False,
values={'name':'Test Lead1','email':'testlead1@test.com'}, lastModified=0,createdAt=0),
Record(name='person', id='', syncariEntityId='',deleted=False,
values={'name':'Test Lead2','email':'testlead2@test.com'}, lastModified=0,createdAt=0)]
try:
synapse_request = Request(type=RequestType.CREATE, connection=get_connection(),
body=SyncRequest(entity=entity,watermark=watermark,data=data)).json()
synapse_response = __route(synapse_request)
assert len(synapse_response) == 2
cnt = 0
for resp in synapse_response:
resp_dict = json.loads(resp)
assert resp_dict['id'] is not None
assert resp_dict['success'] is True
data[cnt].id = resp_dict['id']
data[cnt].values['first_name'] = 'Test First Name'
cnt += 1
synapse_request = Request(type=RequestType.UPDATE, connection=get_connection(),
body=SyncRequest(entity=entity,watermark=watermark,data=data)).json()
synapse_response = __route(synapse_request)
assert len(synapse_response) == 2
for resp in synapse_response:
resp_dict = json.loads(resp)
assert resp_dict['id'] is not None
assert resp_dict['success'] is True
synapse_request = Request(type=RequestType.GET_BY_ID, connection=get_connection(),
body=SyncRequest(entity=entity,watermark=watermark,data=data)).json()
synapse_response = __route(synapse_request)
assert len(synapse_response) == 2
for resp in synapse_response:
resp_dict = json.loads(resp)
assert resp_dict['id'] is not None
assert resp_dict['values']['id'] is not None
assert resp_dict['values']['first_name'] == 'Test First Name'
finally:
synapse_request = Request(type=RequestType.DELETE, connection=get_connection(),
body=SyncRequest(entity=entity,watermark=watermark,data=data)).json()
synapse_response = __route(synapse_request)
assert len(synapse_response) == 2
Testing Webhook Methods
For webhook methods the extract_webhook_identifier_request tests the logic for returning the webhookIdentifier value, while the test_process_webook_request tests processing of a single webhook payload. For both cases we recommend example payload requests.
def test_extract_webhook_identifier_request(): synapse_request = Request(type=RequestType.EXTRACT_WEBHOOK_IDENTIFIER, connection=get_connection(), body=WebhookRequest(body=json.dumps(test_webhook_request_body))).json() synapse_response = __route(synapse_request) assert synapse_response == 2881333 def test_process_webook_request(): synapse_request = Request(type=RequestType.PROCESS_WEBHOOK, connection=get_connection(), body=WebhookRequest(body=json.dumps(test_webhook_request_body))).json() synapse_response = __route(synapse_request) for resp in synapse_response: resp_dict = json.loads(resp) assert resp_dict is not None assert resp_dict['id'] == '1' assert resp_dict['name'] == 'person' assert resp_dict['values']['first_name'] == 'Syncari' assert resp_dict['values']['owner_id'] == 14434274