Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions tests/test_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
from trakt.api import HttpClient
from trakt.core import api
from trakt.tv import TVShow


def test_api_singleton():
"""Test that api() returns the same HttpClient instance when called multiple times."""
api1 = api()
api2 = api()
assert isinstance(api1, HttpClient), "api() should return an HttpClient instance"
assert api1 == api2, "Multiple calls to api() should return the same instance"


def test_tvshow_properties():
show = TVShow("Game of Thrones")
assert show.title == "Game of Thrones"
assert show.certification == "TV-MA"
12 changes: 11 additions & 1 deletion trakt/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,12 +172,16 @@ class TokenAuth(AuthBase):
#: The OAuth2 Redirect URI for your OAuth Application
REDIRECT_URI: str = 'urn:ietf:wg:oauth:2.0:oob'

#: How many times to attempt token auth refresh before failing
MAX_RETRIES = 1

def __init__(self, client: HttpClient, config: AuthConfig):
super().__init__()
self.config = config
self.client = client
# OAuth token validity checked
self.OAUTH_TOKEN_VALID = None
self.refresh_attempts = 0
self.logger = logging.getLogger('trakt.api.token_auth')

def __call__(self, r):
Expand Down Expand Up @@ -220,6 +224,11 @@ def validate_token(self):
def refresh_token(self):
"""Request Trakt API for a new valid OAuth token using refresh_token"""

if self.refresh_attempts >= self.MAX_RETRIES:
self.logger.error("Max token refresh attempts reached. Manual intervention required.")
return
self.refresh_attempts += 1

self.logger.info("OAuth token has expired, refreshing now...")
data = {
'client_id': self.config.CLIENT_ID,
Expand All @@ -230,7 +239,8 @@ def refresh_token(self):
}

try:
response = self.client.post('/oauth/token', data)
response = self.client.post('oauth/token', data)
self.refresh_attempts = 0
except OAuthException:
self.logger.debug(
"Rejected - Unable to refresh expired OAuth token, "
Expand Down
Loading