[3.13] gh-130954: Fix multiprocessing test_notify_n (GH-130955) (#130981)

The test could deadlock trying join on the worker processes.
Apply the same technique as gh-130933.

Join the process before the test ends in `test_notify` as well.
(cherry picked from commit edd1eca)

Co-authored-by: Sam Gross <[email protected]>

Author: miss-islington

Lib/test/_test_multiprocessing.py

@@ -1609,12 +1609,10 @@ def test_notify(self):
         p = self.Process(target=self.f, args=(cond, sleeping, woken))
         p.daemon = True
         p.start()
-        self.addCleanup(p.join)
 
-        p = threading.Thread(target=self.f, args=(cond, sleeping, woken))
-        p.daemon = True
-        p.start()
-        self.addCleanup(p.join)
+        t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
+        t.daemon = True
+        t.start()
 
         # wait for both children to start sleeping
         sleeping.acquire()
@@ -1641,7 +1639,9 @@ def test_notify(self):
 
         # check state is not mucked up
         self.check_invariant(cond)
-        p.join()
+
+        threading_helper.join_thread(t)
+        join_process(p)
 
     def test_notify_all(self):
         cond = self.Condition()
@@ -1718,16 +1718,17 @@ def test_notify_n(self):
         woken = self.Semaphore(0)
 
         # start some threads/processes
+        workers = []
         for i in range(3):
             p = self.Process(target=self.f, args=(cond, sleeping, woken))
             p.daemon = True
             p.start()
-            self.addCleanup(p.join)
+            workers.append(p)
 
             t = threading.Thread(target=self.f, args=(cond, sleeping, woken))
             t.daemon = True
             t.start()
-            self.addCleanup(t.join)
+            workers.append(t)
 
         # wait for them to all sleep
         for i in range(6):
@@ -1762,6 +1763,10 @@ def test_notify_n(self):
         # check state is not mucked up
         self.check_invariant(cond)
 
+        for w in workers:
+            # NOTE: join_process and join_thread are the same
+            threading_helper.join_thread(w)
+
     def test_timeout(self):
         cond = self.Condition()
         wait = TimingWrapper(cond.wait)