e3.testsuite.fragment ===================== .. py:module:: e3.testsuite.fragment Classes ------- .. autoapisummary:: e3.testsuite.fragment.FragmentCallback e3.testsuite.fragment.FragmentData e3.testsuite.fragment.TestFragment e3.testsuite.fragment.ThreadTestFragment e3.testsuite.fragment.ProcessTestFragment Functions --------- .. autoapisummary:: e3.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:: index :type: int Unique integer 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:method:: static_push_error_result(uid: str, index: int, 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 index: Index 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:: 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) 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: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, 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:: fragment_index :type: int .. 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:: 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.