14 from multiprocessing import Process, Pipe, cpu_count
15 from multiprocessing.queues import Queue
16 from multiprocessing.managers import BaseManager
17 from framework import VppTestRunner, running_extended_tests, VppTestCase, \
18 get_testcase_doc_name, get_test_description, PASS, FAIL, ERROR, SKIP, \
20 from debug import spawn_gdb
21 from log import get_parallel_logger, double_line_delim, RED, YELLOW, GREEN, \
22 colorize, single_line_delim
23 from discover_tests import discover_tests
24 from subprocess import check_output, CalledProcessError
25 from util import check_core_path, get_core_path, is_core_present
27 # timeout which controls how long the child has to finish after seeing
28 # a core dump in test temporary directory. If this is exceeded, parent assumes
29 # that child process is stuck (e.g. waiting for shm mutex, which will never
30 # get unlocked) and kill the child
32 min_req_shm = 536870912 # min 512MB shm required
33 # 128MB per extra process
34 shm_per_process = 134217728
37 class StreamQueue(Queue):
42 sys.__stdout__.flush()
43 sys.__stderr__.flush()
46 return self._writer.fileno()
49 class StreamQueueManager(BaseManager):
53 StreamQueueManager.register('StreamQueue', StreamQueue)
56 class TestResult(dict):
57 def __init__(self, testcase_suite, testcases_by_id=None):
58 super(TestResult, self).__init__()
65 self.testcase_suite = testcase_suite
66 self.testcases = [testcase for testcase in testcase_suite]
67 self.testcases_by_id = testcases_by_id
69 def was_successful(self):
70 return 0 == len(self[FAIL]) == len(self[ERROR]) \
71 and len(self[PASS] + self[SKIP]) \
72 == self.testcase_suite.countTestCases() == len(self[TEST_RUN])
74 def no_tests_run(self):
75 return 0 == len(self[TEST_RUN])
77 def process_result(self, test_id, result):
78 self[result].append(test_id)
80 def suite_from_failed(self):
82 for testcase in self.testcase_suite:
84 if tc_id not in self[PASS] and tc_id not in self[SKIP]:
87 return suite_from_failed(self.testcase_suite, rerun_ids)
89 def get_testcase_names(self, test_id):
90 # could be tearDownClass (test_ipsec_esp.TestIpsecEsp1)
91 setup_teardown_match = re.match(
92 r'((tearDownClass)|(setUpClass)) \((.+\..+)\)', test_id)
93 if setup_teardown_match:
94 test_name, _, _, testcase_name = setup_teardown_match.groups()
95 if len(testcase_name.split('.')) == 2:
96 for key in self.testcases_by_id.keys():
97 if key.startswith(testcase_name):
100 testcase_name = self._get_testcase_doc_name(testcase_name)
102 test_name = self._get_test_description(test_id)
103 testcase_name = self._get_testcase_doc_name(test_id)
105 return testcase_name, test_name
107 def _get_test_description(self, test_id):
108 if test_id in self.testcases_by_id:
109 desc = get_test_description(descriptions,
110 self.testcases_by_id[test_id])
115 def _get_testcase_doc_name(self, test_id):
116 if test_id in self.testcases_by_id:
117 doc_name = get_testcase_doc_name(self.testcases_by_id[test_id])
123 def test_runner_wrapper(suite, keep_alive_pipe, stdouterr_queue,
124 finished_pipe, result_pipe, logger):
125 sys.stdout = stdouterr_queue
126 sys.stderr = stdouterr_queue
127 VppTestCase.parallel_handler = logger.handlers[0]
128 result = VppTestRunner(keep_alive_pipe=keep_alive_pipe,
129 descriptions=descriptions,
131 result_pipe=result_pipe,
133 print_summary=False).run(suite)
134 finished_pipe.send(result.wasSuccessful())
135 finished_pipe.close()
136 keep_alive_pipe.close()
139 class TestCaseWrapper(object):
140 def __init__(self, testcase_suite, manager):
141 self.keep_alive_parent_end, self.keep_alive_child_end = Pipe(
143 self.finished_parent_end, self.finished_child_end = Pipe(duplex=False)
144 self.result_parent_end, self.result_child_end = Pipe(duplex=False)
145 self.testcase_suite = testcase_suite
146 if sys.version[0] == '2':
147 self.stdouterr_queue = manager.StreamQueue()
149 from multiprocessing import get_context
150 self.stdouterr_queue = manager.StreamQueue(ctx=get_context())
151 self.logger = get_parallel_logger(self.stdouterr_queue)
152 self.child = Process(target=test_runner_wrapper,
153 args=(testcase_suite,
154 self.keep_alive_child_end,
155 self.stdouterr_queue,
156 self.finished_child_end,
157 self.result_child_end,
161 self.last_test_temp_dir = None
162 self.last_test_vpp_binary = None
163 self._last_test = None
164 self.last_test_id = None
166 self.last_heard = time.time()
167 self.core_detected_at = None
168 self.testcases_by_id = {}
169 self.testclasess_with_core = {}
170 for testcase in self.testcase_suite:
171 self.testcases_by_id[testcase.id()] = testcase
172 self.result = TestResult(testcase_suite, self.testcases_by_id)
176 return self._last_test
179 def last_test(self, test_id):
180 self.last_test_id = test_id
181 if test_id in self.testcases_by_id:
182 testcase = self.testcases_by_id[test_id]
183 self._last_test = testcase.shortDescription()
184 if not self._last_test:
185 self._last_test = str(testcase)
187 self._last_test = test_id
189 def add_testclass_with_core(self):
190 if self.last_test_id in self.testcases_by_id:
191 test = self.testcases_by_id[self.last_test_id]
192 class_name = unittest.util.strclass(test.__class__)
193 test_name = "'{}' ({})".format(get_test_description(descriptions,
197 test_name = self.last_test_id
198 class_name = re.match(r'((tearDownClass)|(setUpClass)) '
199 r'\((.+\..+)\)', test_name).groups()[3]
200 if class_name not in self.testclasess_with_core:
201 self.testclasess_with_core[class_name] = (
203 self.last_test_vpp_binary,
204 self.last_test_temp_dir)
206 def close_pipes(self):
207 self.keep_alive_child_end.close()
208 self.finished_child_end.close()
209 self.result_child_end.close()
210 self.keep_alive_parent_end.close()
211 self.finished_parent_end.close()
212 self.result_parent_end.close()
214 def was_successful(self):
215 return self.result.was_successful()
218 def stdouterr_reader_wrapper(unread_testcases, finished_unread_testcases,
221 while read_testcases.is_set() or unread_testcases:
222 if finished_unread_testcases:
223 read_testcase = finished_unread_testcases.pop()
224 unread_testcases.remove(read_testcase)
225 elif unread_testcases:
226 read_testcase = unread_testcases.pop()
229 while data is not None:
230 sys.stdout.write(data)
231 data = read_testcase.stdouterr_queue.get()
233 read_testcase.stdouterr_queue.close()
234 finished_unread_testcases.discard(read_testcase)
238 def handle_failed_suite(logger, last_test_temp_dir, vpp_pid):
239 if last_test_temp_dir:
240 # Need to create link in case of a timeout or core dump without failure
241 lttd = os.path.basename(last_test_temp_dir)
242 failed_dir = os.getenv('FAILED_DIR')
243 link_path = '%s%s-FAILED' % (failed_dir, lttd)
244 if not os.path.exists(link_path):
245 os.symlink(last_test_temp_dir, link_path)
246 logger.error("Symlink to failed testcase directory: %s -> %s"
249 # Report core existence
250 core_path = get_core_path(last_test_temp_dir)
251 if os.path.exists(core_path):
253 "Core-file exists in test temporary directory: %s!" %
255 check_core_path(logger, core_path)
256 logger.debug("Running 'file %s':" % core_path)
258 info = check_output(["file", core_path])
260 except CalledProcessError as e:
261 logger.error("Subprocess returned with return code "
262 "while running `file' utility on core-file "
264 "rc=%s", e.returncode)
266 logger.error("Subprocess returned with OS error while "
267 "running 'file' utility "
269 "(%s) %s", e.errno, e.strerror)
270 except Exception as e:
271 logger.exception("Unexpected error running `file' utility "
275 # Copy api post mortem
276 api_post_mortem_path = "/tmp/api_post_mortem.%d" % vpp_pid
277 if os.path.isfile(api_post_mortem_path):
278 logger.error("Copying api_post_mortem.%d to %s" %
279 (vpp_pid, last_test_temp_dir))
280 shutil.copy2(api_post_mortem_path, last_test_temp_dir)
283 def check_and_handle_core(vpp_binary, tempdir, core_crash_test):
284 if is_core_present(tempdir):
285 print('VPP core detected in %s. Last test running was %s' %
286 (tempdir, core_crash_test))
287 print(single_line_delim)
288 spawn_gdb(vpp_binary, get_core_path(tempdir))
289 print(single_line_delim)
292 def handle_cores(failed_testcases):
294 for failed_testcase in failed_testcases:
295 tcs_with_core = failed_testcase.testclasess_with_core
297 for test, vpp_binary, tempdir in tcs_with_core.values():
298 check_and_handle_core(vpp_binary, tempdir, test)
301 def process_finished_testsuite(wrapped_testcase_suite,
302 finished_testcase_suites,
303 failed_wrapped_testcases,
305 results.append(wrapped_testcase_suite.result)
306 finished_testcase_suites.add(wrapped_testcase_suite)
308 if failfast and not wrapped_testcase_suite.was_successful():
311 if not wrapped_testcase_suite.was_successful():
312 failed_wrapped_testcases.add(wrapped_testcase_suite)
313 handle_failed_suite(wrapped_testcase_suite.logger,
314 wrapped_testcase_suite.last_test_temp_dir,
315 wrapped_testcase_suite.vpp_pid)
320 def run_forked(testcase_suites):
321 wrapped_testcase_suites = set()
323 # suites are unhashable, need to use list
325 unread_testcases = set()
326 finished_unread_testcases = set()
327 manager = StreamQueueManager()
329 for i in range(concurrent_tests):
331 wrapped_testcase_suite = TestCaseWrapper(testcase_suites.pop(0),
333 wrapped_testcase_suites.add(wrapped_testcase_suite)
334 unread_testcases.add(wrapped_testcase_suite)
338 read_from_testcases = threading.Event()
339 read_from_testcases.set()
340 stdouterr_thread = threading.Thread(target=stdouterr_reader_wrapper,
341 args=(unread_testcases,
342 finished_unread_testcases,
343 read_from_testcases))
344 stdouterr_thread.start()
346 failed_wrapped_testcases = set()
350 while wrapped_testcase_suites:
351 finished_testcase_suites = set()
352 for wrapped_testcase_suite in wrapped_testcase_suites:
353 while wrapped_testcase_suite.result_parent_end.poll():
354 wrapped_testcase_suite.result.process_result(
355 *wrapped_testcase_suite.result_parent_end.recv())
356 wrapped_testcase_suite.last_heard = time.time()
358 while wrapped_testcase_suite.keep_alive_parent_end.poll():
359 wrapped_testcase_suite.last_test, \
360 wrapped_testcase_suite.last_test_vpp_binary, \
361 wrapped_testcase_suite.last_test_temp_dir, \
362 wrapped_testcase_suite.vpp_pid = \
363 wrapped_testcase_suite.keep_alive_parent_end.recv()
364 wrapped_testcase_suite.last_heard = time.time()
366 if wrapped_testcase_suite.finished_parent_end.poll():
367 wrapped_testcase_suite.finished_parent_end.recv()
368 wrapped_testcase_suite.last_heard = time.time()
369 stop_run = process_finished_testsuite(
370 wrapped_testcase_suite,
371 finished_testcase_suites,
372 failed_wrapped_testcases,
377 if wrapped_testcase_suite.last_heard + test_timeout < \
380 wrapped_testcase_suite.logger.critical(
381 "Child test runner process timed out "
382 "(last test running was `%s' in `%s')!" %
383 (wrapped_testcase_suite.last_test,
384 wrapped_testcase_suite.last_test_temp_dir))
385 elif not wrapped_testcase_suite.child.is_alive():
387 wrapped_testcase_suite.logger.critical(
388 "Child test runner process unexpectedly died "
389 "(last test running was `%s' in `%s')!" %
390 (wrapped_testcase_suite.last_test,
391 wrapped_testcase_suite.last_test_temp_dir))
392 elif wrapped_testcase_suite.last_test_temp_dir and \
393 wrapped_testcase_suite.last_test_vpp_binary:
395 wrapped_testcase_suite.last_test_temp_dir):
396 wrapped_testcase_suite.add_testclass_with_core()
397 if wrapped_testcase_suite.core_detected_at is None:
398 wrapped_testcase_suite.core_detected_at = \
400 elif wrapped_testcase_suite.core_detected_at + \
401 core_timeout < time.time():
402 wrapped_testcase_suite.logger.critical(
403 "Child test runner process unresponsive and "
404 "core-file exists in test temporary directory "
405 "(last test running was `%s' in `%s')!" %
406 (wrapped_testcase_suite.last_test,
407 wrapped_testcase_suite.last_test_temp_dir))
411 wrapped_testcase_suite.child.terminate()
413 # terminating the child process tends to leave orphan
415 if wrapped_testcase_suite.vpp_pid:
416 os.kill(wrapped_testcase_suite.vpp_pid,
421 wrapped_testcase_suite.result.crashed = True
422 wrapped_testcase_suite.result.process_result(
423 wrapped_testcase_suite.last_test_id, ERROR)
424 stop_run = process_finished_testsuite(
425 wrapped_testcase_suite,
426 finished_testcase_suites,
427 failed_wrapped_testcases,
430 for finished_testcase in finished_testcase_suites:
431 finished_testcase.child.join()
432 finished_testcase.close_pipes()
433 wrapped_testcase_suites.remove(finished_testcase)
434 finished_unread_testcases.add(finished_testcase)
435 finished_testcase.stdouterr_queue.put(None)
437 while testcase_suites:
438 results.append(TestResult(testcase_suites.pop(0)))
439 elif testcase_suites:
440 new_testcase = TestCaseWrapper(testcase_suites.pop(0),
442 wrapped_testcase_suites.add(new_testcase)
443 unread_testcases.add(new_testcase)
446 for wrapped_testcase_suite in wrapped_testcase_suites:
447 wrapped_testcase_suite.child.terminate()
448 wrapped_testcase_suite.stdouterr_queue.put(None)
451 read_from_testcases.clear()
452 stdouterr_thread.join(test_timeout)
455 handle_cores(failed_wrapped_testcases)
459 class SplitToSuitesCallback:
460 def __init__(self, filter_callback):
462 self.suite_name = 'default'
463 self.filter_callback = filter_callback
464 self.filtered = unittest.TestSuite()
466 def __call__(self, file_name, cls, method):
467 test_method = cls(method)
468 if self.filter_callback(file_name, cls.__name__, method):
469 self.suite_name = file_name + cls.__name__
470 if self.suite_name not in self.suites:
471 self.suites[self.suite_name] = unittest.TestSuite()
472 self.suites[self.suite_name].addTest(test_method)
475 self.filtered.addTest(test_method)
481 def parse_test_option():
482 f = os.getenv(test_option, None)
483 filter_file_name = None
484 filter_class_name = None
485 filter_func_name = None
490 raise Exception("Unrecognized %s option: %s" %
493 if parts[2] not in ('*', ''):
494 filter_func_name = parts[2]
495 if parts[1] not in ('*', ''):
496 filter_class_name = parts[1]
497 if parts[0] not in ('*', ''):
498 if parts[0].startswith('test_'):
499 filter_file_name = parts[0]
501 filter_file_name = 'test_%s' % parts[0]
503 if f.startswith('test_'):
506 filter_file_name = 'test_%s' % f
508 filter_file_name = '%s.py' % filter_file_name
509 return filter_file_name, filter_class_name, filter_func_name
512 def filter_tests(tests, filter_cb):
513 result = unittest.suite.TestSuite()
515 if isinstance(t, unittest.suite.TestSuite):
516 # this is a bunch of tests, recursively filter...
517 x = filter_tests(t, filter_cb)
518 if x.countTestCases() > 0:
520 elif isinstance(t, unittest.TestCase):
521 # this is a single test
522 parts = t.id().split('.')
523 # t.id() for common cases like this:
524 # test_classifier.TestClassifier.test_acl_ip
525 # apply filtering only if it is so
527 if not filter_cb(parts[0], parts[1], parts[2]):
531 # unexpected object, don't touch it
536 class FilterByTestOption:
537 def __init__(self, filter_file_name, filter_class_name, filter_func_name):
538 self.filter_file_name = filter_file_name
539 self.filter_class_name = filter_class_name
540 self.filter_func_name = filter_func_name
542 def __call__(self, file_name, class_name, func_name):
543 if self.filter_file_name:
544 fn_match = fnmatch.fnmatch(file_name, self.filter_file_name)
547 if self.filter_class_name and class_name != self.filter_class_name:
549 if self.filter_func_name and func_name != self.filter_func_name:
554 class FilterByClassList:
555 def __init__(self, classes_with_filenames):
556 self.classes_with_filenames = classes_with_filenames
558 def __call__(self, file_name, class_name, func_name):
559 return '.'.join([file_name, class_name]) in self.classes_with_filenames
562 def suite_from_failed(suite, failed):
563 failed = {x.rsplit('.', 1)[0] for x in failed}
564 filter_cb = FilterByClassList(failed)
565 suite = filter_tests(suite, filter_cb)
569 class AllResults(dict):
571 super(AllResults, self).__init__()
572 self.all_testcases = 0
573 self.results_per_suite = []
580 self.testsuites_no_tests_run = []
582 def add_results(self, result):
583 self.results_per_suite.append(result)
584 result_types = [PASS, FAIL, ERROR, SKIP, TEST_RUN]
585 for result_type in result_types:
586 self[result_type] += len(result[result_type])
588 def add_result(self, result):
590 self.all_testcases += result.testcase_suite.countTestCases()
591 self.add_results(result)
593 if result.no_tests_run():
594 self.testsuites_no_tests_run.append(result.testcase_suite)
599 elif not result.was_successful():
603 self.rerun.append(result.testcase_suite)
607 def print_results(self):
609 print(double_line_delim)
610 print('TEST RESULTS:')
611 print(' Scheduled tests: {}'.format(self.all_testcases))
612 print(' Executed tests: {}'.format(self[TEST_RUN]))
613 print(' Passed tests: {}'.format(
614 colorize(str(self[PASS]), GREEN)))
616 print(' Skipped tests: {}'.format(
617 colorize(str(self[SKIP]), YELLOW)))
618 if self.not_executed > 0:
619 print(' Not Executed tests: {}'.format(
620 colorize(str(self.not_executed), RED)))
622 print(' Failures: {}'.format(
623 colorize(str(self[FAIL]), RED)))
625 print(' Errors: {}'.format(
626 colorize(str(self[ERROR]), RED)))
628 if self.all_failed > 0:
629 print('FAILURES AND ERRORS IN TESTS:')
630 for result in self.results_per_suite:
631 failed_testcase_ids = result[FAIL]
632 errored_testcase_ids = result[ERROR]
633 old_testcase_name = None
634 if failed_testcase_ids or errored_testcase_ids:
635 for failed_test_id in failed_testcase_ids:
636 new_testcase_name, test_name = \
637 result.get_testcase_names(failed_test_id)
638 if new_testcase_name != old_testcase_name:
639 print(' Testcase name: {}'.format(
640 colorize(new_testcase_name, RED)))
641 old_testcase_name = new_testcase_name
642 print(' FAILURE: {} [{}]'.format(
643 colorize(test_name, RED), failed_test_id))
644 for failed_test_id in errored_testcase_ids:
645 new_testcase_name, test_name = \
646 result.get_testcase_names(failed_test_id)
647 if new_testcase_name != old_testcase_name:
648 print(' Testcase name: {}'.format(
649 colorize(new_testcase_name, RED)))
650 old_testcase_name = new_testcase_name
651 print(' ERROR: {} [{}]'.format(
652 colorize(test_name, RED), failed_test_id))
653 if self.testsuites_no_tests_run:
654 print('TESTCASES WHERE NO TESTS WERE SUCCESSFULLY EXECUTED:')
656 for testsuite in self.testsuites_no_tests_run:
657 for testcase in testsuite:
658 tc_classes.add(get_testcase_doc_name(testcase))
659 for tc_class in tc_classes:
660 print(' {}'.format(colorize(tc_class, RED)))
662 print(double_line_delim)
666 def not_executed(self):
667 return self.all_testcases - self[TEST_RUN]
670 def all_failed(self):
671 return self[FAIL] + self[ERROR]
674 def parse_results(results):
676 Prints the number of scheduled, executed, not executed, passed, failed,
677 errored and skipped tests and details about failed and errored tests.
679 Also returns all suites where any test failed.
685 results_per_suite = AllResults()
688 for result in results:
689 result_code = results_per_suite.add_result(result)
692 elif result_code == -1:
695 results_per_suite.print_results()
703 return return_code, results_per_suite.rerun
706 def parse_digit_env(env_var, default):
707 value = os.getenv(env_var, default)
712 print('WARNING: unsupported value "%s" for env var "%s",'
713 'defaulting to %s' % (value, env_var, default))
718 if __name__ == '__main__':
720 verbose = parse_digit_env("V", 0)
722 test_timeout = parse_digit_env("TIMEOUT", 600) # default = 10 minutes
724 retries = parse_digit_env("RETRIES", 0)
726 debug = os.getenv("DEBUG", "n").lower() in ["gdb", "gdbserver"]
728 debug_core = os.getenv("DEBUG", "").lower() == "core"
730 step = os.getenv("STEP", "n").lower() in ("y", "yes", "1")
732 run_interactive = debug or step
734 test_jobs = os.getenv("TEST_JOBS", "1").lower() # default = 1 process
735 if test_jobs == 'auto':
738 print('Interactive mode required, running on one core')
740 shm_free = psutil.disk_usage('/dev/shm').free
741 shm_max_processes = 1
742 if shm_free < min_req_shm:
743 raise Exception('Not enough free space in /dev/shm. Required '
744 'free space is at least %sM.'
745 % (min_req_shm >> 20))
747 extra_shm = shm_free - min_req_shm
748 shm_max_processes += extra_shm / shm_per_process
749 concurrent_tests = min(cpu_count(), shm_max_processes)
750 print('Found enough resources to run tests with %s cores'
752 elif test_jobs.isdigit():
753 concurrent_tests = int(test_jobs)
757 if run_interactive and concurrent_tests > 1:
758 raise NotImplementedError(
759 'Running tests interactively (DEBUG is gdb or gdbserver or STEP '
760 'is set) in parallel (TEST_JOBS is more than 1) is not supported')
762 parser = argparse.ArgumentParser(description="VPP unit tests")
763 parser.add_argument("-f", "--failfast", action='store_true',
764 help="fast failure flag")
765 parser.add_argument("-d", "--dir", action='append', type=str,
766 help="directory containing test files "
767 "(may be specified multiple times)")
768 args = parser.parse_args()
769 failfast = args.failfast
772 print("Running tests using custom test runner") # debug message
773 filter_file, filter_class, filter_func = parse_test_option()
775 print("Active filters: file=%s, class=%s, function=%s" % (
776 filter_file, filter_class, filter_func))
778 filter_cb = FilterByTestOption(filter_file, filter_class, filter_func)
780 ignore_path = os.getenv("VENV_PATH", None)
781 cb = SplitToSuitesCallback(filter_cb)
783 print("Adding tests from directory tree %s" % d)
784 discover_tests(d, cb, ignore_path)
786 # suites are not hashable, need to use list
789 for testcase_suite in cb.suites.values():
790 tests_amount += testcase_suite.countTestCases()
791 suites.append(testcase_suite)
793 print("%s out of %s tests match specified filters" % (
794 tests_amount, tests_amount + cb.filtered.countTestCases()))
796 if not running_extended_tests:
797 print("Not running extended tests (some tests will be skipped)")
799 attempts = retries + 1
801 print("Perform %s attempts to pass the suite..." % attempts)
803 if run_interactive and suites:
804 # don't fork if requiring interactive terminal
805 full_suite = unittest.TestSuite()
806 map(full_suite.addTests, suites)
807 result = VppTestRunner(verbosity=verbose,
809 print_summary=True).run(full_suite)
810 was_successful = result.wasSuccessful()
811 if not was_successful:
812 for test_case_info in result.failed_test_cases_info:
813 handle_failed_suite(test_case_info.logger,
814 test_case_info.tempdir,
815 test_case_info.vpp_pid)
817 test_case_info in result.core_crash_test_cases_info:
818 check_and_handle_core(test_case_info.vpp_bin_path,
819 test_case_info.tempdir,
820 test_case_info.core_crash_test)
822 sys.exit(not was_successful)
825 while suites and attempts > 0:
826 results = run_forked(suites)
827 exit_code, suites = parse_results(results)
830 print('Test run was successful')
832 print('%s attempt(s) left.' % attempts)