pyceo/tests/conftest_ceod_api.py

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)