Browse Source

Implemented first api calls

tags/0.1
Johann Schmitz 3 years ago
parent
commit
5b9f83c2e3
Signed by: ercpe <johann@j-schmitz.net> GPG Key ID: A084064277C501ED
8 changed files with 286 additions and 0 deletions
  1. +1
    -0
      .gitignore
  2. +1
    -0
      requirements.txt
  3. +2
    -0
      requirements_dev.txt
  4. +2
    -0
      tests/__init__.py
  5. +71
    -0
      tests/test_language.py
  6. +89
    -0
      tests/test_login.py
  7. +3
    -0
      tvdbrest/__init__.py
  8. +117
    -0
      tvdbrest/client.py

+ 1
- 0
.gitignore View File

@@ -96,3 +96,4 @@ ENV/
.idea
dev_settings.py*
testresults.xml
dummy.py

+ 1
- 0
requirements.txt View File

@@ -1 +1,2 @@
# Project dependencies
requests

+ 2
- 0
requirements_dev.txt View File

@@ -1,3 +1,5 @@
# Project dependencies for development
pytest
coveralls
mock
coverage

+ 2
- 0
tests/__init__.py View File

@@ -0,0 +1,2 @@
# -*- coding: utf-8 -*-


+ 71
- 0
tests/test_language.py View File

@@ -0,0 +1,71 @@
# -*- coding: utf-8 -*-

import mock
import pytest

from tvdbrest.client import TVDB, Unauthorized, NotFound, Language


@pytest.fixture
def tvdb():
tvdb = TVDB("myusername", "myuserkey", "myapikey")
tvdb.jwttoken = "test-token"
return tvdb


def api_response_mock(json):
m = mock.MagicMock()
m.status_code = 200
m.json = mock.Mock(return_value=json)
return m


def api_response_404_mock():
m = mock.MagicMock()
m.status_code = 404
return m


class TestAPI(object):

@mock.patch('tvdbrest.client.requests.request')
def test_languages(self, request_method_mock, tvdb):
request_method_mock.return_value = api_response_mock({
"data": [
{
"id": 27,
"abbreviation": "zh",
"name": "中文",
"englishName": "Chinese"
},
{
"id": 7,
"abbreviation": "en",
"name": "English",
"englishName": "English"
}
]})
languages = tvdb.languages()
assert languages
assert all(isinstance(x, Language) for x in languages)

@mock.patch('tvdbrest.client.requests.request')
def test_get_language(self, request_method_mock, tvdb):
request_method_mock.return_value = api_response_mock({
"id": 27,
"abbreviation": "zh",
"name": "中文",
"englishName": "Chinese"
})
language = tvdb.language(27)
assert language
assert language.id == 27

@mock.patch('tvdbrest.client.requests.request')
def test_get_language_does_not_exist(self, request_method_mock, tvdb):
request_method_mock.return_value = api_response_404_mock()
with pytest.raises(NotFound):
tvdb.language(42)

+ 89
- 0
tests/test_login.py View File

@@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
import mock
import pytest

from tvdbrest.client import TVDB, Unauthorized


@pytest.fixture
def tvdb():
return TVDB("myusername", "myuserkey", "myapikey")


@pytest.fixture
def empty_positive_response():
m = mock.Mock()
m.status_code = 200
m.json = mock.Mock(return_value={"data": []})
return m


class TestLoginLogout(object):
def test_login_status(self):
tvdb = TVDB("myusername", "myuserkey", "myapikey")
assert not tvdb.logged_in

@mock.patch('tvdbrest.client.requests.request')
def test_login_status_after_login(self, request_mock):
response_mock = mock.MagicMock()
response_mock.status_code = 200
response_mock.json = mock.MagicMock(return_value={
'token': 'jwttoken'
})
request_mock.return_value = response_mock
tvdb = TVDB("myusername", "myuserkey", "myapikey")
tvdb.login()
request_mock.assert_called_with('post', 'https://api.thetvdb.com/login', headers={}, json={
'username': 'myusername',
'userkey': 'myuserkey',
'apikey': 'myapikey'
})

assert tvdb.logged_in

@mock.patch('tvdbrest.client.requests.request')
def test_failed_login(self, request_method_mock):
response_mock = mock.MagicMock()
response_mock.status_code = 401
request_method_mock.return_value = response_mock
tvdb = TVDB("myusername", "myuserkey", "myapikey")
with pytest.raises(Unauthorized):
tvdb.login()
assert not tvdb.logged_in

def test_logout(self, tvdb):
tvdb.jwttoken = "abc"
assert tvdb.logged_in
tvdb.logout()
assert tvdb.jwttoken is None
assert not tvdb.logged_in

@mock.patch('tvdbrest.client.requests.request')
def test_decorator_login_before_api_call(self, request_method_mock, tvdb):
response_mock = mock.MagicMock()
response_mock.status_code = 200
response_mock.json = mock.MagicMock(return_value={"data": []})

request_method_mock.return_value = response_mock

login_mock = mock.MagicMock()
tvdb.login = login_mock
tvdb.languages()
assert login_mock.called

@mock.patch('tvdbrest.client.requests.request')
def test_authorization_for_api_call(self, request_mock, tvdb, empty_positive_response):
request_mock.return_value = empty_positive_response
tvdb.jwttoken = "test"
tvdb.languages()
request_mock.assert_called_with('get', 'https://api.thetvdb.com/languages', headers={
'Authorization': 'Bearer test'
})

+ 3
- 0
tvdbrest/__init__.py View File

@@ -0,0 +1,3 @@
# -*- coding: utf-8 -*-

VERSION = "0.1"

+ 117
- 0
tvdbrest/client.py View File

@@ -0,0 +1,117 @@
# -*- coding: utf-8 -*-
import logging
from functools import wraps
from urllib.parse import urljoin

import requests

from tvdbrest import VERSION

logger = logging.getLogger(__name__)


class Unauthorized(Exception):
pass


class NotFound(Exception):
pass


class APIError(Exception):
pass


class APIObject(object):
def __init__(self, attrs):
self._attrs = attrs
def __getattr__(self, item):
return self._attrs[item]

def __eq__(self, other):
return isinstance(other, self.__class__) and self.id == other.id


class Language(APIObject):
pass


def login_required(f):
@wraps(f)
def wrapper(obj, *args, **kwargs):
if not obj.logged_in:
logger.debug("not logged in")
obj.login()

try:
return f(obj, *args, **kwargs)
except Unauthorized:
logger.info("Unauthorized API error - login again")
obj.login()
return f(obj, *args, **kwargs)
return wrapper


class TVDB(object):
def __init__(self, username, userkey, apikey):
self.username = username
self.userkey = userkey
self.apikey = apikey
assert self.username and self.userkey and self.apikey
self.jwttoken = None
self.useragent = "tvdb-rest %s" % VERSION

def login(self):
self.jwttoken = None
response = self._api_request('post', '/login', json={
'username': self.username,
'userkey': self.userkey,
'apikey': self.apikey,
})
self.jwttoken = response['token']
def logout(self):
self.jwttoken = None
@property
def logged_in(self):
return self.jwttoken is not None
@login_required
def languages(self):
return self._api_request('get', '/languages', response_class=Language, many=True)
@login_required
def language(self, id):
return self._api_request('get', '/languages/%s' % id, response_class=Language)
def _api_request(self, method, relative_url, response_class=None, many=False, **kwargs):

url = urljoin('https://api.thetvdb.com/', relative_url)

headers = kwargs.pop('headers', {})
if self.jwttoken:
headers['Authorization'] = 'Bearer %s' % self.jwttoken

response = requests.request(method, url, headers=headers, **kwargs)
if response.status_code == 401:
raise Unauthorized()
elif response.status_code == 404:
raise NotFound()
elif response.status_code >= 400:
raise APIError()
logger.info("Response: %s", response)
if response_class:
if many:
return [response_class(d) for d in response.json()['data']]
return response_class(response.json())
return response.json()

Loading…
Cancel
Save