Method test_threaded_hashing

Lib/test/test_hashlib.py:1028–1061

(self)

    @threading_helper.reap_threads
    @threading_helper.requires_working_threading()
    def test_threaded_hashing(self):
        # 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.
        hasher = hashlib.sha1()
        num_threads = 5
        smallest_data = b'swineflu'
        data = smallest_data * 200000
        expected_hash = hashlib.sha1(data*num_threads).hexdigest()

        def hash_in_chunks(chunk_size):
            index = 0
            while index < len(data):
                hasher.update(data[index:index + chunk_size])
                index += chunk_size

        threads = []
        for threadnum in range(num_threads):
            chunk_size = len(data) // (10 ** threadnum)
            self.assertGreater(chunk_size, 0)
            self.assertEqual(chunk_size % len(smallest_data), 0)
            thread = threading.Thread(target=hash_in_chunks,
                                      args=(chunk_size,))
            threads.append(thread)

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

        self.assertEqual(expected_hash, hasher.hexdigest())

    def test_get_fips_mode(self):
        fips_mode = self.is_fips_mode