9781682280

Committed 03 Jul 2024 04:38PM UTC coverage: 52.801% (-34.8%) from 87.575%

Build # 9781682280

Build Type

push

github

Committed by

web-flow

Commit Message

Add health service (#90)

* Created health service client that uses a stubbed set of health data. To view health status information, the health status information is included when a user prints an instantiated unity object.

* Added print health status method to health service and updated how health statuses are printed to account for service's historical health status information.

* Removed uneeded health service unit test fixture.

* Updated how health status reports are generated so that the report can be reused in multiple places to improve consistency of the health status report.

* Updated CHANGELOG and version number.

* Testing different syntax to resolve flake8 linting error.

* Updated dictionary references to resolve flake8 linting errors.

* Update python-app.yml

* Update python-app.yml

* Update python-app.yml

* Update python-app.yml

* Update python-app.yml

* Update python-app.yml

* Update python-app.yml

* updating workflows

---------

Co-authored-by: mike-gangl <59702631+mike-gangl@users.noreply.github.com>
Co-authored-by: mike-gangl <michael.e.gangl@jpl.nasa.gov>

Run Details

32 of 36 new or added lines in 3 files covered. (88.89%)

85 existing lines in 8 files now uncovered.

377 of 714 relevant lines covered (52.8%)

3.17 hits per line

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

65.63

/unity_sds_client/resources/process.py

import requests
from unity_sds_client.unity_session import UnitySession
from unity_sds_client.resources.job import Job
from unity_sds_client.utils.http import get_headers

class Process(object):

    def __str__(self):
        return '''unity_sds_client.resources.Process(
    id="{}",
    title="{}",
    abstract="{}",
    execution_unit="{}",
    immediate_deployment={},
    job_control_options={},
    keywords="{}",
    output_transmission={},
    ows_context_url="{}",
    process_version="{}"
)'''.format(
    self.id,
    self.title,
    self.abstract,
    self.execution_unit,
    self.immediate_deployment,
    self.job_control_options,
    self.keywords,
    self.output_transmission,
    self.ows_context_url,
    self.process_version
)

    def __init__(
        self,
        session:UnitySession,
        endpoint:str,
        id:str,
        title:str,
        abstract:str,
        execution_unit:str,
        immediate_deployment:bool,
        job_control_options:list,
        keywords:str,
        output_transmission:list,
        ows_context_url:str,
        process_version:str
    ):
        """
        Initialize the Process class.

        Parameters
        ----------
        session : UnitySession
            The Unity Session that will be used to facilitate making calls to the SPS endpoints.
        endpoint : str
            The endpoint to call for executing processes

        Returns
        -------
        Process
            The Process object.

        """

        self._session = session
        self._endpoint = endpoint
        self.id = id
        self.title = title
        self.abstract = abstract
        self.execution_unit = execution_unit
        self.immediate_deployment = immediate_deployment
        self.job_control_options = job_control_options
        self.keywords = keywords
        self.output_transmission = output_transmission
        self.ows_context_url = ows_context_url
        self.process_version = process_version

    def execute(self, data) -> Job:

        token = self._session.get_auth().get_token()
        headers = get_headers(token, {
            'Content-type': 'application/json'
        })
        url = self._endpoint + "processes/{}/jobs".format(self.id)
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()

        # Parse the job_id from the returned 'location' header
        job_location = response.headers['location']
        if "http://127.0.0.1:5000" in job_location:
            job_location = job_location.replace("http://127.0.0.1:5000/", self._endpoint)
        job_id = job_location.replace(url + "/","")

        job = Job(self._session, self._endpoint, self, job_id)

        return job

1	import requests	6✔
2	from unity_sds_client.unity_session import UnitySession	6✔
3	from unity_sds_client.resources.job import Job	6✔
4	from unity_sds_client.utils.http import get_headers	6✔
5
6	class Process(object):	6✔
7
8	def __str__(self):	6✔
9	return '''unity_sds_client.resources.Process(	6✔
10	id="{}",
11	title="{}",
12	abstract="{}",
13	execution_unit="{}",
14	immediate_deployment={},
15	job_control_options={},
16	keywords="{}",
17	output_transmission={},
18	ows_context_url="{}",
19	process_version="{}"
20	)'''.format(
21	self.id,
22	self.title,
23	self.abstract,
24	self.execution_unit,
25	self.immediate_deployment,
26	self.job_control_options,
27	self.keywords,
28	self.output_transmission,
29	self.ows_context_url,
30	self.process_version
31	)
32
33	def __init__(	6✔
34	self,
35	session:UnitySession,
36	endpoint:str,
37	id:str,
38	title:str,
39	abstract:str,
40	execution_unit:str,
41	immediate_deployment:bool,
42	job_control_options:list,
43	keywords:str,
44	output_transmission:list,
45	ows_context_url:str,
46	process_version:str
47	):
48	"""
49	Initialize the Process class.
50
51	Parameters
52	----------
53	session : UnitySession
54	The Unity Session that will be used to facilitate making calls to the SPS endpoints.
55	endpoint : str
56	The endpoint to call for executing processes
57
58	Returns
59	-------
60	Process
61	The Process object.
62
63	"""
64
65	self._session = session	6✔
66	self._endpoint = endpoint	6✔
67	self.id = id	6✔
68	self.title = title	6✔
69	self.abstract = abstract	6✔
70	self.execution_unit = execution_unit	6✔
71	self.immediate_deployment = immediate_deployment	6✔
72	self.job_control_options = job_control_options	6✔
73	self.keywords = keywords	6✔
74	self.output_transmission = output_transmission	6✔
75	self.ows_context_url = ows_context_url	6✔
76	self.process_version = process_version	6✔
77
78	def execute(self, data) -> Job:	6✔
79
UNCOV 80	token = self._session.get_auth().get_token()	×
UNCOV 81	headers = get_headers(token, {	×
82	'Content-type': 'application/json'
83	})
UNCOV 84	url = self._endpoint + "processes/{}/jobs".format(self.id)	×
UNCOV 85	response = requests.post(url, headers=headers, json=data)	×
UNCOV 86	response.raise_for_status()	×
87
88	# Parse the job_id from the returned 'location' header
UNCOV 89	job_location = response.headers['location']	×
UNCOV 90	if "http://127.0.0.1:5000" in job_location:	×
UNCOV 91	job_location = job_location.replace("http://127.0.0.1:5000/", self._endpoint)	×
UNCOV 92	job_id = job_location.replace(url + "/","")	×
93
UNCOV 94	job = Job(self._session, self._endpoint, self, job_id)	×
95
UNCOV 96	return job	×

unity-sds / unity-py / 9781682280

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