Verified Commit 1a5cace1 authored by Jan Koppe's avatar Jan Koppe
Browse files

baseclient, started groups

parent 03d64284
......@@ -12,49 +12,12 @@
# 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
from .base import Base
from .groups import Groups
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)
class ExternalApi(Base, Groups):
"""
Collection class
"""
pass
# 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.
import logging
import requests
import urllib.parse
logger = logging.getLogger(__name__)
# Supported upstream API version
DEFAULT_API_VERSION = "v1.0.0"
class ApiClient():
def __init__(self, host, user='admin', password='opencast', api_version=DEFAULT_API_VERSION):
# 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.
if not host.startswith('http'):
host = 'https://%s' % host
self.server_url = host
self.api_version = api_version
def req(self, path, method='GET', headers={},
run_as=None,
run_with_roles=[],
parameters={},
sort=None,
limit=None,
offset=None,
**kwargs):
if sort:
parameters['sort'] = sort
if limit:
parameters['limit'] = limit
if offset:
parameters['offset'] = offset
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' % self.api_version
url = urllib.parse.urljoin(self.server_url, '/api' + path)
logger.debug(parameters)
return requests.request(method, url, auth=(self.auth_user, self.auth_password), params=parameters, **kwargs)
def get_json(self, path, **kwargs):
r = self.req(path, **kwargs)
# TODO: define and raise detailed exceptions
r.raise_for_status()
return r.json()
# 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 .apiclient import ApiClient
class Base(ApiClient):
def api_information(self, **kwargs) -> dict():
"""
Returns key characteristics of the API such as the server name and the
default version.
Returns:
dict:
- url
- version
"""
return self.get_json('/', **kwargs)
def me(self, **kwargs) -> dict():
"""
Returns information on the logged in user.
The current user is either the auth_user or, if provided, the run_as
user.
Returns:
dict:
- email
- name
- provider
- userrole
- username
"""
return self.get_json('/info/me', **kwargs)
def me_roles(self, **kwargs) -> list():
"""
Returns current user's roles.
Returns:
list: user's roles
"""
return self.get_json('/info/me/roles', **kwargs)
def organization(self, **kwargs) -> dict():
"""
Returns the current organization.
Returns:
dict:
- adminRole
- anonymousRole
- id
- name
"""
return self.get_json('/info/organization', **kwargs)
def organization_properties(self, **kwargs) -> dict():
"""
Returns the current organization's properties.
Returns:
dict: organization's properties
"""
return self.get_json('/info/organization/properties', **kwargs)
# 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 .apiclient import ApiClient
class Events(ApiClient):
def get_events(self,
sign=False,
with_acl=False,
with_metadata=False,
with_publications=False,
filter=None,
**kwargs) -> dict():
parameters = {
'sign': sign,
'withacl': with_acl,
'withmetadata': with_metadata,
'withpublications': with_publications,
}
# TODO: validate filter
if filter:
parameters['filter'] = filter
return self.get_json('/events', parameters, **kwargs)
# 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 .apiclient import ApiClient
class Groups(ApiClient):
def get_group(self, group_id, **kwargs) -> dict():
return self.get_json('/groups/%s' % group_id, **kwargs)
def get_groups(self, **kwargs) -> dict():
return self.get_json('/groups', **kwargs)
def create_group(self, name, desc="", roles=[], members=[], **kwargs) -> bool():
roles.append('ROLE_GROUP_%s' % name)
data = {
'name': name,
'description': desc,
'roles': ','.join(roles),
'members': ','.join(members),
}
r = self.req('/groups', method='POST', data=data, **kwargs)
return r.status_code == 201
def delete_group(self, name, **kwargs) -> bool():
r = self.req('/groups/%s' % name, method='DELETE', **kwargs)
return r.status_code == 204
# 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.
import unittest
from opencast_externalapi.apiclient import ApiClient
class TestApiClient(unittest.TestCase):
API_VERSION = 'v1.0.0'
def setUp(self):
self.api = ApiClient(host='http://localhost:8080') # TODO: CHANGE ME!
def test_req(self):
r = self.api.req('/')
data = r.json()
self.assertEqual(data['version'], self.API_VERSION)
self.assertIn(r.headers.get('content-type'),
['application/json', 'application/%s+json' % self.API_VERSION])
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()
# 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.
import unittest
from opencast_externalapi import ExternalApi
class TestGroups(unittest.TestCase):
API_VERSION = 'v1.0.0'
GROUP_NAME = 'test_group'
def setUp(self):
self.api = ExternalApi(host='http://localhost:8080') # TODO: CHANGE ME!
def test_get_groups(self):
groups = self.api.get_groups()
self.assertIsInstance(groups, list)
for group in groups:
self.assertIsNotNone(group.get('identifier'))
def test_get_groups_limit(self):
one_group = self.api.get_groups(limit=1)
self.assertEqual(len(one_group), 1)
self.assertIsInstance(one_group, list)
def test_get_group(self):
groups = self.api.get_groups()
group_id = groups[0].get('identifier')
group = self.api.get_group(group_id)
self.assertIsInstance(group, dict)
def test_create_group(self):
success = self.api.create_group(self.GROUP_NAME)
self.assertTrue(success)
self.api.delete_group(self.GROUP_NAME)
def tearDown(self):
pass
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