| 2652 | """ |
| 2653 | |
| 2654 | def __enter__(self): |
| 2655 | self._counter = defaultdict(int) |
| 2656 | self._thread_id = threading.get_ident() |
| 2657 | self._extra_info = [] |
| 2658 | |
| 2659 | def patched_with_thread_info(func): |
| 2660 | def wrap(*args, **kwargs): |
| 2661 | self._extra_info.append(threading.get_ident()) |
| 2662 | return func(*args, **kwargs) |
| 2663 | |
| 2664 | return wrap |
| 2665 | |
| 2666 | import urllib3 |
| 2667 | |
| 2668 | self.patcher = patch.object( |
| 2669 | urllib3.connectionpool.log, "debug", side_effect=patched_with_thread_info(urllib3.connectionpool.log.debug) |
| 2670 | ) |
| 2671 | self.mock = self.patcher.start() |
| 2672 | return self |
| 2673 | |
| 2674 | def __exit__(self, *args, **kwargs) -> None: |
| 2675 | assert len(self.mock.call_args_list) == len(self._extra_info) |