7399750331

Committed 03 Jan 2024 04:21PM UTC coverage: 85.749% (+1.5%) from 84.257%

Build # 7399750331

Build Type

Pull #66

github

Committed by

web-flow

Commit Message

Merge branch 'main' into develop

Pull Request Pull Request #66: Develop 0.2.2 into main

Run Details

21 of 27 new or added lines in 4 files covered. (77.78%)

1 existing line in 1 file now uncovered.

355 of 414 relevant lines covered (85.75%)

5.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.83

/unity_sds_client/unity_session.py

import os
import getpass
import json
import requests

from configparser import ConfigParser
from unity_sds_client.unity_environments import UnityEnvironments
from unity_sds_client.unity_exception import UnityException



class UnitySession(object):
    """
    passable session object containing configuration, auth objects, and environment.
    """

    def __init__(self, env: UnityEnvironments, config: ConfigParser):
        """initialize the unitySession object which holds configuration and auth objects.

        Parameters
        ----------
        env : str
            The environment in use (e.g. dev, test, ops)
        config : type
            the configuration object for the python configuration file.

        Returns
        -------
        class
            Class declaration, returns the initialized UnitySession class.

        """
        self._env = env
        self._config = config

        # set up unity authentication
        self._auth = UnityAuth(self._config.get(env, "client_id"), self._config.get(env, "auth_endpoint"))
        self._unity_href =  self._config.get(env, "unity_href")

        self._project = None
        self._venue = None
        self._venue_id = None

    def get_service_endpoint(self, section, setting):
        """convenience method for getting a configured item from the included configuration.


        Parameters
        ----------
        section : str
            The section of the configuration to read.
        setting : str
            The item within a block to read.

        Returns
        -------
        str
            the configuration entry

        """
        return self._config.get(section.upper(), setting)

    def get_unity_href(self):
        """convenience method for getting the unity href.

                Parameters
                ----------
                none

                Returns
                -------
                str
                    the url to the unity top url

                """
        return self._unity_href


    def get_auth(self):
        """Returns the auth object in use by the session

        Returns
        -------
        UnityAuth
            The authentication object which allows access to a "token" for api calls.

        """
        return self._auth

    def get_config(self):
        """Returns the configuration being used by the session

        Returns
        -------
        Config
            The configuration object which contains configuration information used for api calls.
        """
        return self._config

    def get_venue_id(self):
        if self._venue_id is None:
            if self._project is None or self._venue is None:
                raise UnityException("session variables project and venue or venue_id are required to interact with a "
                                     "processing service.")
            else:
                return self._project + "/" + self._venue
        else:
            return self._venue_id

class UnityAuth(object):
    """
    Unity Auth object for handling cognito authentication on behalf of all service wrappers.
    """

    _token = None
    _token_expiration = None

    # The auth_json is template for authorizing with AWS Cognito for a token that can be used for calls to the
    # data service. For now this is just an empty data structure. You will be prompted for your username and password
    # in a few steps.
    auth_json = '''{
         "AuthParameters" : {
            "USERNAME" : "",
            "PASSWORD" : ""
         },
         "AuthFlow" : "USER_PASSWORD_AUTH",
         "ClientId" : ""
      }'''

    def __init__(self, client_id, auth_endpoint):
        """initialize the Unity Auth class. The initialization looks for username/passwords in the following locations:
        1. The UNITY_USER and UNITY_PASSWORD environment variables.
        2. Prompt a user if no any of the previous fail

        Parameters
        ----------
        client_id : str
            The client ID of the unity API Gateway in use.
        auth_endpoint : str
            the https:// endpoint for use in obtaining a credential

        Returns
        -------
        UnityAuth
            The unityAuth object which allows access to a token

        """
        self._client_id = client_id
        self._endpoint = auth_endpoint

        # order of operations:
        # environment
        # netrc? //todo
        self._user = os.getenv('UNITY_USER', None)
        self._password = os.getenv('UNITY_PASSWORD', None)

        if None in [self._user, self._password]:
            username = input('Please enter your Unity username: ')
            password = getpass.getpass("Please enter your Unity password: ")
            self._user = username
            self._password = password

    def get_token(self):
        """Public convenience method for getting a token. This begins the
        process of returning an already created token or will create a new token if necessary

        Returns
        -------
        str
            The token for use in API calls

        """
        # If the token doesn't exist or the token is expired, create a new one.
        if self._token is None or UnityAuth._is_expired(self._token_expiration):
            self._get_unity_token()

        return self._token

    def _is_expired(expiration_date):
        """Convenience method for checking if a token is expired. static method

        Parameters
        ----------
        expiration_date : datetime
            expiration date of the token

        Returns
        -------
        bool
            True if the token is expired, False if still valid

        """
        if expiration_date is None:
            return True
        else:
            # TODO
            return False

    def _get_unity_token(self):
        """Queries the backing service for a new API Token

        Returns
        -------
        str
            the token generated. Also stored in the UnityAuth._token field

        """
        aj = json.loads(self.auth_json)
        aj['AuthParameters']['USERNAME'] = self._user
        aj['AuthParameters']['PASSWORD'] = self._password
        aj['ClientId'] =self._client_id
        try:
            response = requests.post(self._endpoint, headers={"Content-Type":"application/x-amz-json-1.1", "X-Amz-Target":"AWSCognitoIdentityProviderService.InitiateAuth"}, json=aj)
            json_resp = response.json()
            self._token = json_resp['AuthenticationResult']['AccessToken']
            self._token_expiration = json_resp['AuthenticationResult']['AccessToken']
        except:
            print("Error, check username and password and try again.")
        return self._token

1	import os	6✔
2	import getpass	6✔
3	import json	6✔
4	import requests	6✔
5
6	from configparser import ConfigParser	6✔
7	from unity_sds_client.unity_environments import UnityEnvironments	6✔
8	from unity_sds_client.unity_exception import UnityException	6✔
9
10
11
12	class UnitySession(object):	6✔
13	"""
14	passable session object containing configuration, auth objects, and environment.
15	"""
16
17	def __init__(self, env: UnityEnvironments, config: ConfigParser):	6✔
18	"""initialize the unitySession object which holds configuration and auth objects.
19
20	Parameters
21	----------
22	env : str
23	The environment in use (e.g. dev, test, ops)
24	config : type
25	the configuration object for the python configuration file.
26
27	Returns
28	-------
29	class
30	Class declaration, returns the initialized UnitySession class.
31
32	"""
33	self._env = env	6✔
34	self._config = config	6✔
35
36	# set up unity authentication
37	self._auth = UnityAuth(self._config.get(env, "client_id"), self._config.get(env, "auth_endpoint"))	6✔
38	self._unity_href = self._config.get(env, "unity_href")	6✔
39
40	self._project = None	6✔
41	self._venue = None	6✔
42	self._venue_id = None	6✔
43
44	def get_service_endpoint(self, section, setting):	6✔
45	"""convenience method for getting a configured item from the included configuration.
46
47
48	Parameters
49	----------
50	section : str
51	The section of the configuration to read.
52	setting : str
53	The item within a block to read.
54
55	Returns
56	-------
57	str
58	the configuration entry
59
60	"""
UNCOV 61	return self._config.get(section.upper(), setting)	×
62
63	def get_unity_href(self):	6✔
64	"""convenience method for getting the unity href.
65
66	Parameters
67	----------
68	none
69
70	Returns
71	-------
72	str
73	the url to the unity top url
74
75	"""
76	return self._unity_href	6✔
77
78
79	def get_auth(self):	6✔
80	"""Returns the auth object in use by the session
81
82	Returns
83	-------
84	UnityAuth
85	The authentication object which allows access to a "token" for api calls.
86
87	"""
88	return self._auth	6✔
89
90	def get_config(self):	6✔
91	"""Returns the configuration being used by the session
92
93	Returns
94	-------
95	Config
96	The configuration object which contains configuration information used for api calls.
97	"""
98	return self._config	×
99
100	def get_venue_id(self):	6✔
101	if self._venue_id is None:	6✔
102	if self._project is None or self._venue is None:	6✔
103	raise UnityException("session variables project and venue or venue_id are required to interact with a "	6✔
NEW 104	"processing service.")	×
NEW 105	else:	×
NEW 106	return self._project + "/" + self._venue	×
NEW 107	else:	×
108	return self._venue_id	6✔
109
110	class UnityAuth(object):	6✔
111	"""
112	Unity Auth object for handling cognito authentication on behalf of all service wrappers.
113	"""
114
115	_token = None	6✔
116	_token_expiration = None	6✔
117
118	# The auth_json is template for authorizing with AWS Cognito for a token that can be used for calls to the
119	# data service. For now this is just an empty data structure. You will be prompted for your username and password
120	# in a few steps.
121	auth_json = '''{	6✔
122	"AuthParameters" : {
123	"USERNAME" : "",
124	"PASSWORD" : ""
125	},
126	"AuthFlow" : "USER_PASSWORD_AUTH",
127	"ClientId" : ""
128	}'''
129
130	def __init__(self, client_id, auth_endpoint):	6✔
131	"""initialize the Unity Auth class. The initialization looks for username/passwords in the following locations:
132	1. The UNITY_USER and UNITY_PASSWORD environment variables.
133	2. Prompt a user if no any of the previous fail
134
135	Parameters
136	----------
137	client_id : str
138	The client ID of the unity API Gateway in use.
139	auth_endpoint : str
140	the https:// endpoint for use in obtaining a credential
141
142	Returns
143	-------
144	UnityAuth
145	The unityAuth object which allows access to a token
146
147	"""
148	self._client_id = client_id	6✔
149	self._endpoint = auth_endpoint	6✔
150
151	# order of operations:
152	# environment
153	# netrc? //todo
154	self._user = os.getenv('UNITY_USER', None)	6✔
155	self._password = os.getenv('UNITY_PASSWORD', None)	6✔
156
157	if None in [self._user, self._password]:	6✔
158	username = input('Please enter your Unity username: ')
159	password = getpass.getpass("Please enter your Unity password: ")
160	self._user = username
161	self._password = password
162
163	def get_token(self):	6✔
164	"""Public convenience method for getting a token. This begins the
165	process of returning an already created token or will create a new token if necessary
166
167	Returns
168	-------
169	str
170	The token for use in API calls
171
172	"""
173	# If the token doesn't exist or the token is expired, create a new one.
174	if self._token is None or UnityAuth._is_expired(self._token_expiration):	6✔
175	self._get_unity_token()	6✔
176
177	return self._token	6✔
178
179	def _is_expired(expiration_date):	6✔
180	"""Convenience method for checking if a token is expired. static method
181
182	Parameters
183	----------
184	expiration_date : datetime
185	expiration date of the token
186
187	Returns
188	-------
189	bool
190	True if the token is expired, False if still valid
191
192	"""
193	if expiration_date is None:
194	return True
195	else:
196	# TODO
197	return False
198
199	def _get_unity_token(self):	6✔
200	"""Queries the backing service for a new API Token
201
202	Returns
203	-------
204	str
205	the token generated. Also stored in the UnityAuth._token field
206
207	"""
208	aj = json.loads(self.auth_json)	6✔
209	aj['AuthParameters']['USERNAME'] = self._user	6✔
210	aj['AuthParameters']['PASSWORD'] = self._password	6✔
211	aj['ClientId'] =self._client_id	6✔
212	try:	6✔
213	response = requests.post(self._endpoint, headers={"Content-Type":"application/x-amz-json-1.1", "X-Amz-Target":"AWSCognitoIdentityProviderService.InitiateAuth"}, json=aj)	6✔
214	json_resp = response.json()	6✔
215	self._token = json_resp['AuthenticationResult']['AccessToken']	6✔
216	self._token_expiration = json_resp['AuthenticationResult']['AccessToken']	6✔
217	except:
218	print("Error, check username and password and try again.")
219	return self._token	6✔

unity-sds / unity-py / 7399750331

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous