fix: allocation payload order (#899)

* fix: allocation payload order

* increase cov

* increase cov

* increase cov

Author: ruuushhh

apps/sage_intacct/utils.py

@@ -15,9 +15,9 @@
 from sageintacctsdk import SageIntacctSDK
 from sageintacctsdk.exceptions import WrongParamsError
 
-from apps.sage_intacct.errors.helpers import retry
 from apps.fyle.models import DependentFieldSetting
 from apps.mappings.models import GeneralMapping, LocationEntityMapping
+from apps.sage_intacct.errors.helpers import retry
 from apps.sage_intacct.exports.helpers import get_source_entity_id
 from apps.sage_intacct.models import (
     APPayment,
@@ -1489,6 +1489,7 @@ def __construct_bill(self, bill: Bill, bill_lineitems: list[BillLineitem]) -> di
                 'TRX_AMOUNT': lineitem.amount - lineitem.tax_amount if (lineitem.tax_code and lineitem.tax_amount) else tax_exclusive_amount,
                 'TOTALTRXAMOUNT': lineitem.amount,
                 'ENTRYDESCRIPTION': lineitem.memo,
+                'ALLOCATION': lineitem.allocation_id,
                 'LOCATIONID': lineitem.location_id,
                 'DEPARTMENTID': lineitem.department_id,
                 'PROJECTID': lineitem.project_id,
@@ -1498,7 +1499,6 @@ def __construct_bill(self, bill: Bill, bill_lineitems: list[BillLineitem]) -> di
                 'COSTTYPEID': lineitem.cost_type_id,
                 'CLASSID': lineitem.class_id,
                 'BILLABLE': lineitem.billable,
-                'ALLOCATION': lineitem.allocation_id,
                 'TAXENTRIES': {
                     'TAXENTRY': {
                         'DETAILID': lineitem.tax_code if (lineitem.tax_code and lineitem.tax_amount) else general_mappings.default_tax_code_id
@@ -1759,6 +1759,7 @@ def __construct_base_line_item(self, lineitem: JournalEntryLineitem, dimensions_
         return {
             'currency': journal_entry.currency,
             'description': lineitem.memo,
+            'allocation': lineitem.allocation_id,
             'department': dimensions_values['department_id'],
             'location': dimensions_values['location_id'],
             'projectid': dimensions_values['project_id'],
@@ -1769,7 +1770,6 @@ def __construct_base_line_item(self, lineitem: JournalEntryLineitem, dimensions_
             'itemid': dimensions_values['item_id'],
             'taskid': dimensions_values['task_id'],
             'costtypeid': dimensions_values['cost_type_id'],
-            'allocation': lineitem.allocation_id,
             'customfields': {
                 'customfield': [{
                     'customfieldname': 'FYLE_EXPENSE_URL',

fyle_integrations_imports

@@ -2270,3 +2270,233 @@ def test_sync_methods_persist_count(mocker, db, create_dependent_field_setting):
     sage_intacct_connection.sync_payment_accounts()
     count_record.refresh_from_db()
     assert count_record.payment_accounts_count == 7
+
+
+@pytest.mark.django_db
+def test_is_duplicate_deletion_skipped(db):
+    """
+    Test is_duplicate_deletion_skipped method
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    for attribute_type in ['ACCOUNT', 'VENDOR', 'ITEM', 'CUSTOMER', 'DEPARTMENT', 'CLASS', 'EXPENSE_TYPE', 'PROJECT', 'LOCATION']:
+        result = sage_intacct_connection.is_duplicate_deletion_skipped(attribute_type)
+        assert result is False, f"Expected False for {attribute_type}"
+    for attribute_type in ['CUSTOM_FIELD', 'ALLOCATION', 'TAX_DETAIL']:
+        result = sage_intacct_connection.is_duplicate_deletion_skipped(attribute_type)
+        assert result is True, f"Expected True for {attribute_type}"
+
+
+@pytest.mark.django_db
+def test_is_import_enabled_no_configuration(db):
+    """
+    Test is_import_enabled method when configuration doesn't exist
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    Configuration.objects.filter(workspace_id=workspace_id).delete()
+    result = sage_intacct_connection.is_import_enabled('ACCOUNT')
+    assert result is False
+
+
+@pytest.mark.django_db
+def test_is_import_enabled_with_categories(db):
+    """
+    Test is_import_enabled method with import_categories enabled
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    Configuration.objects.filter(workspace_id=workspace_id).update(import_categories=True)
+    result = sage_intacct_connection.is_import_enabled('ACCOUNT')
+    assert result is True
+    result = sage_intacct_connection.is_import_enabled('EXPENSE_TYPE')
+    assert result is True
+
+
+@pytest.mark.django_db
+def test_is_import_enabled_with_vendors_as_merchants(db):
+    """
+    Test is_import_enabled method with import_vendors_as_merchants enabled
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    Configuration.objects.filter(workspace_id=workspace_id).update(import_vendors_as_merchants=True)
+    result = sage_intacct_connection.is_import_enabled('VENDOR')
+    assert result is True
+
+
+@pytest.mark.django_db
+def test_is_import_enabled_with_mapping_settings(db):
+    """
+    Test is_import_enabled method with mapping settings (PROJECT has import_to_fyle=True in fixtures)
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    result = sage_intacct_connection.is_import_enabled('PROJECT')
+    assert result is True
+    result = sage_intacct_connection.is_import_enabled('DEPARTMENT')
+    assert result is False
+
+
+@pytest.mark.django_db
+def test_get_attribute_disable_callback_path(db):
+    """
+    Test get_attribute_disable_callback_path method
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    result = sage_intacct_connection.get_attribute_disable_callback_path('ACCOUNT')
+    assert result == 'fyle_integrations_imports.modules.categories.disable_categories'
+    result = sage_intacct_connection.get_attribute_disable_callback_path('VENDOR')
+    assert result == 'fyle_integrations_imports.modules.merchants.disable_merchants'
+    result = sage_intacct_connection.get_attribute_disable_callback_path('PROJECT')
+    assert result == 'fyle_integrations_imports.modules.projects.disable_projects'
+
+
+@pytest.mark.django_db
+def test_post_attachments_empty_list(db):
+    """
+    Test post_attachments with empty attachments list
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    result = sage_intacct_connection.post_attachments([], 'test_supdoc', 1)
+    assert result is False
+
+
+@pytest.mark.django_db
+def test_post_attachments_update_exception(mocker, db):
+    """
+    Test post_attachments when attachment update raises exception
+    """
+    workspace_id = 1
+    mocker.patch(
+        'sageintacctsdk.apis.Attachments.post',
+        return_value={'status': 'success', 'key': '3032'}
+    )
+    mocker.patch(
+        'sageintacctsdk.apis.Attachments.update',
+        side_effect=Exception('Update failed')
+    )
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    attachment = [{'download_url': 'url', 'name': 'file.pdf', 'id': 'att1'}]
+    sage_intacct_connection.post_attachments(attachment, 'supdoc1', 1)
+    result = sage_intacct_connection.post_attachments(attachment, 'supdoc1', 2)
+    assert result is False
+
+
+@pytest.mark.django_db
+def test_get_location_id_for_journal_entry(db):
+    """
+    Test __get_location_id_for_journal_entry method
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    GeneralMapping.objects.filter(workspace_id=workspace_id).update(default_location_id=None)
+    result = sage_intacct_connection._SageIntacctConnector__get_location_id_for_journal_entry(workspace_id)
+    location_entity = LocationEntityMapping.objects.filter(workspace_id=workspace_id).exclude(location_entity_name='Top Level').first()
+    if location_entity:
+        assert result == location_entity.destination_id
+    else:
+        assert result is None
+
+
+@pytest.mark.django_db
+def test_get_location_id_for_journal_entry_no_mappings(db):
+    """
+    Test __get_location_id_for_journal_entry when no mappings exist
+    """
+    workspace_id = 1
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    GeneralMapping.objects.filter(workspace_id=workspace_id).update(default_location_id=None)
+    LocationEntityMapping.objects.filter(workspace_id=workspace_id).exclude(location_entity_name='Top Level').delete()
+    result = sage_intacct_connection._SageIntacctConnector__get_location_id_for_journal_entry(workspace_id)
+    assert result is None
+
+
+@pytest.mark.django_db
+def test_sync_cost_codes_with_last_synced_at(db, mocker, create_dependent_field_setting):
+    """
+    Test sync_cost_codes when dependent_field_setting.last_synced_at is set
+    """
+    from apps.fyle.models import DependentFieldSetting
+    workspace_id = 1
+    dependent_field_setting = DependentFieldSetting.objects.get(workspace_id=workspace_id)
+    dependent_field_setting.last_synced_at = datetime.now()
+    dependent_field_setting.save()
+    mocker.patch('sageintacctsdk.apis.Tasks.count', return_value=1)
+    cost_code_data = [[
+        {'RECORDNO': '40', 'TASKID': '112', 'NAME': 'TestCostCode', 'PROJECTID': '1173', 'PROJECTNAME': 'Test Project'}
+    ]]
+    mocker.patch('sageintacctsdk.apis.Tasks.get_all_generator', return_value=cost_code_data)
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    sage_intacct_connection.sync_cost_codes()
+    assert CostCode.objects.filter(workspace_id=workspace_id, task_id='112').exists()
+
+
+@pytest.mark.django_db
+def test_sync_allocations_empty_entries(mocker, db):
+    """
+    Test sync_allocations when allocation_entries is empty
+    """
+    workspace_id = 1
+
+    def mock_allocations_generator(field=None, value=None, updated_at=None):
+        yield [{'ALLOCATIONID': 'EMPTY_ALLOC', 'STATUS': 'active'}]
+
+    def mock_allocation_entry_generator(field, value):
+        yield []
+
+    mocker.patch('sageintacctsdk.apis.Allocations.count', return_value=1)
+    mocker.patch('sageintacctsdk.apis.Allocations.get_all_generator', side_effect=mock_allocations_generator)
+    mocker.patch('sageintacctsdk.apis.AllocationEntry.get_all_generator', side_effect=mock_allocation_entry_generator)
+    sage_intacct_credentials = SageIntacctCredential.objects.get(workspace_id=workspace_id)
+    sage_intacct_connection = SageIntacctConnector(
+        credentials_object=sage_intacct_credentials,
+        workspace_id=workspace_id
+    )
+    sage_intacct_connection.sync_allocations()