diff --git a/kenya_compliance/kenya_compliance/doctype/navari_etims_registered_purchases/navari_etims_registered_purchases.js b/kenya_compliance/kenya_compliance/doctype/navari_etims_registered_purchases/navari_etims_registered_purchases.js index b34d2a9..dd75fb4 100644 --- a/kenya_compliance/kenya_compliance/doctype/navari_etims_registered_purchases/navari_etims_registered_purchases.js +++ b/kenya_compliance/kenya_compliance/doctype/navari_etims_registered_purchases/navari_etims_registered_purchases.js @@ -1,6 +1,6 @@ // Copyright (c) 2024, Navari Ltd and contributors // For license information, please see license.txt -const doctypeName = 'Navari eTims Registered Purchases'; +const doctypeName = "Navari eTims Registered Purchases"; frappe.ui.form.on(doctypeName, { refresh: function (frm) { @@ -8,11 +8,11 @@ frappe.ui.form.on(doctypeName, { if (!frm.is_new()) { frm.add_custom_button( - __('Create Supplier'), + __("Create Supplier"), function () { frappe.call({ method: - 'kenya_compliance.kenya_compliance.apis.apis.create_supplier_from_fetched_registered_purchases', + "kenya_compliance.kenya_compliance.apis.apis.create_supplier_from_fetched_registered_purchases", args: { request_data: { name: frm.doc.name, @@ -28,14 +28,14 @@ frappe.ui.form.on(doctypeName, { }, }); }, - __('eTims Actions'), + __("eTims Actions") ); frm.add_custom_button( - __('Create Items'), + __("Create Items"), function () { frappe.call({ method: - 'kenya_compliance.kenya_compliance.apis.apis.create_items_from_fetched_registered_purchases', + "kenya_compliance.kenya_compliance.apis.apis.create_items_from_fetched_registered_purchases", args: { request_data: { name: frm.doc.name, @@ -49,14 +49,14 @@ frappe.ui.form.on(doctypeName, { }, }); }, - __('eTims Actions'), + __("eTims Actions") ); frm.add_custom_button( - __('Create Purchase Invoice'), + __("Create Purchase Invoice"), function () { frappe.call({ method: - 'kenya_compliance.kenya_compliance.apis.apis.create_purchase_invoice_from_request', + "kenya_compliance.kenya_compliance.apis.apis.create_purchase_invoice_from_request", args: { request_data: { name: frm.doc.name, @@ -75,7 +75,7 @@ frappe.ui.form.on(doctypeName, { }, }); }, - __('eTims Actions'), + __("eTims Actions") ); // frm.add_custom_button( // __("Create Purchase Receipt"), diff --git a/kenya_compliance/kenya_compliance/doctype/navari_kra_etims_settings/test_navari_kra_etims_settings.py b/kenya_compliance/kenya_compliance/doctype/navari_kra_etims_settings/test_navari_kra_etims_settings.py index 2219ab0..c968f2e 100644 --- a/kenya_compliance/kenya_compliance/doctype/navari_kra_etims_settings/test_navari_kra_etims_settings.py +++ b/kenya_compliance/kenya_compliance/doctype/navari_kra_etims_settings/test_navari_kra_etims_settings.py @@ -283,3 +283,4 @@ def test_auto_creation_of_acct_dimension(self) -> None: new_setting.save() self.assertTrue(frappe.db.exists("Accounting Dimension", "Branch", cache=False)) + \ No newline at end of file diff --git a/kenya_compliance/kenya_compliance/overrides/client/purchase_invoice.js b/kenya_compliance/kenya_compliance/overrides/client/purchase_invoice.js index 452beaa..1e36dd8 100644 --- a/kenya_compliance/kenya_compliance/overrides/client/purchase_invoice.js +++ b/kenya_compliance/kenya_compliance/overrides/client/purchase_invoice.js @@ -1,11 +1,11 @@ -const parentDoctype = 'Purchase Invoice'; -const settingsDoctypeName = 'Navari KRA eTims Settings'; +const parentDoctype = "Purchase Invoice"; +const settingsDoctypeName = "Navari KRA eTims Settings"; frappe.ui.form.on(parentDoctype, { refresh: function (frm) { - frm.set_value('update_stock', 1); + frm.set_value("update_stock", 1); if (frm.doc.update_stock === 1) { - frm.toggle_reqd('set_warehouse', true); + frm.toggle_reqd("set_warehouse", true); } }, validate: function (frm) { @@ -13,38 +13,38 @@ frappe.ui.form.on(parentDoctype, { settingsDoctypeName, { is_active: 1, - bhfid: frm.doc.branch ?? '00', - company: frappe.defaults.get_user_default('Company'), + bhfid: frm.doc.branch ?? "00", + company: frappe.defaults.get_user_default("Company"), }, [ - 'name', - 'company', - 'bhfid', - 'purchases_purchase_type', - 'purchases_receipt_type', - 'purchases_payment_type', - 'purchases_purchase_status', + "name", + "company", + "bhfid", + "purchases_purchase_type", + "purchases_receipt_type", + "purchases_payment_type", + "purchases_purchase_status", ], (response) => { if (!frm.doc.custom_purchase_type) { frm.set_value( - 'custom_purchase_type', - response.purchases_purchase_type, + "custom_purchase_type", + response.purchases_purchase_type ); } if (!frm.doc.custom_receipt_type) { - frm.set_value('custom_receipt_type', response.purchases_receipt_type); + frm.set_value("custom_receipt_type", response.purchases_receipt_type); } if (!frm.doc.custom_purchase_status) { frm.set_value( - 'custom_purchase_status', - response.purchases_purchase_status, + "custom_purchase_status", + response.purchases_purchase_status ); } if (!frm.doc.custom_payment_type) { - frm.set_value('custom_payment_type', response.purchases_payment_type); + frm.set_value("custom_payment_type", response.purchases_payment_type); } - }, + } ); }, }); diff --git a/kenya_compliance/kenya_compliance/overrides/client/sales_invoice.js b/kenya_compliance/kenya_compliance/overrides/client/sales_invoice.js index a523ef1..2ab4497 100644 --- a/kenya_compliance/kenya_compliance/overrides/client/sales_invoice.js +++ b/kenya_compliance/kenya_compliance/overrides/client/sales_invoice.js @@ -1,15 +1,15 @@ -const parentDoctype = 'Sales Invoice'; +const parentDoctype = "Sales Invoice"; const childDoctype = `${parentDoctype} Item`; -const packagingUnitDoctypeName = 'Navari eTims Packaging Unit'; -const unitOfQuantityDoctypeName = 'Navari eTims Unit of Quantity'; -const taxationTypeDoctypeName = 'Navari KRA eTims Taxation Type'; -const settingsDoctypeName = 'Navari KRA eTims Settings'; +const packagingUnitDoctypeName = "Navari eTims Packaging Unit"; +const unitOfQuantityDoctypeName = "Navari eTims Unit of Quantity"; +const taxationTypeDoctypeName = "Navari KRA eTims Taxation Type"; +const settingsDoctypeName = "Navari KRA eTims Settings"; frappe.ui.form.on(parentDoctype, { refresh: function (frm) { - frm.set_value('update_stock', 1); + frm.set_value("update_stock", 1); if (frm.doc.update_stock === 1) { - frm.toggle_reqd('set_warehouse', true); + frm.toggle_reqd("set_warehouse", true); } }, validate: function (frm) { @@ -18,26 +18,26 @@ frappe.ui.form.on(parentDoctype, { { is_active: 1, bhfid: frm.doc.branch, - company: frappe.defaults.get_user_default('Company'), + company: frappe.defaults.get_user_default("Company"), }, [ - 'name', - 'company', - 'bhfid', - 'sales_payment_type', - 'sales_transaction_progress', + "name", + "company", + "bhfid", + "sales_payment_type", + "sales_transaction_progress", ], (response) => { if (!frm.doc.custom_payment_type) { - frm.set_value('custom_payment_type', response.sales_payment_type); + frm.set_value("custom_payment_type", response.sales_payment_type); } if (!frm.doc.custom_transaction_progres) { frm.set_value( - 'custom_transaction_progres', - response.sales_transaction_progress, + "custom_transaction_progres", + response.sales_transaction_progress ); } - }, + } ); }, }); @@ -49,14 +49,14 @@ frappe.ui.form.on(childDoctype, { if (!taxationType) { frappe.db.get_value( - 'Item', + "Item", { item_code: item }, - ['custom_taxation_type'], + ["custom_taxation_type"], (response) => { locals[cdt][cdn].custom_taxation_type = response.custom_taxation_type; locals[cdt][cdn].custom_taxation_type_code = response.custom_taxation_type; - }, + } ); } }, @@ -69,12 +69,12 @@ frappe.ui.form.on(childDoctype, { { name: packagingUnit, }, - ['code'], + ["code"], (response) => { const code = response.code; locals[cdt][cdn].custom_packaging_unit_code = code; - frm.refresh_field('custom_packaging_unit_code'); - }, + frm.refresh_field("custom_packaging_unit_code"); + } ); } }, @@ -87,12 +87,12 @@ frappe.ui.form.on(childDoctype, { { name: unitOfQuantity, }, - ['code'], + ["code"], (response) => { const code = response.code; locals[cdt][cdn].custom_unit_of_quantity_code = code; - frm.refresh_field('custom_unit_of_quantity_code'); - }, + frm.refresh_field("custom_unit_of_quantity_code"); + } ); } }, diff --git a/kenya_compliance/kenya_compliance/overrides/server/item.py b/kenya_compliance/kenya_compliance/overrides/server/item.py index 72ebef3..ae6c1c3 100644 --- a/kenya_compliance/kenya_compliance/overrides/server/item.py +++ b/kenya_compliance/kenya_compliance/overrides/server/item.py @@ -20,7 +20,7 @@ def before_insert(doc: Document, method: str) -> None: """Item doctype before insertion hook""" - item_registration_data = { + item_registration_data = { "name": doc.name, "company_name": frappe.defaults.get_user_default("Company"), "itemCd": doc.custom_item_code_etims, diff --git a/kenya_compliance/kenya_compliance/overrides/server/pos_invoice.py b/kenya_compliance/kenya_compliance/overrides/server/pos_invoice.py index 0e0b376..943c8c8 100644 --- a/kenya_compliance/kenya_compliance/overrides/server/pos_invoice.py +++ b/kenya_compliance/kenya_compliance/overrides/server/pos_invoice.py @@ -7,4 +7,4 @@ def on_submit(doc: Document, method: str) -> None: """Intercepts POS invoice on submit event""" if not doc.custom_successfully_submitted: - generic_invoices_on_submit_override(doc, "POS Invoice") + generic_invoices_on_submit_override(doc, "POS Invoice") diff --git a/kenya_compliance/kenya_compliance/overrides/server/purchase_invoice.py b/kenya_compliance/kenya_compliance/overrides/server/purchase_invoice.py index df86847..37bc720 100644 --- a/kenya_compliance/kenya_compliance/overrides/server/purchase_invoice.py +++ b/kenya_compliance/kenya_compliance/overrides/server/purchase_invoice.py @@ -118,7 +118,7 @@ def build_purchase_invoice_payload(doc: Document) -> dict: "taxblAmtB": doc.custom_taxbl_amount_b or 0, "taxblAmtC": doc.custom_taxbl_amount_c or 0, "taxblAmtD": doc.custom_taxbl_amount_d or 0, - "taxblAmtE": doc.custom_taxbl_amount_e or 0, + "taxblAmtE": doc.custom_taxbl_amount_e or 0, "taxRtA": 0, "taxRtB": 16 if doc.custom_tax_b else 0, "taxRtC": 0, diff --git a/kenya_compliance/kenya_compliance/overrides/server/shared_overrides.py b/kenya_compliance/kenya_compliance/overrides/server/shared_overrides.py index 45de37e..f13d24f 100644 --- a/kenya_compliance/kenya_compliance/overrides/server/shared_overrides.py +++ b/kenya_compliance/kenya_compliance/overrides/server/shared_overrides.py @@ -65,7 +65,7 @@ def generic_invoices_on_submit_override( timeout=300, job_name=f"{doc.name}_send_sales_request", doctype=invoice_type, - document_name=doc.name, + document_name=doc.name, ) diff --git a/kenya_compliance/kenya_compliance/overrides/server/stock_ledger_entry.py b/kenya_compliance/kenya_compliance/overrides/server/stock_ledger_entry.py index e6a014f..bc54e54 100644 --- a/kenya_compliance/kenya_compliance/overrides/server/stock_ledger_entry.py +++ b/kenya_compliance/kenya_compliance/overrides/server/stock_ledger_entry.py @@ -13,7 +13,7 @@ ) from ...utils import ( build_headers, - extract_document_series_number, + extract_document_series_number, get_route_path, get_server_url, split_user_email, @@ -26,7 +26,7 @@ def on_update(doc: Document, method: str | None = None) -> None: company_name = doc.company all_items = frappe.db.get_all( "Item", ["*"] - ) # Get all items to filter and fetch metadata + ) # Get all items to filter and fetch metadata record = frappe.get_doc(doc.voucher_type, doc.voucher_no) series_no = extract_document_series_number(record) payload = { @@ -35,7 +35,7 @@ def on_update(doc: Document, method: str | None = None) -> None: "regTyCd": "M", "custTin": None, "custNm": None, - "custBhfId": get_warehouse_branch_id(doc.warehouse) or None, + "custBhfId": get_warehouse_branch_id(doc.get('warehouse')) or None, "ocrnDt": record.posting_date.strftime("%Y%m%d"), "totTaxblAmt": 0, "totItemCnt": len(record.items), @@ -47,7 +47,16 @@ def on_update(doc: Document, method: str | None = None) -> None: "modrNm": record.modified_by, "modrId": split_user_email(record.modified_by), } - headers = build_headers(company_name, record.branch) + try: + headers = build_headers(company_name, record.branch) + except Exception as e: + frappe.log_error(message=str(e), title="Header Building Error") + headers = { + "tin": "", + "bhfId": "", + "cmcKey": "", + "Content-Type": "application/json", + } if doc.voucher_type == "Stock Reconciliation": items_list = get_stock_recon_movement_items_details( @@ -78,9 +87,7 @@ def on_update(doc: Document, method: str | None = None) -> None: if doc.voucher_type == "Stock Entry": items_list = get_stock_entry_movement_items_details(record.items, all_items) - current_item = list( - filter(lambda item: item["itemNm"] == doc.item_code, items_list) - ) + current_item = [item for item in items_list if item["itemNm"] in [i.item_code for i in doc.items]] payload["itemList"] = current_item payload["totItemCnt"] = len(current_item) @@ -224,7 +231,13 @@ def on_update(doc: Document, method: str | None = None) -> None: else: payload["sarTyCd"] = "11" - server_url = get_server_url(company_name, record.branch) + + try: + server_url = get_server_url(company_name, record.branch) + except Exception as e: + frappe.log_error(message=str(e), title="Header Building Error") + server_url = "https://etims-api-sbx.kra.go.ke/etims-api" + route_path, last_request_date = get_route_path("StockIOSaveReq") if headers and server_url and route_path: @@ -258,7 +271,14 @@ def get_stock_entry_movement_items_details( ) -> list[dict]: items_list = [] - for item in records: + for i in records: + try: + item = frappe.get_doc("Item", i["item_code"]) + prc = round(int(i["basic_rate"]), 2) if i["basic_rate"] else 0 + qty = abs(i["qty"]) + except AttributeError: + prc = round(int(item.basic_rate), 2) if item.basic_rate else 0 + qty = abs(item.qty) for fetched_item in all_items: if item.item_code == fetched_item.name: items_list.append( @@ -271,17 +291,13 @@ def get_stock_entry_movement_items_details( "pkgUnitCd": fetched_item.custom_packaging_unit_code, "pkg": 1, "qtyUnitCd": fetched_item.custom_unit_of_quantity_code, - "qty": abs(item.qty), + "qty": qty, "itemExprDt": "", - "prc": ( - round(int(item.basic_rate), 2) if item.basic_rate else 0 - ), - "splyAmt": ( - round(int(item.basic_rate), 2) if item.basic_rate else 0 - ), + "prc": prc, + "splyAmt": prc, # TODO: Handle discounts properly "totDcAmt": 0, - "taxTyCd": fetched_item.custom_taxation_type_code or "B", + "taxTyCd": fetched_item.custom_taxation_type or "B", "taxblAmt": 0, "taxAmt": 0, "totAmt": 0, @@ -297,7 +313,14 @@ def get_stock_recon_movement_items_details( items_list = [] # current_qty - for item in records: + for i in records: + try: + item = frappe.get_doc("Item", i["item_code"]) + prc = round(int(i["valuation_rate"]), 2) if i["valuation_rate"] else 0 + qty = abs(int(i["quantity_difference"])) + except AttributeError: + prc = round(int(item.valuation_rate), 2) if item.valuation_rate else 0 + qty = abs(item.quantity_difference) for fetched_item in all_items: if item.item_code == fetched_item.name: items_list.append( @@ -310,24 +333,16 @@ def get_stock_recon_movement_items_details( "pkgUnitCd": fetched_item.custom_packaging_unit_code, "pkg": 1, "qtyUnitCd": fetched_item.custom_unit_of_quantity_code, - "qty": abs(int(item.quantity_difference)), + "qty": qty, "itemExprDt": "", - "prc": ( - round(int(item.valuation_rate), 2) - if item.valuation_rate - else 0 - ), - "splyAmt": ( - round(int(item.valuation_rate), 2) - if item.valuation_rate - else 0 - ), + "prc": prc, + "splyAmt": prc, "totDcAmt": 0, - "taxTyCd": fetched_item.custom_taxation_type_code or "B", + "taxTyCd": fetched_item.custom_taxation_type or "B", "taxblAmt": 0, "taxAmt": 0, "totAmt": 0, - "quantity_difference": item.quantity_difference, + "quantity_difference": i["quantity_difference"], } ) @@ -339,7 +354,14 @@ def get_purchase_docs_items_details( ) -> list[dict]: items_list = [] - for item in items: + for i in items: + try: + item = frappe.get_doc("Item", i["item_code"]) + prc = round(int(i["valuation_rate"]), 2) if i["valuation_rate"] else 0 + qty = abs(i["qty"]) + except AttributeError: + prc = round(int(item.valuation_rate), 2) if item.valuation_rate else 0 + qty = abs(item.qty) for fetched_item in all_present_items: if item.item_code == fetched_item.name: items_list.append( @@ -352,20 +374,12 @@ def get_purchase_docs_items_details( "pkgUnitCd": fetched_item.custom_packaging_unit_code, "pkg": 1, "qtyUnitCd": fetched_item.custom_unit_of_quantity_code, - "qty": abs(item.qty), + "qty": qty, "itemExprDt": "", - "prc": ( - round(int(item.valuation_rate), 2) - if item.valuation_rate - else 0 - ), - "splyAmt": ( - round(int(item.valuation_rate), 2) - if item.valuation_rate - else 0 - ), + "prc": prc, + "splyAmt": prc, "totDcAmt": 0, - "taxTyCd": fetched_item.custom_taxation_type_code or "B", + "taxTyCd": fetched_item.custom_taxation_type or "B", "taxblAmt": 0, "taxAmt": 0, "totAmt": 0, @@ -380,7 +394,7 @@ def get_purchase_docs_items_details( } ) - return items_list + return items_list def get_notes_docs_items_details( @@ -388,7 +402,14 @@ def get_notes_docs_items_details( ) -> list[dict]: items_list = [] - for item in items: + for i in items: + try: + item = frappe.get_doc("Item", i["item_code"]) + prc = round(int(i["base_net_rate"]), 2) if i["base_net_rate"] else 0 + qty = abs(i["qty"]) + except AttributeError: + prc = round(int(item.basic_rate), 2) if item.basic_rate else 0 + qty = abs(item.qty) for fetched_item in all_present_items: if item.item_code == fetched_item.name: items_list.append( @@ -401,20 +422,12 @@ def get_notes_docs_items_details( "pkgUnitCd": fetched_item.custom_packaging_unit_code, "pkg": 1, "qtyUnitCd": fetched_item.custom_unit_of_quantity_code, - "qty": abs(item.qty), + "qty": qty, "itemExprDt": "", - "prc": ( - round(int(item.base_net_rate), 2) - if item.base_net_rate - else 0 - ), - "splyAmt": ( - round(int(item.base_net_rate), 2) - if item.base_net_rate - else 0 - ), + "prc": prc, + "splyAmt": prc, "totDcAmt": 0, - "taxTyCd": fetched_item.custom_taxation_type_code or "B", + "taxTyCd": fetched_item.custom_taxation_type or "B", "taxblAmt": 0, "taxAmt": 0, "totAmt": 0, @@ -425,11 +438,14 @@ def get_notes_docs_items_details( def get_warehouse_branch_id(warehouse_name: str) -> str | Literal[0]: - branch_id = frappe.db.get_value( - "Warehouse", {"name": warehouse_name}, ["custom_branch"], as_dict=True - ) + try: + branch_id = frappe.db.get_value( + "Warehouse", {"name": warehouse_name}, ["custom_branch"], as_dict=True + ) - if branch_id: - return branch_id.custom_branch + if branch_id: + return branch_id.custom_branch + except: + pass return 0 diff --git a/kenya_compliance/kenya_compliance/overrides/server/test_shared_overrides.py b/kenya_compliance/kenya_compliance/overrides/server/test_shared_overrides.py new file mode 100644 index 0000000..9b8be5f --- /dev/null +++ b/kenya_compliance/kenya_compliance/overrides/server/test_shared_overrides.py @@ -0,0 +1,105 @@ +import frappe +from frappe.tests.utils import FrappeTestCase +from unittest.mock import patch, MagicMock +from .shared_overrides import generic_invoices_on_submit_override + +class TestGenericInvoicesOnSubmit(FrappeTestCase): + """Test Cases for Generic Invoices On Submit""" + + def setUp(self) -> None: + pass + + @patch('frappe.enqueue') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.build_headers') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.get_server_url') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.get_route_path') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.build_invoice_payload') + def test_generic_invoices_on_submit_sales_invoice( + self, mock_build_invoice_payload, mock_get_route_path, mock_get_server_url, mock_build_headers, mock_enqueue + ): + """Test case for submitting a Sales Invoice.""" + + # Mocking the expected outputs + mock_build_headers.return_value = {'Authorization': 'Bearer token'} + mock_get_server_url.return_value = 'https://server.url' + mock_get_route_path.return_value = ('/api/route', 'last_date') + mock_build_invoice_payload.return_value = {'invcNo': 'INV-001'} + + # Creating a mock document object + doc = MagicMock() + doc.company = "Test Company" + doc.branch = "Main Branch" + doc.is_return = False + doc.name = "INV-001" + + # Call the function being tested + generic_invoices_on_submit_override(doc, "Sales Invoice") + + # Validating the behavior + mock_build_headers.assert_called_with('Test Company', 'Main Branch') + mock_get_server_url.assert_called_with('Test Company', 'Main Branch') + mock_get_route_path.assert_called_with('TrnsSalesSaveWrReq') + mock_build_invoice_payload.assert_called_with(doc, "S", "Test Company") + mock_enqueue.assert_called_once() + + @patch('frappe.enqueue') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.build_headers') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.get_server_url') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.get_route_path') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.build_invoice_payload') + def test_generic_invoices_on_submit_pos_invoice( + self, mock_build_invoice_payload, mock_get_route_path, mock_get_server_url, mock_build_headers, mock_enqueue + ): + """Test case for submitting a POS Invoice.""" + + # Mocking the expected outputs + mock_build_headers.return_value = {'Authorization': 'Bearer token'} + mock_get_server_url.return_value = 'https://server.url' + mock_get_route_path.return_value = ('/api/route', 'last_date') + mock_build_invoice_payload.return_value = {'invcNo': 'POS-001'} + + # Creating a mock document object + doc = MagicMock() + doc.company = "Test Company" + doc.branch = "Main Branch" + doc.is_return = True + doc.name = "POS-001" + + # Call the function being tested + generic_invoices_on_submit_override(doc, "POS Invoice") + + # Validating the behavior + mock_build_headers.assert_called_with('Test Company', 'Main Branch') + mock_get_server_url.assert_called_with('Test Company', 'Main Branch') + mock_get_route_path.assert_called_with('TrnsSalesSaveWrReq') + mock_build_invoice_payload.assert_called_with(doc, "C", "Test Company") + mock_enqueue.assert_called_once() + + @patch('frappe.enqueue') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.build_headers') + @patch('kenya_compliance.kenya_compliance.overrides.server.shared_overrides.get_server_url') + def test_generic_invoices_on_submit_missing_server_url( + self, mock_get_server_url, mock_build_headers, mock_enqueue + ): + """Test case when the server URL is missing.""" + + # Mocking the expected outputs + mock_build_headers.return_value = {'Authorization': 'Bearer token'} + mock_get_server_url.return_value = None # No server URL + + # Creating a mock document object + doc = MagicMock() + doc.company = "Test Company" + doc.branch = "Main Branch" + doc.is_return = False + doc.name = "INV-001" + + # Call the function being tested + generic_invoices_on_submit_override(doc, "Sales Invoice") + + # Validating the behavior: enqueue should not be called since server URL is missing + mock_enqueue.assert_not_called() + + def tearDown(self) -> None: + """Clean up after each test.""" + super().tearDown() diff --git a/kenya_compliance/kenya_compliance/overrides/server/test_stock_ledger_entry.py b/kenya_compliance/kenya_compliance/overrides/server/test_stock_ledger_entry.py new file mode 100644 index 0000000..cdd510f --- /dev/null +++ b/kenya_compliance/kenya_compliance/overrides/server/test_stock_ledger_entry.py @@ -0,0 +1,224 @@ +import frappe +from frappe.tests.utils import FrappeTestCase +from .stock_ledger_entry import ( + on_update, + get_stock_entry_movement_items_details, + get_stock_recon_movement_items_details, + get_purchase_docs_items_details, + get_notes_docs_items_details, + get_warehouse_branch_id +) +from hashlib import sha256 + +class TestStockLedgerEntry(FrappeTestCase): + + def setUp(self) -> None: + """ + Setup method to create necessary test data before running each test. + """ + # Define test data + self.company_name = "Test Company" + self.customer_name = "Test Customer" + self.supplier_name = "Test Supplier" + self.item_code = "TEST-ITEM" + self.warehouse_name = "Test Warehouse" + self.from_currency = "KES" + self.to_currency = "USD" + + # Define document types and their filters for cleanup + self.doc_types = [ + ("Stock Entry", {"company": self.company_name}), + ("Stock Reconciliation", {"company": self.company_name}), + ("Purchase Invoice", {"company": self.company_name}), + ("Delivery Note", {"company": self.company_name}), + ("Currency Exchange", {"from_currency": self.from_currency, "to_currency": self.to_currency}), + ("Item", {"item_code": self.item_code}), + ("Warehouse", {"warehouse_name": self.warehouse_name}), + ("Supplier", {"supplier_name": self.supplier_name}), + ("Customer", {"customer_name": self.customer_name}), + ] + + # Cleanup existing records if they exist + for doc_type, filters in self.doc_types: + for doc in frappe.get_all(doc_type, filters=filters): + frappe.delete_doc(doc_type, doc.name) + frappe.delete_doc( + "Company", + frappe.get_value( + "Company", + {"abbr": "TTC", "company_name": "Test Company"}, + ), + force=1, + ignore_permissions=True, + ) + frappe.delete_doc("Branch", "00", force=1, ignore_permissions=True) + + # Create Test Company and branch + self.company, self.branch_doc = create_test_company() + + # Create test records + self.customer = frappe.get_doc({ + "doctype": "Customer", + "customer_name": self.customer_name + }).insert() + + self.supplier = frappe.get_doc({ + "doctype": "Supplier", + "supplier_name": self.supplier_name + }).insert() + + self.warehouse = frappe.get_doc({ + "doctype": "Warehouse", + "warehouse_name": self.warehouse_name, + "company": self.company.name + }).insert() + + # Fetch existing records for item setup + self.custom_product_type = frappe.get_all("Navari eTims Product Type", limit=1)[0] + self.packaging_unit_doc = frappe.get_all("Navari eTims Packaging Unit", limit=1)[0] + self.unit_of_quantity_doc = frappe.get_all("Navari eTims Unit of Quantity", limit=1)[0] + self.custom_item_classification = frappe.get_all("Navari KRA eTims Item Classification", limit=1)[0] + self.custom_etims_country_of_origin = frappe.get_all("Navari eTims Country", limit=1)[0] + + # Create or get a test item + self.item = frappe.get_doc({ + "doctype": "Item", + "item_code": self.item_code, + "item_name": "Test Item", + "item_group": "All Item Groups", + "stock_uom": "Unit", + "custom_product_type": self.custom_product_type.name, + "custom_packaging_unit": self.packaging_unit_doc.name, + "custom_unit_of_quantity": self.unit_of_quantity_doc.name, + "custom_item_classification": self.custom_item_classification.name, + "custom_etims_country_of_origin": self.custom_etims_country_of_origin.name, + "default_warehouse": self.warehouse.name + }).insert() + + # Create a test stock entry + self.stock_entry = frappe.get_doc({ + 'doctype': 'Stock Entry', + 'stock_entry_type': 'Material Receipt', + 'company': 'Test Company', + 'items': [{ + 'item_code': self.item.item_code, + 'qty': 10, + 't_warehouse': self.warehouse.name + }] + }).insert() + + # Add currency exchange rate + self.currency_exchange_rate = frappe.get_doc({ + "doctype": "Currency Exchange", + "from_currency": self.from_currency, + "to_currency": self.to_currency, + "exchange_rate": 0.007 + }).insert() + + def tearDown(self) -> None: + """ + Cleanup method to remove test data after each test. + """ + # Delete test documents + frappe.delete_doc('Stock Entry', self.stock_entry.name) + frappe.delete_doc('Item', self.item.name) + frappe.delete_doc('Warehouse', self.warehouse.name) + + def test_get_warehouse_branch_id(self) -> None: + # Set branch ID for warehouse + frappe.db.set_value('Warehouse', self.warehouse.name, 'custom_branch', 'BR001') + result = get_warehouse_branch_id(self.warehouse.name) + self.assertEqual(result, 'BR001') + + def test_on_update_stock_entry(self) -> None: + # Get the Stock Entry document before calling on_update + doc_before = frappe.get_doc('Stock Entry', self.stock_entry.name) + doc_before.voucher_type = "Stock Entry" + doc_before.voucher_no = doc_before.name + + # Call the on_update method with the document + on_update(doc_before) + + # Reload the document to reflect any updates made by the on_update function + doc_after = frappe.get_doc('Stock Entry', self.stock_entry.name) + + # Assertions to verify expected updates or effects of on_update + self.assertEqual(doc_after.modified_by, frappe.session.user, "Document should be modified by the current user") + self.assertIsNotNone(doc_after.modified, "Modified time should be updated after on_update") + + # Check if a job for the stock entry exists in the job queue + job_queue = frappe.get_all('RQ Job', filters={'job_name': sha256(f"{doc_before.name}{doc_before.creation}{doc_before.modified}".encode(), usedforsecurity=False).hexdigest()}) + self.assertGreater(len(job_queue), 0, "A job for the stock entry should be present in the job queue") + + def test_get_stock_entry_movement_items_details(self) -> None: + records = [{ + 'item_code': self.item.item_code, + 'qty': 10, + 'basic_rate': 100 + }] + stock_movement_details = get_stock_entry_movement_items_details(records, [self.item]) + self.assertEqual(len(stock_movement_details), 1) + self.assertEqual(stock_movement_details[0]['itemNm'], self.item.item_code) + self.assertEqual(stock_movement_details[0]['prc'], 100) + + def test_get_stock_recon_movement_items_details(self) -> None: + records = [{ + 'item_code': self.item.item_code, + 'quantity_difference': 5, + 'valuation_rate': 50 + }] + recon_movement_details = get_stock_recon_movement_items_details(records, [self.item]) + self.assertEqual(len(recon_movement_details), 1) + self.assertEqual(recon_movement_details[0]['itemNm'], self.item.item_code) + self.assertEqual(recon_movement_details[0]['qty'], 5) + + def test_get_purchase_docs_items_details(self) -> None: + items = [{ + 'item_code': self.item.item_code, + 'qty': 10, + 'valuation_rate': 100 + }] + purchase_details = get_purchase_docs_items_details(items, [self.item]) + self.assertEqual(len(purchase_details), 1) + self.assertEqual(purchase_details[0]['itemNm'], self.item.item_code) + + def test_get_notes_docs_items_details(self) -> None: + items = [{ + 'item_code': self.item.item_code, + 'qty': 10, + 'base_net_rate': 100 + }] + notes_details = get_notes_docs_items_details(items, [self.item]) + self.assertEqual(len(notes_details), 1) + self.assertEqual(notes_details[0]['itemNm'], self.item.item_code) + self.assertEqual(notes_details[0]['prc'], 100) + +def create_test_company(): + # Delete any existing test company with the same identifier + existing_company = frappe.get_all("Company", filters={"abbr": "TTC", "company_name": "Test Company"}) + if existing_company: + company_name = frappe.get_value("Company", {"abbr": "TTC", "company_name": "Test Company"}) + frappe.delete_doc("Company", company_name, force=1, ignore_permissions=True) + + # Create the test company + company = frappe.get_doc({ + "doctype": "Company", + "company_name": "Test Company", + "abbr": "TTC", + "default_currency": "USD", + "country": "Kenya", + "tax_id": "A123456787Z" + }) + company.insert(ignore_permissions=True) + + # Create a test branch + branch = frappe.get_doc({ + "doctype": "Branch", + "branch": "00", + "company": company.name, + "abbr": "BRNT", + "is_group": 0 + }) + branch.insert(ignore_permissions=True) + + return company, branch