test: timeout certain tests in work order to avoid stuck tests (#28666)
[skip ci]
This commit is contained in:
@@ -21,9 +21,10 @@ from erpnext.stock.doctype.item.test_item import create_item, make_item
|
|||||||
from erpnext.stock.doctype.stock_entry import test_stock_entry
|
from erpnext.stock.doctype.stock_entry import test_stock_entry
|
||||||
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
|
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
|
||||||
from erpnext.stock.utils import get_bin
|
from erpnext.stock.utils import get_bin
|
||||||
|
from erpnext.tests.utils import ERPNextTestCase, timeout
|
||||||
|
|
||||||
|
|
||||||
class TestWorkOrder(unittest.TestCase):
|
class TestWorkOrder(ERPNextTestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.warehouse = '_Test Warehouse 2 - _TC'
|
self.warehouse = '_Test Warehouse 2 - _TC'
|
||||||
self.item = '_Test Item'
|
self.item = '_Test Item'
|
||||||
@@ -376,6 +377,7 @@ class TestWorkOrder(unittest.TestCase):
|
|||||||
self.assertEqual(len(ste.additional_costs), 1)
|
self.assertEqual(len(ste.additional_costs), 1)
|
||||||
self.assertEqual(ste.total_additional_costs, 1000)
|
self.assertEqual(ste.total_additional_costs, 1000)
|
||||||
|
|
||||||
|
@timeout(seconds=60)
|
||||||
def test_job_card(self):
|
def test_job_card(self):
|
||||||
stock_entries = []
|
stock_entries = []
|
||||||
bom = frappe.get_doc('BOM', {
|
bom = frappe.get_doc('BOM', {
|
||||||
@@ -769,6 +771,7 @@ class TestWorkOrder(unittest.TestCase):
|
|||||||
total_pl_qty
|
total_pl_qty
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@timeout(seconds=60)
|
||||||
def test_job_card_scrap_item(self):
|
def test_job_card_scrap_item(self):
|
||||||
items = ['Test FG Item for Scrap Item Test', 'Test RM Item 1 for Scrap Item Test',
|
items = ['Test FG Item for Scrap Item Test', 'Test RM Item 1 for Scrap Item Test',
|
||||||
'Test RM Item 2 for Scrap Item Test']
|
'Test RM Item 2 for Scrap Item Test']
|
||||||
|
|||||||
@@ -2,6 +2,7 @@
|
|||||||
# License: GNU General Public License v3. See license.txt
|
# License: GNU General Public License v3. See license.txt
|
||||||
|
|
||||||
import copy
|
import copy
|
||||||
|
import signal
|
||||||
import unittest
|
import unittest
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
from typing import Any, Dict, NewType, Optional
|
from typing import Any, Dict, NewType, Optional
|
||||||
@@ -135,3 +136,23 @@ def execute_script_report(
|
|||||||
report_execute_fn(filter_with_optional_param)
|
report_execute_fn(filter_with_optional_param)
|
||||||
|
|
||||||
return report_data
|
return report_data
|
||||||
|
|
||||||
|
|
||||||
|
def timeout(seconds=30, error_message="Test timed out."):
|
||||||
|
""" Timeout decorator to ensure a test doesn't run for too long.
|
||||||
|
|
||||||
|
adapted from https://stackoverflow.com/a/2282656"""
|
||||||
|
def decorator(func):
|
||||||
|
def _handle_timeout(signum, frame):
|
||||||
|
raise Exception(error_message)
|
||||||
|
|
||||||
|
def wrapper(*args, **kwargs):
|
||||||
|
signal.signal(signal.SIGALRM, _handle_timeout)
|
||||||
|
signal.alarm(seconds)
|
||||||
|
try:
|
||||||
|
result = func(*args, **kwargs)
|
||||||
|
finally:
|
||||||
|
signal.alarm(0)
|
||||||
|
return result
|
||||||
|
return wrapper
|
||||||
|
return decorator
|
||||||
|
|||||||
Reference in New Issue
Block a user