Method do_test_threaded_hashing

Lib/test/test_hashlib.py:1121–1161

(self, constructor, is_shake)

                    self.do_test_threaded_hashing(constructor, is_shake)

    def do_test_threaded_hashing(self, constructor, is_shake):
        # Updating the same hash object from several threads at once
        # using data chunk sizes containing the same byte sequences.
        #
        # If the internal locks are working to prevent multiple
        # updates on the same object from running at once, the resulting
        # hash will be the same as doing it single threaded upfront.

        # The data to hash has length s|M|q^N and the chunk size for the i-th
        # thread is s|M|q^(N-i), where N is the number of threads, M is a fixed
        # message of small length, and s >= 1 and q >= 2 are small integers.
        smallest_size, num_threads, s, q = 8, 5, 2, 10

        smallest_data = os.urandom(smallest_size)
        data = s * smallest_data * (q ** num_threads)

        h1 = constructor(usedforsecurity=False)
        h2 = constructor(data * num_threads, usedforsecurity=False)

        def update(chunk_size):
            for index in range(0, len(data), chunk_size):
                h1.update(data[index:index + chunk_size])

        threads = []
        for thread_num in range(num_threads):
            # chunk_size = len(data) // (q ** thread_num)
            chunk_size = s * smallest_size * q ** (num_threads - thread_num)
            self.assertGreater(chunk_size, 0)
            self.assertEqual(chunk_size % smallest_size, 0)
            thread = threading.Thread(target=update, args=(chunk_size,))
            threads.append(thread)

        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()

        if is_shake:
            self.assertEqual(h1.hexdigest(16), h2.hexdigest(16))
        else:
            self.assertEqual(h1.hexdigest(), h2.hexdigest())

    def test_get_fips_mode(self):
        fips_mode = self.is_fips_mode