9782245928

Committed 03 Jul 2024 05:22PM UTC coverage: 46.923% (-40.7%) from 87.575%

Build # 9782245928

Build Type

Pull #92

github

Committed by

web-flow

Commit Message

Merge cbea3871d into d40a2f637

Pull Request Pull Request #92: Ogc processes api

Run Details

56 of 165 new or added lines in 7 files covered. (33.94%)

80 existing lines in 8 files now uncovered.

366 of 780 relevant lines covered (46.92%)

2.82 hits per line

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

36.59

/unity_sds_client/resources/process.py

import traceback

import requests
from unity_sps_ogc_processes_api_python_client import Execute
from unity_sps_ogc_processes_api_python_client.exceptions import NotFoundException

from unity_sds_client.resources.job_status import JobStatus
from unity_sds_client.unity_exception import UnityException
from unity_sds_client.unity_session import UnitySession
from unity_sds_client.resources.job import Job
import unity_sps_ogc_processes_api_python_client
from unity_sps_ogc_processes_api_python_client.rest import ApiException
from unity_sps_ogc_processes_api_python_client.models.process_list import ProcessList


class Process(object):

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

    def __init__(
            self,
            session: UnitySession,
            endpoint: str,
            process_id: str,
            title: str,
            abstract: str,
            process_version: str,
            keywords: str,
            job_control_options: list,
            inputs: object = None,
            outputs: object = None,
    ):
        """
        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 = process_id
        self.title = title
        self.job_control_options = job_control_options
        self.keywords = keywords
        self.process_version = process_version
        self.abstract = abstract
        self.inputs = inputs
        self.outputs = outputs

    def execute(self, data) -> Job:
        '''

        @param data:
        @return Job:
        '''
        token = self._session.get_auth().get_token()
        url = self._endpoint
        execution = Execute.from_dict(data)
        configuration = unity_sps_ogc_processes_api_python_client.Configuration(
            host=url,
            access_token=token
        )
        with unity_sps_ogc_processes_api_python_client.ApiClient(configuration) as api_client:
            # Create an instance of the API class
            api_instance = unity_sps_ogc_processes_api_python_client.DefaultApi(api_client)
            try:
                # Retrieve the list of available processes
                job = api_instance.execute_processes_process_id_execution_post(self.id, execution)  # add auth
                return Job(
                            self._session,
                            self._endpoint,
                            job.process_id,
                            job.job_id,
                            JobStatus(job.status.value)
                    )

            except NotFoundException as nfe:
                raise UnityException(nfe.body)
            except Exception as e:
                print(traceback.format_exc())
                print("Exception when calling DefaultApi->process_list_processes_get: %s\n" % e)
                raise UnityException(e.body)


1	import traceback	6✔
2
3	import requests	6✔
4	from unity_sps_ogc_processes_api_python_client import Execute	6✔
5	from unity_sps_ogc_processes_api_python_client.exceptions import NotFoundException	6✔
6
7	from unity_sds_client.resources.job_status import JobStatus	6✔
8	from unity_sds_client.unity_exception import UnityException	6✔
9	from unity_sds_client.unity_session import UnitySession	6✔
10	from unity_sds_client.resources.job import Job	6✔
11	import unity_sps_ogc_processes_api_python_client	6✔
12	from unity_sps_ogc_processes_api_python_client.rest import ApiException	6✔
13	from unity_sps_ogc_processes_api_python_client.models.process_list import ProcessList	6✔
14
15
16	class Process(object):	6✔
17
18	def __str__(self):	6✔
UNCOV 19	return '''unity_sds_client.resources.Process(	×
20	id="{}",
21	process_version="{}"
22	title="{}",
23	abstract="{}",
24	keywords="{}"
25	)'''.format(
26	self.id,
27	self.process_version,
28	self.title,
29	self.abstract,
30	self.keywords
31	)
32
33	def __init__(	6✔
34	self,
35	session: UnitySession,
36	endpoint: str,
37	process_id: str,
38	title: str,
39	abstract: str,
40	process_version: str,
41	keywords: str,
42	job_control_options: list,
43	inputs: object = None,
44	outputs: object = None,
45	):
46	"""
47	Initialize the Process class.
48
49	Parameters
50	----------
51	session : UnitySession
52	The Unity Session that will be used to facilitate making calls to the SPS endpoints.
53	endpoint : str
54	The endpoint to call for executing processes
55
56	Returns
57	-------
58	Process
59	The Process object.
60
61	"""
UNCOV 62	self._session = session	×
UNCOV 63	self._endpoint = endpoint	×
NEW 64	self.id = process_id	×
UNCOV 65	self.title = title	×
UNCOV 66	self.job_control_options = job_control_options	×
UNCOV 67	self.keywords = keywords	×
UNCOV 68	self.process_version = process_version	×
NEW 69	self.abstract = abstract	×
NEW 70	self.inputs = inputs	×
NEW 71	self.outputs = outputs	×
72
73	def execute(self, data) -> Job:	6✔
74	'''
75
76	@param data:
77	@return Job:
78	'''
UNCOV 79	token = self._session.get_auth().get_token()	×
NEW 80	url = self._endpoint	×
NEW 81	execution = Execute.from_dict(data)	×
NEW 82	configuration = unity_sps_ogc_processes_api_python_client.Configuration(	×
83	host=url,
84	access_token=token
85	)
NEW 86	with unity_sps_ogc_processes_api_python_client.ApiClient(configuration) as api_client:	×
87	# Create an instance of the API class
NEW 88	api_instance = unity_sps_ogc_processes_api_python_client.DefaultApi(api_client)	×
NEW 89	try:	×
90	# Retrieve the list of available processes
NEW 91	job = api_instance.execute_processes_process_id_execution_post(self.id, execution) # add auth	×
NEW 92	return Job(	×
93	self._session,
94	self._endpoint,
95	job.process_id,
96	job.job_id,
97	JobStatus(job.status.value)
98	)
99
NEW 100	except NotFoundException as nfe:	×
NEW 101	raise UnityException(nfe.body)	×
NEW 102	except Exception as e:	×
NEW 103	print(traceback.format_exc())	×
NEW 104	print("Exception when calling DefaultApi->process_list_processes_get: %s\n" % e)	×
NEW 105	raise UnityException(e.body)	×
106

unity-sds / unity-py / 9782245928

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