add test

Author: ashm-dev

Lib/test/test_struct.py

@@ -4,6 +4,7 @@
 import gc
 import math
 import operator
+import threading
 import unittest
 import platform
 import struct
@@ -39,6 +40,24 @@ def bigendian_to_native(value):
     else:
         return string_reverse(value)
 
+# Helper for test_endian_table_init_subinterpreters
+class _Result:
+    def __init__(self):
+        self.ret = -1
+        self.err = None
+
+def _run_in_subinterp_worker(code, barrier, result):
+    """
+    Worker function for a thread. Waits on the barrier, then runs the
+    code in a subinterpreter.
+    """
+    try:
+        # Wait until all threads are ready to start simultaneously.
+        barrier.wait()
+        result.ret = support.run_in_subinterp(code)
+    except Exception as e:
+        result.err = e
+
 class StructTest(ComplexesAreIdenticalMixin, unittest.TestCase):
     def test_isbigendian(self):
         self.assertEqual((struct.pack('=i', 1)[0] == 0), ISBIGENDIAN)
@@ -800,6 +819,44 @@ def test_c_complex_round_trip(self):
                     round_trip = struct.unpack(f, struct.pack(f, z))[0]
                     self.assertComplexesAreIdentical(z, round_trip)
 
+    def test_endian_table_init_subinterpreters(self):
+        # Verify that struct works correctly in subinterpreters after
+        # once-only endian table initialization (gh-140260), when
+        # initialized concurrently.
+        # Use struct in the main interpreter first.
+        self.assertEqual(struct.unpack('>i', struct.pack('>i', 1))[0], 1)
+        self.assertEqual(struct.unpack('<i', struct.pack('<i', 1))[0], 1)
+
+        code = (
+            "import struct\n"
+            "x = struct.pack('>i', 1)\n"
+            "assert struct.unpack('>i', x)[0] == 1\n"
+            "y = struct.pack('<i', 1)\n"
+            "assert struct.unpack('<i', y)[0] == 1\n"
+        )
+
+        num_threads = 3
+        barrier = threading.Barrier(num_threads)
+        results = [_Result() for _ in range(num_threads)]
+        threads = [
+            threading.Thread(
+                target=_run_in_subinterp_worker,
+                args=(code, barrier, results[i])
+            )
+            for i in range(num_threads)
+        ]
+
+        for t in threads:
+            t.start()
+
+        for t in threads:
+            t.join()
+
+        for result in results:
+            if result.err:
+                raise result.err
+            self.assertEqual(result.ret, 0)
+
 
 class UnpackIteratorTest(unittest.TestCase):
     """