Handle boolean realtionship between hermes and pd

app/base_schema.py

@@ -1,5 +1,5 @@
 import json
-from typing import TYPE_CHECKING, Any, Optional, Type
+from typing import TYPE_CHECKING, Any, Optional, Type, Union
 
 from fastapi.exceptions import RequestValidationError
 from pydantic import BaseModel, ConfigDict, Field
@@ -150,6 +150,7 @@ async def get_custom_field_vals(cls, obj) -> dict:
  hermes_model.
 
  FK fields are handled differently because we need to get the ID of the related object.
+ bool fields are handled differently because they are stored as strings in Pipedrive.
  """
  from app.models import CustomField
 
@@ -172,6 +173,10 @@ async def get_custom_field_vals(cls, obj) -> dict:
  if val:
  val = json.dumps(val)
 
+ elif cf.field_type == CustomField.TYPE_BOOL:
+ val = getattr(obj, cf.hermes_field_name, None)
+ if isinstance(val, bool):
+ val = 'true' if val else 'false'
  else:
  val = getattr(obj, cf.hermes_field_name, None)
  elif cf.tc2_machine_name:
@@ -219,7 +224,7 @@ async def get_custom_fieldinfo(
  elif field.field_type == CustomField.TYPE_STR:
  field_kwargs.update(annotation=Optional[str], default=None)
  elif field.field_type == CustomField.TYPE_BOOL:
- field_kwargs.update(annotation=Optional[bool], default=None)
+ field_kwargs.update(annotation=Optional[Union[bool, str]], default=None)
  elif field.field_type == CustomField.TYPE_FK_FIELD:
  field_kwargs.update(
  annotation=Optional[int],

app/pipedrive/_schema.py

@@ -94,11 +94,13 @@ async def from_company(cls, company: Company) -> 'Organisation':
  return cls(**final_kwargs)
 
  async def company_dict(self, custom_fields: list[CustomField]) -> dict:
- cf_data_from_hermes = {
- c.hermes_field_name: getattr(self, c.machine_name)
- for c in custom_fields
- if c.hermes_field_name and c.field_type != CustomField.TYPE_FK_FIELD
- }
+ cf_data_from_hermes = {}
+ for c in custom_fields:
+ if c.hermes_field_name and c.field_type != CustomField.TYPE_FK_FIELD:
+ value = getattr(self, c.machine_name)
+ if c.field_type == CustomField.TYPE_BOOL:
+ value = value.strip().lower() == 'true'
+ cf_data_from_hermes[c.hermes_field_name] = value
 
  admins_from_hermes = {}
  if hasattr(self, 'support_person') and self.support_person:

tests/test_combined.py

@@ -125,6 +125,58 @@ async def test_create_company_with_tc_custom_field_empty(self, mock_pd_request,
  await source_field.delete()
  await build_custom_field_schema()
 
+ @mock.patch('app.tc2.api.session.request')
+ @mock.patch('app.pipedrive.api.session.request')
+ async def test_create_company_with_tc_bool_custom_field(self, mock_pd_request, mock_tc2_get):
+ mock_pd_request.side_effect = fake_pd_request(self.pipedrive)
+ mock_tc2_get.side_effect = mock_tc2_request()
+
+ admin = await Admin.create(
+ tc2_admin_id=30,
+ first_name='Brain',
+ last_name='Johnson',
+ username='brian@tc.com',
+ password='foo',
+ pd_owner_id=10,
+ )
+
+ has_signed_up = await CustomField.create(
+ linked_object_type='Company',
+ pd_field_id='123_has_signed_up_456',
+ name='Has Signed Up',
+ field_type=CustomField.TYPE_BOOL,
+ )
+ await build_custom_field_schema()
+
+ assert not await Company.exists()
+ pd_org_data = {
+ 'v': 1,
+ 'matches_filters': {'current': []},
+ 'meta': {'action': 'updated', 'object': 'organization'},
+ 'current': {
+ 'owner_id': 10,
+ 'id': 20,
+ 'name': 'Test company',
+ 'address_country': None,
+ '123_has_signed_up_456': 'false',
+ },
+ 'previous': {},
+ 'event': 'updated.organization',
+ }
+
+ r = await self.client.post(self.pipedrive_callback, json=pd_org_data)
+ assert r.status_code == 200, r.json()
+ company = await Company.get()
+ assert company.name == 'Test company'
+ assert company.sales_person_id == admin.id
+ cf = await CustomFieldValue.get()
+ assert cf.value == 'false'
+
+ assert not company.has_signed_up
+
+ await has_signed_up.delete()
+ await build_custom_field_schema()
+
  @mock.patch('app.tc2.api.session.request')
  @mock.patch('app.pipedrive.api.session.request')
  async def test_tc2_cb_company_exists_in_tc_and_pd_but_not_in_hermes(self, mock_pd_request, mock_tc2_get):
@@ -377,6 +429,73 @@ async def test_tc2_cb_create_company_create_org_update_org(self, mock_pd_request
  }
  }
 
+ @mock.patch('app.tc2.api.session.request')
+ @mock.patch('app.pipedrive.api.session.request')
+ async def test_tc2_cb_create_company_create_org_update_org_bool(self, mock_pd_request, mock_tc2_get):
+ mock_pd_request.side_effect = fake_pd_request(self.pipedrive)
+ mock_tc2_get.side_effect = fake_tc2_request(self.tc2)
+
+ admin = await Admin.create(
+ tc2_admin_id=30,
+ first_name='Brain',
+ last_name='Johnson',
+ username='brian@tc.com',
+ )
+
+ await CustomField.create(
+ linked_object_type='Company',
+ pd_field_id='123_sales_person_456',
+ hermes_field_name='sales_person',
+ name='Sales Person',
+ field_type=CustomField.TYPE_FK_FIELD,
+ )
+
+ await CustomField.create(
+ linked_object_type='Company',
+ pd_field_id='123_bdr_person_456',
+ hermes_field_name='bdr_person',
+ name='BDR Person',
+ field_type=CustomField.TYPE_FK_FIELD,
+ )
+
+ await CustomField.create(
+ linked_object_type='Company',
+ pd_field_id='123_has_signed_up_456',
+ hermes_field_name='has_signed_up',
+ name='Has Signed Up',
+ field_type=CustomField.TYPE_BOOL,
+ )
+
+ await build_custom_field_schema()
+
+ modified_data = client_full_event_data()
+ modified_data['subject']['meta_agency']['name'] = 'MyTutors'
+ modified_data['subject']['bdr_person'] = None
+ events = [modified_data]
+ data = {'_request_time': 123, 'events': events}
+ r = await self.client.post('/tc2/callback/', json=data, headers={'Webhook-Signature': self._tc2_sig(data)})
+ assert r.status_code == 200, r.json()
+
+ assert await Company.exists()
+ company = await Company.get()
+ assert company.name == 'MyTutors'
+ assert company.has_signed_up
+ assert not company.bdr_person
+ assert await company.support_person == await company.sales_person == admin
+
+ assert self.pipedrive.db['organizations'] == {
+ 1: {
+ 'id': 1,
+ 'name': 'MyTutors',
+ 'address_country': 'GB',
+ 'owner_id': None,
+ '123_hermes_id_456': company.id,
+ '123_sales_person_456': admin.id,
+ '123_bdr_person_456': None,
+ '123_has_signed_up_456': 'true',
+ }
+ }
+
  @mock.patch('app.tc2.api.session.request')
  @mock.patch('app.pipedrive.api.session.request')
  async def test_tc2_cb_create_company_cf_json(self, mock_pd_request, mock_tc2_get):