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[T](Protocol):
           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.


