Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_basic():
with HTTMock(mock.sync_get, mock.unknown):
with Trakt.configuration.auth('mock', 'mock'):
collection = Trakt['sync/playback'].get()
# Ensure collection is valid
assert_that(collection, not_none())
# Batman Begins (2005)
assert_that(collection[('imdb', 'tt0372784')], has_properties({
'id': 13
}))
# Breaking Bad (2008)
assert_that(collection[('tvdb', '81189')], has_properties({
# Seasons
'seasons': all_of(
has_length(1),
def test_basic():
with HTTMock(mock.fixtures, mock.unknown):
with Trakt.configuration.auth('mock', 'mock'):
items = Trakt['users/me/lists/people'].items()
# Ensure collection is valid
assert_that(items, not_none())
# Validate items
assert_that(items, contains(
# Bryan Cranston
all_of(
instance_of(Person),
has_properties({
'pk': ('tmdb', '17419'),
'name': 'Bryan Cranston',
# Timestamps
'listed_at': datetime(2014, 6, 17, 6, 52, 3, tzinfo=tzutc()),
def test_basic():
with HTTMock(mock.fixtures, mock.unknown):
with Trakt.configuration.auth('mock', 'mock'):
collection = Trakt['sync/ratings'].movies()
# Ensure collection is valid
assert_that(collection, not_none())
# 100 Bloody Acres (2012)
assert_that(collection[('imdb', 'tt2290065')], has_properties({
'pk': ('imdb', 'tt2290065'),
'title': '100 Bloody Acres',
'year': 2012,
'rating': has_properties({
'value': 8,
'votes': None,
'timestamp': datetime(2015, 1, 28, 2, 26, 37, tzinfo=tzutc())
}),
def test_basic():
with HTTMock(mock.fixtures, mock.unknown):
with Trakt.configuration.auth('mock', 'mock'):
collection = Trakt['sync/collection'].shows()
# Ensure collection is valid
assert_that(collection, not_none())
assert_that(collection, has_length(4))
# Chuck (2007)
assert_that(collection[('tvdb', '80348')], has_properties({
'pk': ('tvdb', '80348'),
'title': 'Chuck',
'year': 2007,
# Seasons
'seasons': all_of(
has_length(1),
has_entry(1, has_properties({
def test_watched():
with HTTMock(mock.fixtures, mock.unknown):
with Trakt.configuration.auth('mock', 'mock'):
items = Trakt['users/me/lists/shows'].items()
# Ensure collection is valid
assert_that(items, not_none())
assert_that(items, has_length(3))
# Validate items
assert_that(items, contains(
# Game of Thrones (2011)
all_of(
instance_of(Show),
has_properties({
'pk': ('tvdb', '121361'),
'title': 'Game of Thrones',
'year': 2011,
def test_request():
with HTTMock(mock.oauth_token, mock.fixtures, mock.unknown):
# Mock authorization
authorization = {
'access_token': 'mock',
'token_type': 'bearer',
'created_at': calendar.timegm(datetime.datetime.utcnow().utctimetuple()),
'expires_in': 7 * 24 * 60 * 60,
'refresh_token': 'mock-refresh_token',
'scope': 'public'
}
# Test valid token
with Trakt.configuration.oauth.from_response(authorization):
assert Trakt['sync/collection'].movies() is not None
# Test expired token
authorization['expires_in'] = 0
with Trakt.configuration.oauth.from_response(authorization):
assert Trakt['sync/collection'].movies() is None
# Test token refreshing
with Trakt.configuration.oauth.from_response(authorization, refresh=True):
assert Trakt['sync/collection'].movies() is not None
print('Authentication required')
exit(1)
# Simulate expired token
self.authorization['expires_in'] = 0
# Test authenticated calls
with Trakt.configuration.oauth.from_response(self.authorization):
# Expired token, requests will return `None`
print(Trakt['sync/collection'].movies())
with Trakt.configuration.oauth.from_response(self.authorization, refresh=True):
# Expired token will be refreshed automatically (as `refresh=True`)
print(Trakt['sync/collection'].movies())
with Trakt.configuration.oauth.from_response(self.authorization):
# Current token is still valid
print(Trakt['sync/collection'].movies())
secret='bf00575b1ad252b514f14b2c6171fe650d474091daad5eb6fa890ef24d581f65'
)
# Application
Trakt.configuration.defaults.app(
name='trakt (for Plex)',
version=PLUGIN_VERSION
)
# Http
Trakt.base_url = (
config.get('protocol', 'https') + '://' +
config.get('hostname', 'api.trakt.tv')
)
Trakt.configuration.defaults.http(
timeout=timeout
)
# Configure keep-alive
Trakt.http.keep_alive = config.get_boolean('keep_alive', True)
# Configure requests adapter
Trakt.http.adapter_kwargs = {
'pool_connections': config.get_int('pool_connections', 10),
'pool_maxsize': config.get_int('pool_size', 10),
'max_retries': Retry(
total=config.get_int('connect_retries', 3),
read=0
)
}
print('Token exchanged - authorization: %r' % self.authorization)
return True
def on_token_refreshed(self, response):
# OAuth token refreshed, save token for future calls
self.authorization = response
print('Token refreshed - authorization: %r' % self.authorization)
if __name__ == '__main__':
# Configure
Trakt.base_url = 'http://api.staging.trakt.tv'
Trakt.configuration.defaults.app(
id=os.environ.get('APP_ID')
)
Trakt.configuration.defaults.client(
id=os.environ.get('CLIENT_ID'),
secret=os.environ.get('CLIENT_SECRET')
)
# Start application
app = Application()
app.run()