(self)
| 140 | return remaining <= 0 |
| 141 | |
| 142 | def _should_refresh(self): |
| 143 | if self._frozen_token is None: |
| 144 | # We don't have a token yet, mandatory refresh |
| 145 | return "mandatory" |
| 146 | |
| 147 | expiration = self._frozen_token.expiration |
| 148 | if expiration is None: |
| 149 | # No expiration, so assume we don't need to refresh. |
| 150 | return None |
| 151 | |
| 152 | now = self._time_fetcher() |
| 153 | if now < self._next_refresh: |
| 154 | return None |
| 155 | |
| 156 | remaining = total_seconds(expiration - now) |
| 157 | |
| 158 | if remaining < self._mandatory_refresh_timeout: |
| 159 | return "mandatory" |
| 160 | elif remaining < self._advisory_refresh_timeout: |
| 161 | return "advisory" |
| 162 | |
| 163 | return None |
| 164 | |
| 165 | |
| 166 | class TokenProviderChain: |
no test coverage detected