(self)
| 90 | return self._frozen_token |
| 91 | |
| 92 | def _refresh(self): |
| 93 | # If we don't need to refresh just return |
| 94 | refresh_type = self._should_refresh() |
| 95 | if not refresh_type: |
| 96 | return None |
| 97 | |
| 98 | # Block for refresh if we're in the mandatory refresh window |
| 99 | block_for_refresh = refresh_type == "mandatory" |
| 100 | if self._refresh_lock.acquire(block_for_refresh): |
| 101 | try: |
| 102 | self._protected_refresh() |
| 103 | finally: |
| 104 | self._refresh_lock.release() |
| 105 | |
| 106 | def _protected_refresh(self): |
| 107 | # This should only be called after acquiring the refresh lock |
no test coverage detected