testsuite.fragment ================== .. py:module:: testsuite.fragment Classes ------- .. autoapisummary:: testsuite.fragment.FragmentCallback testsuite.fragment.FragmentData testsuite.fragment.TestFragment testsuite.fragment.ThreadTestFragment testsuite.fragment.ProcessTestFragment Functions --------- .. autoapisummary:: testsuite.fragment.run_fragment Module Contents --------------- .. py:class:: FragmentCallback Bases: :py:obj:`Protocol` Base class for protocol classes. Protocol classes are defined as:: class Proto(Protocol): def meth(self) -> int: ... Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing). For example:: class C: def meth(self) -> int: return 0 def func(x: Proto) -> int: return x.meth() func(C()) # Passes static type check See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:: class GenProto(Protocol[T]): def meth(self) -> T: ... .. py:method:: __call__(previous_values: Dict[str, Any], slot: int) -> None .. py:class:: FragmentData Data for a job unit in the testsuite. Each ``FragmentData`` instance is recorded in the testsuite global DAG to control the order of execution of all fragments with the requested level of parallelism. Note that the job scheduler turns ``FragmentData`` instances into ``TestFragment`` ones during the execution (see ``Testsuite.job_factory`` callback). .. py:attribute:: uid :type: str .. py:attribute:: driver :type: e3.testsuite.driver.TestDriver .. py:attribute:: name :type: str .. py:attribute:: callback :type: FragmentCallback .. py:attribute:: callback_by_name :type: bool Whether ``callback`` is just the ``name`` method of ``driver``. .. py:method:: matches(driver_cls: Type[e3.testsuite.driver.TestDriver], name: str) -> bool Return whether this fragment matches the given name/test driver. If ``name`` is left to None, just check the driver type. .. py:method:: clear_driver_data() -> None Remove references to ``TestDriver`` instances and related data. Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data. .. py:class:: TestFragment Base class for testcase scheduling units. .. py:attribute:: uid :type: str Unique string identifier for this test fragment. .. py:attribute:: driver :type: e3.testsuite.driver.TestDriver Test driver that is responsible for this test fragment. .. py:attribute:: running_status :type: e3.testsuite.running_status.RunningStatus RunningStatus instance to signal when job starts/completes. .. py:attribute:: result_queue :type: e3.testsuite.driver.ResultQueue List of test results that this fragments plans to integrate to the testsuite report. .. py:attribute:: event_notifier :type: e3.testsuite.event_notifications.EventNotifier Event notifier for test start/end events. .. py:attribute:: started_test :type: bool Set when the fragment starts: whether this fragment was the first to start for its owning test driver. .. py:attribute:: ended_test :type: bool Set when the fragment completes: whether this fragment was the last to complete for its owning test driver. .. py:method:: maybe_notify_started() -> None Send a test start notification when appropriate. If this fragment was the first one to start for its test driver, send the corresponding notification. .. py:method:: maybe_notify_ended() -> None Send a test end notification when appropriate. If this fragment was the last one to complete for its test driver, send the corresponding notification. .. py:method:: static_push_error_result(uid: str, driver: e3.testsuite.driver.TestDriver) -> None :staticmethod: Generate a test result to log the exception and traceback. This helper method is meant to be used when the execution of the test fragments aborts because of an uncaught exception. We must report a test error, and we provide exception information for post-mortem investigation. :param uid: UID for the test fragment. :param driver: TestDriver for the test fragment. .. py:method:: push_error_result(exc: Exception) -> None Shortcut for static_push_error_result on the current fragment. .. py:method:: must_run() -> bool Return if the fragment must be run. It must be skipped when the testsuite has aborted because of too many consecutive failures: we push a SKIP result in that case. .. py:method:: clear_driver_data() -> None :abstractmethod: Remove references to ``TestDriver`` instances and related data. Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data. .. py:class:: ThreadTestFragment(uid: str, driver: e3.testsuite.driver.TestDriver, callback: FragmentCallback, previous_values: Dict[str, Any], notify_end: Callable[[str], None], running_status: e3.testsuite.running_status.RunningStatus, event_notifier: e3.testsuite.event_notifications.EventNotifier) Bases: :py:obj:`e3.job.Job`, :py:obj:`TestFragment` Run a test fragment in a thread. .. py:attribute:: driver Test driver that is responsible for this test fragment. .. py:attribute:: previous_values .. py:attribute:: running_status RunningStatus instance to signal when job starts/completes. .. py:attribute:: result_queue :type: e3.testsuite.driver.ResultQueue List of test results that this fragments plans to integrate to the testsuite report. .. py:attribute:: event_notifier Event notifier for test start/end events. .. py:method:: clear_driver_data() -> None Remove references to ``TestDriver`` instances and related data. Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data. .. py:method:: run() -> None Run the test fragment. .. py:class:: ProcessTestFragment(uid: str, driver: e3.testsuite.driver.TestDriver, callback_name: str, slot: int, running_status: e3.testsuite.running_status.RunningStatus, event_notifier: e3.testsuite.event_notifications.EventNotifier, env: e3.env.Env) Bases: :py:obj:`e3.testsuite.multiprocess_scheduler.Worker`, :py:obj:`TestFragment` Run a test fragment in a separate process. .. py:class:: Input Subprocess input data. .. py:attribute:: fragment_uid :type: str .. py:attribute:: driver_cls :type: Type[e3.testsuite.driver.TestDriver] .. py:attribute:: test_env :type: Dict[str, Any] .. py:attribute:: callback_name :type: str .. py:attribute:: slot :type: int .. py:class:: Output Subprocess output data. .. py:attribute:: result_queue :type: e3.testsuite.driver.ResultQueue .. py:attribute:: exchange_file :type: str Name of the file to exchange data with the subprocess. .. py:attribute:: running_status RunningStatus instance to signal when job starts/completes. .. py:attribute:: event_notifier Event notifier for test start/end events. .. py:attribute:: result_queue :value: [] List of test results that this fragments plans to integrate to the testsuite report. .. py:method:: clear_driver_data() -> None Remove references to ``TestDriver`` instances and related data. Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data. .. py:method:: start() -> e3.os.process.Run Create and return the subprocess to do the work. All subclasses must override this. .. py:method:: collect_result() -> None .. py:method:: extract_result_queue() -> None Read the result queue from the exchange file. Try to extract the result queue from the exchange file and put results in the driver's result queue. If anything goes sour, create an error result in the same result queue. .. py:function:: run_fragment(argv: Optional[List[str]] = None) -> None Run a test fragment. This function is meant to be the entry point of a standalone script, to run a fragment in a subprocess, separate from the main testsuite process.