67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
|
import json
|
||
|
|
import socket
|
||
|
|
|
||
|
|
from flask.testing import FlaskClient
|
||
|
|
import gssapi
|
||
|
|
import pytest
|
||
|
|
from requests import Request
|
||
|
|
from requests_gssapi import HTTPSPNEGOAuth
|
||
|
|
from zope import component
|
||
|
|
|
||
|
|
from ceo_common.interfaces import IConfig
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope='session')
|
||
|
|
def client(app):
|
||
|
|
app_client = app.test_client()
|
||
|
|
return CeodTestClient(app_client)
|
||
|
|
|
||
|
|
|
||
|
|
class CeodTestClient:
|
||
|
|
def __init__(self, app_client: FlaskClient):
|
||
|
|
cfg = component.getUtility(IConfig)
|
||
|
|
self.client = app_client
|
||
|
|
self.admin_principal = cfg.get('ldap_admin_principal')
|
||
|
|
# this is only used for the HTTPSNEGOAuth
|
||
|
|
self.base_url = f'http://{socket.getfqdn()}'
|
||
|
|
self.cached_auth = {}
|
||
|
|
|
||
|
|
def get_auth(self, principal):
|
||
|
|
if principal in self.cached_auth:
|
||
|
|
return self.cached_auth[principal]
|
||
|
|
name = gssapi.Name(principal)
|
||
|
|
creds = gssapi.Credentials(name=name, usage='initiate')
|
||
|
|
auth = HTTPSPNEGOAuth(
|
||
|
|
opportunistic_auth=True,
|
||
|
|
target_name='ceod',
|
||
|
|
creds=creds,
|
||
|
|
)
|
||
|
|
self.cached_auth[principal] = auth
|
||
|
|
return auth
|
||
|
|
|
||
|
|
def get_headers(self, principal):
|
||
|
|
# method doesn't matter here because we just need the headers
|
||
|
|
req = Request('GET', self.base_url, auth=self.get_auth(principal))
|
||
|
|
return req.prepare().headers.items()
|
||
|
|
|
||
|
|
def request(self, method, path, principal, **kwargs):
|
||
|
|
if principal is None:
|
||
|
|
principal = self.admin_principal
|
||
|
|
resp = self.client.open(
|
||
|
|
path, method=method, headers=self.get_headers(principal), **kwargs)
|
||
|
|
status = int(resp.status.split(' ', 1)[0])
|
||
|
|
try:
|
||
|
|
data = json.loads(resp.data)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
data = resp.data.decode()
|
||
|
|
return status, data
|
||
|
|
|
||
|
|
def get(self, path, principal=None, **kwargs):
|
||
|
|
return self.request('GET', path, principal, **kwargs)
|
||
|
|
|
||
|
|
def post(self, path, principal=None, **kwargs):
|
||
|
|
return self.request('POST', path, principal, **kwargs)
|
||
|
|
|
||
|
|
def delete(self, path, principal=None, **kwargs):
|
||
|
|
return self.request('DELETE', path, principal, **kwargs)
|