Verified Commit 03d64284 authored by Jan Koppe's avatar Jan Koppe
Browse files

WIP

parent aca945ac
# Copyright 2017 The WWU eLectures Team All rights reserved.
#
# Licensed under the Educational Community License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://opensource.org/licenses/ECL-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Copyright 2017 The WWU eLectures Team All rights reserved.
#
# Licensed under the Educational Community License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://opensource.org/licenses/ECL-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from base64 import b64encode
import logging
import requests
import urllib.parse
logger = logging.getLogger(__name__)
# Supported upstream API version
API_VERSION = "v1.0.0"
class BaseClient():
def __init__(self, host, user, password):
# Account credentials to authenticate with against the API. make sure
# that this account has the required API_* roles for your desired
# functions.
self.auth_user = user
self.auth_password = password
# External API uses Basic Auth, which transmits credentials Base64
# encoded. Using HTTP would transmit passwords in plain text, this
# is a bad idea.
self.server_url = 'https://%s' % host
def req(self, path, method='GET', headers={},
run_as=None, run_with_roles=[], **kwargs):
url = urllib.parse.urljoin(self.server_url, '/api' + path)
if run_as:
headers['X-RUN-AS-USER'] = run_as
if len(run_with_roles) > 0:
headers['X-RUN-WITH-ROLES'] = ''.join(run_with_roles, ',')
headers['Accept'] = 'application/%s+json' % API_VERSION
# We need to immediately send the basic auth headers.
auth_bytes = str.encode('%s:%s' % (self.auth_user, self.auth_password))
headers['Authorization'] = 'Basic %s' %\
b64encode(auth_bytes).decode('ascii')
return requests.request(method, url, **kwargs)
......@@ -43,7 +43,7 @@ setup(
'Topic :: Software Development :: Libraries :: Python Modules'
],
keywords='api client development opencast externalapi',
packages=['opencast-externalapi'],
packages=['opencast_externalapi'],
install_requires=['requests'],
test_suite='nose.collector',
tests_require=['nose', 'flake8'],
......
import unittest
from opencast_externalapi import BaseClient
class TestBaseClient(unittest.TestCase):
def setUp(self):
self.base_client = BaseClient(host='develop.opencast.org',
user='admin',
password='opencast'
)
def test_server_url(self):
self.assertEqual(self.base_client.server_url,
'https://develop.opencast.org')
def test_req(self):
r = self.base_client.req('/info/me')
me = r.json()
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment