fix: task 10 + review fix #1 — thread safety and sleep bug

Fix #1: Don't sleep after the final exhausted retry attempt in
_graph_request(). The sleep on the last iteration was pure overhead —
the loop exits immediately after, so the caller just waited an extra
backoff period for no reason. Guard with attempt < _MAX_RETRIES - 1.

Task 10: Add threading.Lock (_edits_lock) for all compound operations
on active_edits, which is accessed from both the UI thread and
background edit threads:
- __init__: declare self._edits_lock = threading.Lock()
- open_file: snapshot already_editing and at_limit under the lock,
  then release before showing blocking UI dialogs
- process_file: wrap initial dict assignment, waiting=True, the
  in-check+waiting=False, and in-check+del under the lock
- on_done_editing_clicked: lock the items() snapshot used to build
  waiting_files, preventing iteration over a dict being mutated

Add 8 new unit tests (1 for Fix #1 sleep count, 7 for Task 10 lock).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Martin Tranberg
2026-04-12 10:36:29 +02:00
parent 0dfef3e611
commit 7fd69a9c3f
2 changed files with 101 additions and 15 deletions

View File

@@ -421,6 +421,81 @@ class TestGraphRequest(unittest.TestCase):
"_MAX_RETRIES constant not found in sharepoint_browser"
)
def test_no_sleep_after_final_retry(self):
"""Fix #1: sleep() is NOT called after the last exhausted attempt."""
resp_429 = MagicMock(status_code=429, headers={"Retry-After": "0"})
responses = [resp_429] * sb._MAX_RETRIES
with patch("sharepoint_browser.requests.request", side_effect=responses):
with patch("sharepoint_browser.time.sleep") as mock_sleep:
sb._graph_request("GET", "https://example.com/", headers={})
# sleep should be called for the first N-1 failures, NOT the last one
self.assertEqual(
mock_sleep.call_count,
sb._MAX_RETRIES - 1,
f"sleep() called {mock_sleep.call_count} times; expected {sb._MAX_RETRIES - 1} "
f"(no sleep after the final failed attempt)"
)
# ---------------------------------------------------------------------------
# Task 10: threading.Lock for active_edits compound operations
# ---------------------------------------------------------------------------
class TestActiveEditsLock(unittest.TestCase):
"""Verify that _edits_lock exists and guards all compound active_edits operations."""
def test_edits_lock_declared_in_init(self):
"""Task 10: SharePointApp.__init__ creates self._edits_lock."""
source = inspect.getsource(sb.SharePointApp.__init__)
self.assertIn("_edits_lock", source,
"__init__ does not declare _edits_lock")
def test_edits_lock_is_threading_lock(self):
"""Task 10: _edits_lock initialisation uses threading.Lock()."""
source = inspect.getsource(sb.SharePointApp.__init__)
self.assertIn("threading.Lock()", source,
"__init__ does not initialise _edits_lock with threading.Lock()")
def test_open_file_uses_lock(self):
"""Task 10: open_file() acquires _edits_lock before checking active_edits."""
source = inspect.getsource(sb.SharePointApp.open_file)
self.assertIn("_edits_lock", source,
"open_file does not use _edits_lock")
def test_process_file_uses_lock(self):
"""Task 10: process_file() acquires _edits_lock when writing active_edits."""
source = inspect.getsource(sb.SharePointApp.process_file)
self.assertIn("_edits_lock", source,
"process_file does not use _edits_lock")
def test_process_file_lock_on_initial_assign(self):
"""Task 10: active_edits[item_id] = ... assignment is inside a lock block."""
source = inspect.getsource(sb.SharePointApp.process_file)
# Check lock wraps the initial dict assignment
lock_idx = source.find("_edits_lock")
assign_idx = source.find('self.active_edits[item_id] = ')
self.assertGreater(assign_idx, 0,
"active_edits assignment not found in process_file")
# The lock must appear before the assignment
self.assertLess(lock_idx, assign_idx,
"_edits_lock must appear before the active_edits assignment in process_file")
def test_process_file_lock_on_delete(self):
"""Task 10: del active_edits[item_id] is inside a lock block in process_file."""
source = inspect.getsource(sb.SharePointApp.process_file)
self.assertIn("del self.active_edits[item_id]", source,
"del active_edits[item_id] not found in process_file")
# Count lock usages — there should be at least 2
lock_count = source.count("_edits_lock")
self.assertGreaterEqual(lock_count, 2,
f"Expected at least 2 uses of _edits_lock in process_file, found {lock_count}")
def test_on_done_editing_uses_lock(self):
"""Task 10: on_done_editing_clicked acquires lock for active_edits iteration."""
source = inspect.getsource(sb.SharePointApp.on_done_editing_clicked)
self.assertIn("_edits_lock", source,
"on_done_editing_clicked does not use _edits_lock")
if __name__ == "__main__":
unittest.main(verbosity=2)