diff --git a/openhtf/core/monitors.py b/openhtf/core/monitors.py index bfedf0a3e..7f6b9510d 100644 --- a/openhtf/core/monitors.py +++ b/openhtf/core/monitors.py @@ -71,6 +71,11 @@ def __init__(self, measurement_name: Text, self.measurement_name = measurement_name self.monitor_desc = monitor_desc self.test_state = test_state + # Capture the monitored phase's PhaseState on the phase thread so this + # thread can bind to it. TestState tracks the running phase per thread, so + # without this the monitor thread would have no running phase when other + # phases execute concurrently. + self._phase_state = test_state.running_phase_state self.interval_ms = interval_ms self.extra_kwargs = extra_kwargs @@ -91,6 +96,7 @@ def get_value(self) -> Any: return self.monitor_desc.with_args(**kwargs)(self.test_state) def _thread_proc(self): + self.test_state.running_phase_state = self._phase_state measurement = getattr(self.test_state.test_api.measurements, self.measurement_name) start_time = time.time() diff --git a/openhtf/core/phase_executor.py b/openhtf/core/phase_executor.py index 029a6768f..cdf84a969 100644 --- a/openhtf/core/phase_executor.py +++ b/openhtf/core/phase_executor.py @@ -36,7 +36,7 @@ import time import traceback import types -from typing import Any, Dict, Optional, Text, Tuple, Type, TYPE_CHECKING, Union +from typing import Any, Dict, Optional, Set, TYPE_CHECKING, Text, Tuple, Type, Union import attr from openhtf import util @@ -170,10 +170,12 @@ def __init__(self, phase_desc: phase_descriptor.PhaseDescriptor, self._phase_desc = phase_desc self._test_state = test_state self._subtest_rec = subtest_rec + self._phase_state = test_state.running_phase_state self._phase_execution_outcome = None # type: Optional[PhaseExecutionOutcome] def _thread_proc(self) -> None: """Execute the encompassed phase and save the result.""" + self._test_state.running_phase_state = self._phase_state # Call the phase, save the return value, or default it to CONTINUE. phase_return = self._phase_desc(self._test_state) if phase_return is None: @@ -239,11 +241,19 @@ def __init__(self, test_state: 'htf_test_state.TestState'): # _execute_phase_once is setting up the next phase thread. self._current_phase_thread_lock = threading.Lock() self._current_phase_thread = None # type: Optional[PhaseExecutorThread] + self._active_phase_threads = set() # type: Set[PhaseExecutorThread] self._stopping = threading.Event() - def _should_repeat(self, phase: phase_descriptor.PhaseDescriptor, - phase_execution_outcome: PhaseExecutionOutcome, is_last_repeat: bool) -> bool: + def _should_repeat( + self, + phase: phase_descriptor.PhaseDescriptor, + phase_execution_outcome: PhaseExecutionOutcome, + is_last_repeat: bool, + ) -> bool: """Returns whether a phase should be repeated.""" + # Never repeat a phase that was killed due to stop or abort. + if phase_execution_outcome.is_aborted: + return False if phase_execution_outcome.is_timeout and phase.options.repeat_on_timeout: return True elif phase_execution_outcome.is_repeat: @@ -251,10 +261,16 @@ def _should_repeat(self, phase: phase_descriptor.PhaseDescriptor, elif phase.options.force_repeat: return True elif phase.options.repeat_on_measurement_fail: - last_phase_outcome = self.test_state.test_record.phases[-1].outcome - last_repeat_failed = last_phase_outcome == test_record.PhaseOutcome.FAIL + # Set by running_phase_context on this thread; only None if the phase + # was short-circuited (e.g. by run_if) before any phase ran here. + last_phase_rec = self.test_state.last_phase_record + if last_phase_rec is None: + return False + last_repeat_failed = ( + last_phase_rec.outcome == test_record.PhaseOutcome.FAIL + ) if last_repeat_failed and not is_last_repeat: - self.test_state.test_record.phases[-1].outcome = test_record.PhaseOutcome.SKIP + last_phase_rec.outcome = test_record.PhaseOutcome.SKIP return last_repeat_failed return False @@ -320,7 +336,6 @@ def _execute_phase_once( phase_desc.name) return PhaseExecutionOutcome(phase_descriptor.PhaseResult.SKIP), None - override_result = None with self.test_state.running_phase_context(phase_desc) as phase_state: if subtest_rec: @@ -347,6 +362,7 @@ def _execute_phase_once( run_with_profiling, subtest_rec) phase_thread.start() self._current_phase_thread = phase_thread + self._active_phase_threads.add(phase_thread) phase_state.result = phase_thread.join_or_die() if phase_state.result.is_repeat and is_last_repeat: # pyrefly: ignore[missing-attribute] @@ -354,7 +370,10 @@ def _execute_phase_once( phase_state.hit_repeat_limit = True override_result = PhaseExecutionOutcome( phase_descriptor.PhaseResult.STOP) - self._current_phase_thread = None + with self._current_phase_thread_lock: + self._active_phase_threads.discard(phase_thread) + if self._current_phase_thread == phase_thread: + self._current_phase_thread = None # Refresh the result in case a validation for a partially set measurement # or phase diagnoser raised an exception. @@ -433,18 +452,23 @@ def stop( """ self._stopping.set() with self._current_phase_thread_lock: - phase_thread = self._current_phase_thread - if not phase_thread: - return - - if phase_thread.is_alive(): - phase_thread.kill() - - self.logger.debug('Waiting for cancelled phase to exit: %s', phase_thread) - timeout = timeouts.PolledTimeout.from_seconds(timeout_s) - while phase_thread.is_alive() and not timeout.has_expired(): - time.sleep(0.1) - self.logger.debug('Cancelled phase %s exit', - "didn't" if phase_thread.is_alive() else 'did') + threads_to_kill = list(self._active_phase_threads) + + timeout = timeouts.PolledTimeout.from_seconds(timeout_s) + for phase_thread in threads_to_kill: + if phase_thread.is_alive(): + phase_thread.kill() + + for phase_thread in threads_to_kill: + if phase_thread.is_alive(): + self.logger.debug( + 'Waiting for cancelled phase to exit: %s', phase_thread + ) + while phase_thread.is_alive() and not timeout.has_expired(): + time.sleep(0.1) + self.logger.debug( + 'Cancelled phase %s exit', + "didn't" if phase_thread.is_alive() else 'did', + ) # Clear the currently running phase, whether it finished or timed out. self.test_state.stop_running_phase() diff --git a/openhtf/core/test_descriptor.py b/openhtf/core/test_descriptor.py index 3ab89036c..5dbcbf6d4 100644 --- a/openhtf/core/test_descriptor.py +++ b/openhtf/core/test_descriptor.py @@ -20,6 +20,7 @@ import argparse import collections +import copy import logging import os import sys @@ -589,6 +590,14 @@ def get_measurement( Returns: an ImmutableMeasurement or None if the measurement cannot be found. """ + # Check the phase this TestApi is bound to first. This keeps lookups + # correct from threads started by the phase (which have no running phase of + # their own) while other phases execute concurrently. + current_measurements = self._running_phase_state.measurements + if measurement_name in current_measurements: + return htf_measurements.ImmutableMeasurement.from_measurement( + current_measurements[measurement_name] + ) return self._running_test_state.get_measurement(measurement_name) def get_measurement_strict( @@ -608,7 +617,7 @@ def get_measurement_strict( Raises: MeasurementNotFoundError: Thrown when the test measurement is not found. """ - measurement = self._running_test_state.get_measurement(measurement_name) + measurement = self.get_measurement(measurement_name) if measurement is None: raise MeasurementNotFoundError( f'Failed to find test measurement {measurement_name}') @@ -627,6 +636,10 @@ def get_attachment( Returns: A copy of the attachment or None if the attachment cannot be found. """ + # See get_measurement() for why the bound phase is checked first. + current_attachments = self._running_phase_state.phase_record.attachments + if attachment_name in current_attachments: + return copy.deepcopy(current_attachments[attachment_name]) return self._running_test_state.get_attachment(attachment_name) def get_attachment_strict( diff --git a/openhtf/core/test_state.py b/openhtf/core/test_state.py index a428acee6..cc9792da4 100644 --- a/openhtf/core/test_state.py +++ b/openhtf/core/test_state.py @@ -33,6 +33,7 @@ import os import socket import sys +import threading from typing import Any, Dict, Iterator, List, Optional, Set, TYPE_CHECKING, Text, Tuple, Union import attr @@ -94,6 +95,44 @@ class InternalError(Exception): """An internal error.""" +class _TestStateThreadLocal(threading.local): + """Per-thread state for a TestState. + + Subclassing threading.local ensures __init__ runs once per thread on first + access, so every attribute always has a default value in each thread. + + Why this exists: a single TestState is shared by every phase thread, and + much of the API (TestState.running_phase_state, TestState.test_api, + TestState.logger, TestState.last_phase_record, and callers in + phase_executor/test_executor/TestApi) reaches phase-specific data through + the TestState rather than through the PhaseState that owns it. Before phases + could run concurrently there was at most one running phase, so storing these + values directly on TestState was sufficient. With concurrent phases, each + thread needs its own view of "the current phase", so these values are kept + per-thread here to preserve the existing TestState-centric API. + + This is a compatibility shim: it hides a design issue where TestState is + passed around in places where the PhaseState (or a phase-scoped TestApi) + is the object that should actually be passed. Ideally, phase-scoped data + would be threaded explicitly through PhaseState, and this class removed. + + TODO: Refactor phase-scoped APIs to take PhaseState explicitly + and remove this thread-local state. + + Attributes: + phase_state: PhaseState for the phase running in the current thread, if any. + last_phase_record: PhaseRecord of the last phase run in the current thread, + if any. + test_api: Cached TestApi for the phase running in the current thread. + """ + + def __init__(self): + super().__init__() + self.phase_state: Optional['PhaseState'] = None + self.last_phase_record: Optional[test_record.PhaseRecord] = None + self.test_api: Optional['test_descriptor.TestApi'] = None + + class TestState(util.SubscribableStateMixin): """This class handles tracking the state of a running Test. @@ -118,6 +157,9 @@ class TestState(util.SubscribableStateMixin): internal-only structures from being accidentally modified. Note that if there is no running phase, test_api is also None. execution_uid: A UUID that is specific to this execution. + state_lock: Reentrant lock (threading.RLock) guarding state shared across + concurrently running phase threads, such as the list of active + PhaseStates, test_record.phases, and phase record attachments. """ class Status(enum.Enum): @@ -137,6 +179,17 @@ def __init__(self, test_desc: 'test_descriptor.TestDescriptor', test_options: test_options passed through from Test. """ super(TestState, self).__init__() + # Reentrant lock guarding state shared across concurrently running phase + # threads: + # - self._active_phase_states: the list of currently running PhaseStates. + # - self.test_record.phases: the list of completed PhaseRecords, appended + # to when a phase finishes and read by get_attachment/get_measurement. + # - PhaseState.phase_record.attachments: attachments added via + # PhaseState.attach(). + # Thread-local state (self._thread_local) is not guarded by this lock. + self.state_lock = threading.RLock() + self._thread_local = _TestStateThreadLocal() + self._active_phase_states = list() self._status = self.Status.WAITING_FOR_TEST_START # type: TestState.Status self.test_record = test_record.TestRecord( @@ -155,7 +208,6 @@ def __init__(self, test_desc: 'test_descriptor.TestDescriptor', self.diagnoses_manager = diagnoses_lib.DiagnosesManager( self.state_logger.getChild('diagnoses')) self.running_phase_state = None # type: Optional['PhaseState'] - self._running_test_api = None # type: Optional['test_descriptor.TestApi'] # TODO(arsharma): Change to Dict[Any, Any] when pytype handles it correctly. self.user_defined_state = {} # type: Any self.execution_uid = execution_uid @@ -180,6 +232,32 @@ def logger(self) -> logging.Logger: 'Calling `logger` attribute while phase not running; use state_logger ' 'instead.') + @property + def running_phase_state(self): + state = self._thread_local.phase_state + if state is not None: + return state + + with self.state_lock: + if not self._active_phase_states or len(self._active_phase_states) > 1: + return None + return self._active_phase_states[0] + + @property + def active_phases(self) -> List['PhaseState']: + """Get all currently active phase states.""" + with self.state_lock: + return list(self._active_phase_states) + + @running_phase_state.setter + def running_phase_state(self, value): + self._thread_local.phase_state = value + + @property + def last_phase_record(self) -> Optional[test_record.PhaseRecord]: + """PhaseRecord of the last phase run in the current thread, if any.""" + return self._thread_local.last_phase_record + @property def test_api(self) -> 'test_descriptor.TestApi': """Create a TestApi for access to this TestState. @@ -195,15 +273,28 @@ def test_api(self) -> 'test_descriptor.TestApi': openhtf.TestApi """ if not self.running_phase_state: + if len(self.active_phases) > 1: + raise ValueError( + 'test_api is ambiguous: this thread is not bound to a phase and ' + 'multiple phases are running concurrently. Pass the TestApi (or ' + 'PhaseState) from the owning phase to threads it starts, or set ' + 'TestState.running_phase_state on the new thread.' + ) raise ValueError('test_api only available when phase is running.') - if not self._running_test_api: - self._running_test_api = openhtf.TestApi( + api = self._thread_local.test_api + if ( + api is None + or api._running_phase_state != self.running_phase_state # pylint: disable=protected-access + ): + api = openhtf.TestApi( # pyrefly: ignore[missing-argument] measurements=measurements.Collection( - self.running_phase_state.measurements), + self.running_phase_state.measurements + ), running_phase_state=self.running_phase_state, running_test_state=self, ) - return self._running_test_api + self._thread_local.test_api = api + return api def get_attachment(self, attachment_name: Text) -> Optional[test_record.Attachment]: @@ -222,10 +313,11 @@ def get_attachment(self, attachment_name) return copy.deepcopy(attachment) - for phase_record in self.test_record.phases: - if attachment_name in phase_record.attachments: - attachment = phase_record.attachments[attachment_name] - return copy.deepcopy(attachment) + with self.state_lock: + for phase_record in self.test_record.phases: + if attachment_name in phase_record.attachments: + attachment = phase_record.attachments[attachment_name] + return copy.deepcopy(attachment) self.state_logger.warning('Could not find attachment: %s', attachment_name) return None @@ -256,11 +348,14 @@ def get_measurement( # Iterate through phases in reversed order to return most recent (necessary # because measurement and phase names are not necessarily unique) - for phase_record in reversed(self.test_record.phases): - if (phase_record.result not in ignore_outcomes and - measurement_name in phase_record.measurements): - measurement = phase_record.measurements[measurement_name] - return measurements.ImmutableMeasurement.from_measurement(measurement) + with self.state_lock: + for phase_record in reversed(self.test_record.phases): + if ( + phase_record.result not in ignore_outcomes + and measurement_name in phase_record.measurements + ): + measurement = phase_record.measurements[measurement_name] + return measurements.ImmutableMeasurement.from_measurement(measurement) self.state_logger.warning('Could not find measurement: %s', measurement_name) @@ -285,30 +380,50 @@ def running_phase_context( Yields: PhaseState to track transient state. """ - assert not self.running_phase_state, 'Phase already running!' + assert not self._thread_local.phase_state, 'Phase already running!' phase_logger = self.state_logger.getChild('phase.' + phase_desc.name) - phase_state = self.running_phase_state = PhaseState.from_descriptor( - phase_desc, self, phase_logger) - self.notify_update() # New phase started. + with self.state_lock: + phase_state = PhaseState.from_descriptor(phase_desc, self, phase_logger) + self.running_phase_state = phase_state + self._active_phase_states.append(phase_state) + self.notify_update() # New phase started. try: yield phase_state finally: - phase_state.finalize() - self.test_record.add_phase_record(phase_state.phase_record) + with self.state_lock: + phase_state.finalize() + self._thread_local.last_phase_record = phase_state.phase_record + self.test_record.add_phase_record(phase_state.phase_record) + if phase_state in self._active_phase_states: + self._active_phase_states.remove(phase_state) + self.notify_update() # Phase finished. self.running_phase_state = None - self._running_test_api = None - self.notify_update() # Phase finished. def as_base_types(self) -> Dict[Text, Any]: """Convert to a dict representation composed exclusively of base types.""" - running_phase_state = None - if self.running_phase_state: - running_phase_state = self.running_phase_state.as_base_types() + with self.state_lock: + running_phase_states = [ + phase.as_base_types() for phase in self._active_phase_states + ] + + if running_phase_states: + running_phase_output = running_phase_states[0] + elif self.running_phase_state: + # Includes manually/legacy set PhaseState. + legacy_basetype = self.running_phase_state.as_base_types() + running_phase_output = legacy_basetype + running_phase_states = [legacy_basetype] + else: + running_phase_output = None + return { 'status': data.convert_to_base_types(self._status), 'test_record': self.test_record.as_base_types(), 'plugs': self.plug_manager.as_base_types(), - 'running_phase_state': running_phase_state, + # TODO(wuchiju): Remove running_phase_state once WebUI migrates to + # running_phase_states. + 'running_phase_state': running_phase_output, + 'running_phase_states': running_phase_states, } def _asdict(self) -> Dict[Text, Any]: @@ -321,6 +436,8 @@ def is_finalized(self) -> bool: def stop_running_phase(self) -> None: """Stops the currently running phase, allowing another phase to run.""" + with self.state_lock: + self._active_phase_states.clear() self.running_phase_state = None @property @@ -473,6 +590,10 @@ def finalize_normally(self) -> None: if not phases: # Vacuously PASS a TestRecord with no phases. self._finalize(test_record.Outcome.PASS) + elif any( + phase.outcome == test_record.PhaseOutcome.ERROR for phase in phases + ): + self._finalize(test_record.Outcome.ERROR) elif any( phase.outcome == test_record.PhaseOutcome.FAIL for phase in phases ): @@ -696,18 +817,20 @@ def attach(self, DuplicateAttachmentError: Raised if there is already an attachment with the given name. """ - if name in self.phase_record.attachments: - raise DuplicateAttachmentError('Duplicate attachment for %s' % name) + with self.test_state.state_lock: + if name in self.phase_record.attachments: + raise DuplicateAttachmentError('Duplicate attachment for %s' % name) + + if mimetype is INFER_MIMETYPE: + mimetype = mimetypes.guess_type(name)[0] + elif mimetype is not None and not mimetypes.guess_extension(mimetype): + self.logger.warning( + 'Unrecognized MIME type: "%s" for attachment "%s"', mimetype, name + ) - if mimetype is INFER_MIMETYPE: - mimetype = mimetypes.guess_type(name)[0] - elif mimetype is not None and not mimetypes.guess_extension(mimetype): - self.logger.warning('Unrecognized MIME type: "%s" for attachment "%s"', - mimetype, name) - - attach_record = test_record.Attachment(binary_data, mimetype) # pyrefly: ignore[bad-argument-type] - self.phase_record.attachments[name] = attach_record - self._cached['attachments'][name] = attach_record._asdict() + attach_record = test_record.Attachment(binary_data, mimetype) # pyrefly: ignore[bad-argument-type] + self.phase_record.attachments[name] = attach_record + self._cached['attachments'][name] = attach_record._asdict() def attach_from_file(self, filename: Text, diff --git a/openhtf/util/test.py b/openhtf/util/test.py index 22dec4cb0..598f33e2a 100644 --- a/openhtf/util/test.py +++ b/openhtf/util/test.py @@ -290,8 +290,10 @@ class FakeTestApi(test_descriptor.TestApi): def __init__(self): self.mock_logger = mock.create_autospec(logging.Logger) + # PhaseState attributes are attrs instance fields, so autospec of the class + # does not expose them; provide the ones TestApi reads directly. self.mock_phase_state = mock.create_autospec( - test_state.PhaseState, logger=self.mock_logger) + test_state.PhaseState, logger=self.mock_logger, measurements={}) self.mock_test_state = mock.create_autospec( test_state.TestState, test_record=test_record.TestRecord('DUT', 'STATION'), diff --git a/openhtf/util/threads.py b/openhtf/util/threads.py index 16d2b9ac9..a172f29dd 100644 --- a/openhtf/util/threads.py +++ b/openhtf/util/threads.py @@ -22,6 +22,7 @@ import threading _LOG = logging.getLogger(__name__) +_PROFILE_LOCK = threading.Lock() class ThreadTerminationError(SystemExit): @@ -97,6 +98,7 @@ def __init__(self, *args, logger: logging.Logger = _LOG, **kwargs): self._profiler = cProfile.Profile() else: self._profiler = None + self._profiler_active = False self._logger = logger def run(self): @@ -104,8 +106,17 @@ def run(self): with self._running_lock: if self._killed.is_set(): raise ThreadTerminationError() - if self._profiler is not None: - self._profiler.enable() + # If parallel threads are running, only start profiling in one of them. + # This is correct behavior for Python 3.12+ which changed how profiling + # works across threads. + if self._profiler is not None and _PROFILE_LOCK.acquire(False): + try: + self._profiler.enable() + except ValueError: + _PROFILE_LOCK.release() + self._profiler = None + else: + self._profiler_active = True self._thread_proc() except Exception: # pylint: disable=broad-except if not self._thread_exception(*sys.exc_info()): @@ -114,13 +125,16 @@ def run(self): finally: self._thread_finished() self._logger.debug('Thread finished: %s', self.name) - if self._profiler is not None: + if self._profiler is not None and self._profiler_active: self._profiler.disable() + _PROFILE_LOCK.release() - def get_profile_stats(self) -> pstats.Stats: - """Returns profile_stats from profiler. Raises if profiling not enabled.""" - if self._profiler is not None: + def get_profile_stats(self) -> pstats.Stats | None: + """Returns profile_stats from profiler, or None if not active.""" + if self._profiler is not None and self._profiler_active: return pstats.Stats(self._profiler) + if self._run_with_profiling: + return None raise InvalidUsageError( 'Profiling not enabled via __init__, or thread has not run yet.') diff --git a/test/core/monitors_test.py b/test/core/monitors_test.py index b97166761..6042e8386 100644 --- a/test/core/monitors_test.py +++ b/test/core/monitors_test.py @@ -13,6 +13,7 @@ # limitations under the License. import queue +import threading import time import unittest from unittest import mock @@ -20,7 +21,11 @@ from openhtf import plugs from openhtf.core import base_plugs from openhtf.core import monitors +from openhtf.core import phase_collections from openhtf.core import phase_descriptor +from openhtf.core import test_descriptor +from openhtf.core import test_record +from openhtf.core import test_state class EmptyPlug(base_plugs.BasePlug): @@ -113,3 +118,69 @@ def monitor_func(test): 1000, ) self.assertEqual(1, monitor_thread.get_value()) + + +class TestMonitorsWithConcurrentPhases(unittest.TestCase): + """Monitors must keep working while other phases run on other threads.""" + + def setUp(self): + super().setUp() + self.test_state = test_state.TestState( + test_descriptor.TestDescriptor( + phase_collections.PhaseSequence(()), + test_record.CodeInfo.uncaptured(), + {}, + ), + '01234567890', + test_descriptor.TestOptions(), + ) + + def test_monitor_samples_while_another_phase_is_running(self): + samples = queue.Queue() + + def monitor_func(test): + del test # Unused. + samples.put(1) + return 1 + + @monitors.monitors('meas', monitor_func, poll_interval_ms=50) + def monitored_phase(test): + del test # Unused. + # Wait for a couple of samples, but never hang the test if the monitor + # thread died. + deadline = time.monotonic() + 5 + while samples.qsize() < 2 and time.monotonic() < deadline: + time.sleep(0.05) + + @phase_descriptor.PhaseOptions(name='other_phase') + def other_phase(): + pass + + # Keep a second phase running on another thread for the whole test, so the + # TestState has more than one active phase and no thread-independent + # "current" phase. + entered = threading.Event() + release = threading.Event() + + def _hold_other_phase(): + with self.test_state.running_phase_context(other_phase): + entered.set() + release.wait() + + holder = threading.Thread(target=_hold_other_phase) + holder.start() + self.assertTrue(entered.wait(timeout=5)) + try: + with self.test_state.running_phase_context( + monitored_phase + ) as phase_state: + monitored_phase(self.test_state) + finally: + release.set() + holder.join(timeout=5) + + self.assertGreaterEqual(samples.qsize(), 2, 'Monitor thread did not run.') + measured = phase_state.measurements['meas'].measured_value + self.assertTrue(measured.is_value_set) + self.assertGreaterEqual(len(measured.value_dict), 1) + self.assertEqual({1}, set(measured.value_dict.values())) diff --git a/test/core/phase_executor_test.py b/test/core/phase_executor_test.py index 9f2b2c351..47f19436e 100644 --- a/test/core/phase_executor_test.py +++ b/test/core/phase_executor_test.py @@ -56,6 +56,19 @@ def test_execute_phase_with_repeat_limit_max_exceeds_default_limit(self): expected_call_count=phase_descriptor.DEFAULT_REPEAT_LIMIT + 1, ) + def test_aborted_phase_does_not_repeat_with_force_repeat(self): + call_count = 0 + + @openhtf.PhaseOptions(force_repeat=True, repeat_limit=3) + def aborted_phase(): + nonlocal call_count + call_count += 1 + raise openhtf.util.threads.ThreadTerminationError() + + test = openhtf.Test(aborted_phase) + test.execute() + self.assertEqual(call_count, 1) + class PhaseExecuterRunIfTest(htf_test.TestCase): diff --git a/test/test_state_test.py b/test/test_state_test.py index 576c1d591..a36e91932 100644 --- a/test/test_state_test.py +++ b/test/test_state_test.py @@ -12,10 +12,12 @@ # See the License for the specific language governing permissions and # limitations under the License. +import contextlib import copy import logging import sys import tempfile +import threading from unittest import mock from absl.testing import parameterized @@ -38,6 +40,11 @@ def test_phase(): """Test docstring.""" +@openhtf.PhaseOptions(name='other_phase') +def other_phase(): + """A second phase, used to simulate a concurrently running phase.""" + + PHASE_STATE_BASE_TYPE_INITIAL = { 'name': 'test_phase', 'codeinfo': { @@ -86,9 +93,7 @@ def test_phase(): 'outcome': None, 'outcome_details': [], 'marginal': None, - 'metadata': { - 'config': {} - }, + 'metadata': {'config': {}}, 'phases': [], 'subtests': [], 'branches': [], @@ -101,6 +106,7 @@ def test_phase(): 'plug_states': {}, }, 'running_phase_state': copy.deepcopy(PHASE_STATE_BASE_TYPE_INITIAL), + 'running_phase_states': [copy.deepcopy(PHASE_STATE_BASE_TYPE_INITIAL)], } @@ -220,6 +226,9 @@ def test_test_state_cache(self): descriptor_id = basetypes['running_phase_state']['descriptor_id'] expected_initial_basetypes['running_phase_state']['descriptor_id'] = ( descriptor_id) + expected_initial_basetypes['running_phase_states'][0][ + 'descriptor_id' + ] = descriptor_id self.assertEqual(expected_initial_basetypes, basetypes) self.running_phase_state._finalize_measurements() self.test_record.add_phase_record(self.running_phase_state.phase_record) @@ -253,3 +262,155 @@ def test_test_state_finalize_from_phase_outcome_exception_info(self): self.test_state.finalize_from_phase_outcome(phase_exe_outcome, 'MyPhase') self.assertEqual(self.test_state.test_record.outcome, test_record.Outcome.ERROR) + + +class TestTestStateConcurrentPhases(parameterized.TestCase): + """Tests TestState behavior when phases run on multiple threads. + + TestState keeps the running phase per thread. These tests exercise what a + thread that is *not* bound to a phase (e.g. a monitor thread, or a worker + thread started by a phase) observes while zero, one, or several phases are + running. + """ + + def setUp(self): + super().setUp() + self.test_state = test_state.TestState( + test_descriptor.TestDescriptor( + phase_collections.PhaseSequence((test_phase, other_phase)), + test_record.CodeInfo.uncaptured(), + {'config': {}}, + ), + 'testing-123', + test_descriptor.TestOptions(), + ) + + @contextlib.contextmanager + def _hold_phase_open(self, phase_desc): + """Enters running_phase_context for phase_desc on a dedicated thread. + + Yields a dict with the thread's PhaseState and TestApi. The phase stays + running until the context exits. + """ + entered = threading.Event() + release = threading.Event() + box = {} + + def _run(): + with self.test_state.running_phase_context(phase_desc) as phase_state: + box['phase_state'] = phase_state + box['test_api'] = self.test_state.test_api + entered.set() + release.wait() + + thread = threading.Thread(target=_run, name=f'hold-{phase_desc.name}') + thread.start() + self.assertTrue(entered.wait(timeout=5), 'Phase thread did not start.') + try: + yield box + finally: + release.set() + thread.join(timeout=5) + self.assertFalse(thread.is_alive(), 'Phase thread did not exit.') + + def _call_on_unbound_thread(self, func): + """Calls func on a fresh thread that is not bound to any phase.""" + box = {} + + def _run(): + try: + box['result'] = func() + except Exception as e: # pylint: disable=broad-except + box['error'] = e + + thread = threading.Thread(target=_run, name='unbound') + thread.start() + thread.join(timeout=5) + self.assertFalse(thread.is_alive(), 'Unbound thread did not exit.') + if 'error' in box: + raise box['error'] + return box['result'] + + def test_running_phase_state_from_unbound_thread(self): + get_running = lambda: self.test_state.running_phase_state + + # No phases running: nothing to fall back to. + self.assertIsNone(self._call_on_unbound_thread(get_running)) + + with self._hold_phase_open(test_phase) as first: + # Exactly one phase running: unbound threads see it. This is the + # sequential-execution case and must keep working for existing callers. + self.assertIs( + first['phase_state'], self._call_on_unbound_thread(get_running) + ) + + with self._hold_phase_open(other_phase): + # Multiple phases running: there is no single "current" phase. + self.assertIsNone(self._call_on_unbound_thread(get_running)) + self.assertLen(self.test_state.active_phases, 2) + + self.assertIs( + first['phase_state'], self._call_on_unbound_thread(get_running) + ) + + self.assertIsNone(self._call_on_unbound_thread(get_running)) + + def test_test_api_from_unbound_thread_with_multiple_phases_raises(self): + with self._hold_phase_open(test_phase), self._hold_phase_open(other_phase): + with self.assertRaisesRegex(ValueError, 'ambiguous'): + self._call_on_unbound_thread(lambda: self.test_state.test_api) + + def test_test_api_get_measurement_from_unbound_thread(self): + with self._hold_phase_open(test_phase) as first, self._hold_phase_open( + other_phase + ): + api = first['test_api'] + api.measurements['test_measurement'] = 5 + + measurement = self._call_on_unbound_thread( + lambda: api.get_measurement('test_measurement') + ) + self.assertIsNotNone(measurement) + self.assertEqual(5, measurement.value) + self.assertEqual( + 5, + self._call_on_unbound_thread( + lambda: api.get_measurement_strict('test_measurement') + ).value, + ) + + def test_test_api_get_attachment_from_unbound_thread(self): + with self._hold_phase_open(test_phase) as first, self._hold_phase_open( + other_phase + ): + api = first['test_api'] + api.attach('attachment.txt', b'data', 'text/plain') + + attachment = self._call_on_unbound_thread( + lambda: api.get_attachment('attachment.txt') + ) + self.assertIsNotNone(attachment) + self.assertEqual(b'data', attachment.data) + self.assertEqual( + b'data', + self._call_on_unbound_thread( + lambda: api.get_attachment_strict('attachment.txt') + ).data, + ) + + def test_test_api_get_measurement_falls_back_to_previous_phases(self): + # Run and finish a phase that records a measurement. + with self.test_state.running_phase_context(test_phase): + self.test_state.test_api.measurements['test_measurement'] = 7 + + # A later phase's TestApi still finds it, even from an unbound thread while + # another phase is running. + with self._hold_phase_open(other_phase) as later, self._hold_phase_open( + test_phase + ): + api = later['test_api'] + measurement = self._call_on_unbound_thread( + lambda: api.get_measurement('test_measurement') + ) + self.assertIsNotNone(measurement) + self.assertEqual(7, measurement.value)