import json import socket import flask from flask.testing import FlaskClient import gssapi import pytest from requests import Request from requests_gssapi import HTTPSPNEGOAuth from .utils import gssapi_token_ctx __all__ = ['client'] @pytest.fixture(scope='session') def client(app): app_client = app.test_client() yield CeodTestClient(app_client) class CeodTestClient: def __init__(self, app_client: FlaskClient): self.client = app_client self.syscom_principal = 'ctdalek' # this is only used for the HTTPSNEGOAuth self.base_url = f'http://{socket.getfqdn()}' # for SPNEGO self.target_name = gssapi.Name('ceod/' + socket.getfqdn()) def get_auth(self, principal: str, delegate: bool): """Acquire a HTTPSPNEGOAuth instance for the principal.""" with gssapi_token_ctx(principal) as token: return HTTPSPNEGOAuth( opportunistic_auth=True, target_name=self.target_name, creds=gssapi.Credentials(token=token), delegate=delegate, ) def get_headers(self, principal: str, delegate: bool): # Get the Authorization header (SPNEGO). # The method doesn't matter here because we just need to extract # the header using req.prepare(). req = Request('GET', self.base_url, auth=self.get_auth(principal, delegate)) headers = list(req.prepare().headers.items()) return headers def request(self, method, path, principal, need_auth, delegate, **kwargs): # make sure that we're not already in a Flask context assert not flask.has_app_context() if need_auth: principal = principal or self.syscom_principal headers = self.get_headers(principal, delegate) else: headers = [] resp = self.client.open(path, method=method, headers=headers, **kwargs) status = int(resp.status.split(' ', 1)[0]) if resp.headers['content-type'] == 'application/json': data = json.loads(resp.data) else: data = [json.loads(line) for line in resp.data.splitlines()] return status, data def get(self, path, principal=None, need_auth=True, delegate=True, **kwargs): return self.request('GET', path, principal, need_auth, delegate, **kwargs) def post(self, path, principal=None, need_auth=True, delegate=True, **kwargs): return self.request('POST', path, principal, need_auth, delegate, **kwargs) def patch(self, path, principal=None, need_auth=True, delegate=True, **kwargs): return self.request('PATCH', path, principal, need_auth, delegate, **kwargs) def delete(self, path, principal=None, need_auth=True, delegate=True, **kwargs): return self.request('DELETE', path, principal, need_auth, delegate, **kwargs)