jenkins-bot merged this change.

View Change

[tests] Tests for log_page/unlock_page implementation

- Add a test class to test whether log_page/unlock_page works
  successfull with threads or detect deadlocks.
  If a deadlock is detected a timeout occures with a thread
  and is_alive() is True in that case.

tests detached from I1ff2c39

Change-Id: I83b2de3378ef1b7c9f2ab0852a37a7511bc6c40c
---
M tests/site_tests.py
1 file changed, 37 insertions(+), 0 deletions(-)

diff --git a/tests/site_tests.py b/tests/site_tests.py
index 306ea27..58a8a3b 100644
--- a/tests/site_tests.py
+++ b/tests/site_tests.py
@@ -9,14 +9,17 @@
 
 import os
 import pickle
+import random
 import re
 import sys
+import time
 
 try:
     from collections.abc import Iterable, Mapping
 except ImportError:  # Python 2.7
     from collections import Iterable, Mapping
 from datetime import datetime
+import threading
 
 import pywikibot
 
@@ -1087,6 +1090,40 @@
         self.assertRaises(AssertionError, func, 'm', 2, 1, True, True)
 
 
+class TestThreadsLockingPage(DefaultSiteTestCase):
+    """Test cases for lock/unlock a page within threads."""
+
+    cached = True
+
+    def worker(self):
+        """Lock a page, wait few seconds and unlock the page."""
+        page = pywikibot.Page(self.site, 'Foo')
+        page.site.lock_page(page=page, block=True)
+        wait = random.randint(1, 25) / 10
+        time.sleep(wait)
+        page.site.unlock_page(page=page)
+
+    def test_lock_page(self):
+        """Test the site.lock_page() and site.unlock_page() method."""
+        # Start few threads
+        for i in range(5):
+            thread = threading.Thread(target=self.worker)
+            thread.setDaemon(True)
+            thread.start()
+
+        current_thread = threading.currentThread()
+        for thread in threading.enumerate():
+            if thread is current_thread:
+                continue
+            thread.join(15)  # maximum wait time for all threads
+
+            with self.subTest(name=thread.getName()):
+                # Check whether a timeout happened.
+                # In that case is_alive() is True
+                self.assertFalse(thread.is_alive(),
+                                 'test page is still locked')
+
+
 class TestSiteGeneratorsUsers(DefaultSiteTestCase):
     """Test cases for Site methods with users."""
 

To view, visit change 542438. To unsubscribe, or for help writing mail filters, visit settings.