6693153155

Committed 30 Oct 2023 01:00PM UTC coverage: 81.021%. First build

Build # 6693153155

Build Type

Pull #687

github

Committed by

web-flow

Commit Message

Merge 2966ddad2 into 9e21b93d3

Pull Request Pull Request #687: fix tests

Run Details

5 of 5 new or added lines in 3 files covered. (100.0%)

5110 of 6307 relevant lines covered (81.02%)

0.81 hits per line

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

50.0

/pywps/processing/basic.py

##################################################################
# Copyright 2018 Open Source Geospatial Foundation and others    #
# licensed under MIT, Please consult LICENSE.txt for details     #
##################################################################
import os

from pywps.processing.job import Job


class Processing(object):
    """
    :class:`Processing` is an interface for running jobs.
    """

    def __init__(self, process, wps_request, wps_response):
        self.job = Job(process, wps_request, wps_response)

    def start(self):
        raise NotImplementedError("Needs to be implemented in subclass.")

    def cancel(self):
        raise NotImplementedError("Needs to be implemented in subclass.")


class MultiProcessing(Processing):
    """
    :class:`MultiProcessing` is the default implementation to run jobs using the
    :module:`multiprocessing` module.
    """

    def start(self):
        import multiprocessing
        process = multiprocessing.Process(
            target=getattr(self.job.process, self.job.method),
            args=(self.job.wps_request, self.job.wps_response)
        )
        process.start()


class DetachProcessing(Processing):
    """
    :class:`DetachProcessing` run job as detached process. The process will be run as child of pid 1
    """

    def start(self):
        pid = os.fork()
        if pid != 0:
            # Wait that the children get detached.
            os.waitpid(pid, 0)
            return

        # Detach ourself.

        # Ensure that we are the session leader to avoid to be zombified.
        os.setsid()
        if os.fork():
            # Stop running now
            os._exit(0)

        # We are the detached child, run the actual process
        try:
            getattr(self.job.process, self.job.method)(self.job.wps_request, self.job.wps_response)
        except Exception:
            pass
        # Ensure to stop ourself here what ever append.
        os._exit(0)

1	##################################################################
2	# Copyright 2018 Open Source Geospatial Foundation and others #
3	# licensed under MIT, Please consult LICENSE.txt for details #
4	##################################################################
5	import os	1✔
6
7	from pywps.processing.job import Job	1✔
8
9
10	class Processing(object):	1✔
11	"""
12	:class:`Processing` is an interface for running jobs.
13	"""
14
15	def __init__(self, process, wps_request, wps_response):	1✔
16	self.job = Job(process, wps_request, wps_response)	1✔
17
18	def start(self):	1✔
19	raise NotImplementedError("Needs to be implemented in subclass.")	×
20
21	def cancel(self):	1✔
22	raise NotImplementedError("Needs to be implemented in subclass.")	×
23
24
25	class MultiProcessing(Processing):	1✔
26	"""
27	:class:`MultiProcessing` is the default implementation to run jobs using the
28	:module:`multiprocessing` module.
29	"""
30
31	def start(self):	1✔
32	import multiprocessing	1✔
33	process = multiprocessing.Process(	1✔
34	target=getattr(self.job.process, self.job.method),
35	args=(self.job.wps_request, self.job.wps_response)
36	)
37	process.start()	1✔
38
39
40	class DetachProcessing(Processing):	1✔
41	"""
42	:class:`DetachProcessing` run job as detached process. The process will be run as child of pid 1
43	"""
44
45	def start(self):	1✔
46	pid = os.fork()	×
47	if pid != 0:	×
48	# Wait that the children get detached.
49	os.waitpid(pid, 0)	×
50	return	×
51
52	# Detach ourself.
53
54	# Ensure that we are the session leader to avoid to be zombified.
55	os.setsid()	×
56	if os.fork():	×
57	# Stop running now
58	os._exit(0)	×
59
60	# We are the detached child, run the actual process
61	try:	×
62	getattr(self.job.process, self.job.method)(self.job.wps_request, self.job.wps_response)	×
63	except Exception:	×
64	pass	×
65	# Ensure to stop ourself here what ever append.
66	os._exit(0)	×

geopython / pywps / 6693153155

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