jenkins-bot has submitted this change and it was merged. ( https://gerrit.wikimedia.org/r/542438 )

Change subject: [tests] Tests for log_page/unlock_page implementation ......................................................................

[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(-)

Approvals: Zhuyifei1999: Looks good to me, approved jenkins-bot: Verified

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."""