Python CSC Electronic Office
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

73 lines
2.7 KiB

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']
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(
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: str, path: str, principal: str, delegate: bool, **kwargs):
# make sure that we're not already in a Flask context
assert not flask.has_app_context()
if principal is None:
principal = self.syscom_principal
headers = self.get_headers(principal, delegate)
resp =, method=method, headers=headers, **kwargs)
status = int(resp.status.split(' ', 1)[0])
if resp.headers['content-type'] == 'application/json':
data = json.loads(
data = [json.loads(line) for line in]
return status, data
def get(self, path, principal=None, delegate=True, **kwargs):
return self.request('GET', path, principal, delegate, **kwargs)
def post(self, path, principal=None, delegate=True, **kwargs):
return self.request('POST', path, principal, delegate, **kwargs)
def patch(self, path, principal=None, delegate=True, **kwargs):
return self.request('PATCH', path, principal, delegate, **kwargs)
def delete(self, path, principal=None, delegate=True, **kwargs):
return self.request('DELETE', path, principal, delegate, **kwargs)