CW-10: preserve Python 3.14 timeout behavior and improve test diagnostics

.github/workflows/test.yml

@@ -6,13 +6,13 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest]
-        python-version: [3.6, 3.7, 3.8]
+        os: [ubuntu-22.04]
+        python-version: ['3.8', '3.10', '3.11', '3.14.6']
     runs-on: ${{ matrix.os }}
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python-version }}
       - name: Test

README.md

@@ -20,6 +20,51 @@ def example_tests():
         test.assert_equals(add(1, 1), 2, 'Optional Message on Failure')
 ```
 
+### Assertion control flow and timeouts
+
+Assertions report a failure and continue by default. With `allow_raise=True`,
+the assertion stops the current decorated block and reports its failure once.
+The enclosing `it` or `describe` still closes normally, and its `after` hook
+runs. At top level, the framework's assertion exception still propagates.
+
+`@test.timeout(seconds)` executes immediately in a child process. On the
+Codewars Linux runtime it uses a local `fork` context so existing nested test
+functions work on Python 3.14 without changing the process-wide start method.
+Normal completion retains the existing additional passing assertion. An expired
+timeout, abnormal exit or exit before the wrapper completes reports a failure;
+timed-out children are terminated and joined, including children that ignore
+`SIGTERM`. Previously printed assertions remain visible.
+
+The fork context preserves the existing Linux execution model. Tests that start
+threads or import libraries with background threads still require validation in
+the full runner image. This compatibility change does not make fork safe for
+arbitrary multithreaded programs.
+
+### Development tests
+
+Failed comparisons retain the author's message and emit escaped `Expected`,
+`Actual` and `Location` details. Values are limited to 2,000 characters per
+field, with `[truncated]` marking omitted text. A failing bare `assert` gets a
+nonempty message and user-code location; passing bare assertions and empty
+groups do not add implicit successful assertions.
+
+An `after` hook runs even when `before` fails, and runs before the group closes.
+A failed `before` skips the body. Hook errors remain visible alongside any
+original body failure, and later sibling tests can still run. Cleanup hooks
+should therefore tolerate partially initialized state.
+
+Run the output fixtures and compatibility tests with the interpreter under test:
+
+```bash
+python -m unittest discover -v
+```
+
+The timeout regression tests run on Linux and check success/failure counts,
+nested functions, unchanged global start-method selection, early exits, signals,
+startup errors, child cleanup and `allow_raise` behavior. Other platforms skip
+these Linux-specific checks. Set `CW10_RESULTS` to a writable JSON file path to
+capture their raw outputs for integration checks with the Exo result parser.
+
 <!--
 
 ### Using Other Assertions

codewars_test/test_framework.py

@@ -1,5 +1,52 @@
 from __future__ import print_function
 
+import sys
+import traceback
+
+
+_VALUE_LIMIT = 2000
+
+
+def _bounded(value):
+    if len(value) > _VALUE_LIMIT:
+        return value[:_VALUE_LIMIT] + ' ... [truncated]'
+    return value
+
+
+def _safe_repr(value):
+    try:
+        return _bounded(repr(value))
+    except Exception:
+        return '<{}: repr unavailable>'.format(type(value).__name__)
+
+
+def _value_details(value, representation):
+    # Multiline strings are easier to compare as text. display() transports
+    # newlines inside a single escaped log event, including protocol-like text.
+    return _bounded(value) if type(value) is str else representation
+
+
+def _location(tb=None):
+    if tb is not None:
+        frames = traceback.extract_tb(tb)
+    else:
+        frames = traceback.extract_stack()[:-1]
+    for frame in reversed(frames):
+        if frame.filename != __file__:
+            return '{}:{} in {}'.format(frame.filename, frame.lineno, frame.name)
+    return ''
+
+
+def _failure(message, allow_raise=False, details=(), tb=None):
+    display('FAILED', message)
+    for label, value in details:
+        display('LOG', value, label=label, mode='ESC')
+    location = _location(tb)
+    if location:
+        display('LOG', location, label='Location', mode='ESC')
+    if allow_raise:
+        raise AssertException(message)
+
 
 class AssertException(Exception):
     pass
@@ -20,30 +67,36 @@ def expect(passed=None, message=None, allow_raise=False):
         display('PASSED', 'Test Passed')
     else:
         message = message or "Value is not what was expected"
-        display('FAILED', message)
-        if allow_raise:
-            raise AssertException(message)
+        _failure(message, allow_raise)
 
 
 def assert_equals(actual, expected, message=None, allow_raise=False):
-    equals_msg = "{0} should equal {1}".format(repr(actual), repr(expected))
+    r_actual, r_expected = _safe_repr(actual), _safe_repr(expected)
+    equals_msg = "{0} should equal {1}".format(r_actual, r_expected)
     if message is None:
         message = equals_msg
     else:
         message += ": " + equals_msg
 
-    expect(actual == expected, message, allow_raise)
+    if actual == expected:
+        pass_()
+    else:
+        _failure(message, allow_raise, (
+            ('Expected', _value_details(expected, r_expected)), ('Actual', _value_details(actual, r_actual))))
 
 
 def assert_not_equals(actual, expected, message=None, allow_raise=False):
-    r_actual, r_expected = repr(actual), repr(expected)
+    r_actual, r_expected = _safe_repr(actual), _safe_repr(expected)
     equals_msg = "{0} should not equal {1}".format(r_actual, r_expected)
     if message is None:
         message = equals_msg
     else:
         message += ": " + equals_msg
 
-    expect(not (actual == expected), message, allow_raise)
+    if not (actual == expected):
+        pass_()
+    else:
+        _failure(message, allow_raise, (('Unexpected value', _value_details(actual, r_actual)),))
 
 
 def expect_error(message, function, exception=Exception):
@@ -77,13 +130,18 @@ def fail(message): expect(False, message)
 def assert_approx_equals(
         actual, expected, margin=1e-9, message=None, allow_raise=False):
     msg = "{0} should be close to {1} with absolute or relative margin of {2}"
-    equals_msg = msg.format(repr(actual), repr(expected), repr(margin))
+    r_actual, r_expected = _safe_repr(actual), _safe_repr(expected)
+    equals_msg = msg.format(r_actual, r_expected, _safe_repr(margin))
     if message is None:
         message = equals_msg
     else:
         message += ": " + equals_msg
     div = max(abs(actual), abs(expected), 1)
-    expect(abs((actual - expected) / div) < margin, message, allow_raise)
+    if abs((actual - expected) / div) < margin:
+        pass_()
+    else:
+        _failure(message, allow_raise, (
+            ('Expected', _value_details(expected, r_expected)), ('Actual', _value_details(actual, r_actual))))
 
 
 '''
@@ -98,27 +156,46 @@ def it1():
 
 def _timed_block_factory(opening_text):
     from timeit import default_timer as timer
-    from traceback import format_exception
-    from sys import exc_info
+
+    def run_part(func, hook=None):
+        try:
+            func()
+        except AssertException:
+            # The assertion already emitted its failure and location.
+            pass
+        except AssertionError as error:
+            message = str(error) or 'Assertion failed'
+            if hook:
+                message = '{} hook: {}'.format(hook, message)
+            _failure(message, tb=sys.exc_info()[2])
+        except Exception:
+            if hook:
+                display('ERROR', '{} hook raised an exception'.format(hook))
+                display('LOG', ''.join(traceback.format_exception(*sys.exc_info())),
+                        label='Traceback', mode='ESC')
+            else:
+                fail('Unexpected exception raised')
+                display('ERROR', ''.join(traceback.format_exception(*sys.exc_info())))
+        else:
+            return True
+        return False
 
     def _timed_block_decorator(s, before=None, after=None):
         display(opening_text, s)
 
         def wrapper(func):
-            if callable(before):
-                before()
             time = timer()
             try:
-                func()
-            except AssertionError as e:
-                display('FAILED', str(e))
-            except Exception:
-                fail('Unexpected exception raised')
-                tb_str = ''.join(format_exception(*exc_info()))
-                display('ERROR', tb_str)
-            display('COMPLETEDIN', '{:.2f}'.format((timer() - time) * 1000))
-            if callable(after):
-                after()
+                if not callable(before) or run_part(before, 'before'):
+                    run_part(func)
+            finally:
+                try:
+                    if callable(after):
+                        run_part(after, 'after')
+                finally:
+                    # Cleanup belongs to the open group, including when setup
+                    # fails. Closing last prevents errors leaking into siblings.
+                    display('COMPLETEDIN', '{:.2f}'.format((timer() - time) * 1000))
         return wrapper
     return _timed_block_decorator
 
@@ -139,16 +216,71 @@ def some_tests():
 
 def timeout(sec):
     def wrapper(func):
-        from multiprocessing import Process
+        from multiprocessing import get_context
+        import os
+        import signal
+
         msg = 'Should not throw any exceptions inside timeout'
+        process = None
+        receiver = sender = None
 
         def wrapped():
-            expect_no_error(msg, func)
-        process = Process(target=wrapped)
-        process.start()
-        process.join(sec)
-        if process.is_alive():
-            fail('Exceeded time limit of {:.3f} seconds'.format(sec))
-            process.terminate()
-            process.join()
+            receiver.close()
+            try:
+                try:
+                    func()
+                except AssertException:
+                    # Preserve fail-fast behavior without reporting twice.
+                    pass
+                except BaseException as error:
+                    fail('{}: {}'.format(msg, repr(error)))
+                else:
+                    # Retain the existing extra pass for normal completion.
+                    pass_()
+                sender.send_bytes(b'completed')
+            finally:
+                sender.close()
+
+        try:
+            # Codewars runs on Linux. Existing kata use nested functions, which
+            # cannot be pickled by Python 3.14's default forkserver context.
+            # A local context leaves user code's start-method selection intact.
+            context = get_context('fork')
+            receiver, sender = context.Pipe(duplex=False)
+            process = context.Process(target=wrapped)
+            process.start()
+            sender.close()
+            process.join(sec)
+            if process.is_alive():
+                fail('Exceeded time limit of {:.3f} seconds'.format(sec))
+            elif process.exitcode != 0:
+                fail('Timeout process exited with code {}'.format(process.exitcode))
+            else:
+                completed = False
+                if receiver.poll():
+                    try:
+                        completed = receiver.recv_bytes() == b'completed'
+                    except EOFError:
+                        pass
+                if not completed:
+                    # os._exit(0) is not proof that the test body finished.
+                    fail('Timeout process exited before completing the test block')
+        except Exception as error:
+            fail('Unable to run timed tests: {}'.format(repr(error)))
+        finally:
+            if process is not None and process.pid is not None:
+                if process.is_alive():
+                    process.terminate()
+                    process.join(1)
+                    if process.is_alive():
+                        # A test can install or inherit a SIGTERM handler.
+                        try:
+                            os.kill(process.pid, signal.SIGKILL)
+                        except ProcessLookupError:
+                            pass
+                process.join()
+            if receiver is not None:
+                receiver.close()
+            if sender is not None:
+                sender.close()
     return wrapper

tests/fixtures/expect_error_sample.py

@@ -13,7 +13,8 @@ def f1():
 
 # BaseException >> Exception >> ArithmeticError >> ZeroDivisionError
 def f2():
-    return 1 // 0
+    # Exercise the exception hierarchy with stable text across Python versions.
+    raise ZeroDivisionError("integer division or modulo by zero")
 
 
 # BaseException >> Exception >> LookupError >> KeyError