integ_test_base.IntegTestBase._get_process() - Code Metrics - Inspection of "v 0.8.9 (#353)" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 305552...00a4d3 )

by Oleksandr

created 2019-10-22 00:09 UTC

integ_test_base.IntegTestBase._get_process() A

↳ Parent: integ_test_base

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	2
rs	10
c	0
b	0
f	0
cc	1
nop	1

import coverage
import http.client
import os
from pathlib import Path
import platform
import shutil
import signal
import subprocess
import tabpy
import tempfile
import time
import unittest


class IntegTestBase(unittest.TestCase):
    '''
    Base class for integration tests.
    '''

    def __init__(self, methodName="runTest"):
        super(IntegTestBase, self).__init__(methodName)
        self.process = None
        self.delete_temp_folder = True

    def set_delete_temp_folder(self, delete_temp_folder: bool):
        '''
        Specify if temporary folder for state, config and log
        files should be deleted when test is done.
        By default the folder is deleted.

        Parameters
        ----------
        delete_test_folder: bool
            If True temp folder will be deleted.
        '''
        self.delete_temp_folder = delete_temp_folder

    def _get_state_file_path(self) -> str:
        '''
        Generates state.ini and returns absolute path to it.
        Overwrite this function for tests to run against not default state
        file.

        Returns
        -------
        str
            Absolute path to state file folder.
        '''
        state_file = open(os.path.join(self.tmp_dir, 'state.ini'), 'w+')
        state_file.write(
            '[Service Info]\n'
            'Name = TabPy Serve\n'
            'Description = \n'
            'Creation Time = 0\n'
            'Access-Control-Allow-Origin = \n'
            'Access-Control-Allow-Headers = \n'
            'Access-Control-Allow-Methods = \n'
            '\n'
            '[Query Objects Service Versions]\n'
            '\n'
            '[Query Objects Docstrings]\n'
            '\n'
            '[Meta]\n'
            'Revision Number = 1\n')
        state_file.close()

        return self.tmp_dir

    def _get_port(self) -> str:
        '''
        Returns port TabPy should run on. Default implementation
        returns '9004'.

        Returns
        -------
        str
            Port number.
        '''
        return '9004'

    def _get_pwd_file(self) -> str:
        '''
        Returns absolute or relative path to password file.
        Overwrite to create and/or specify your own file.
        Default implementation returns None which means
        TABPY_PWD_FILE setting won't be added to config.

        Returns
        -------
        str
            Absolute or relative path to password file.
            If None TABPY_PWD_FILE setting won't be added to
            config.
        '''
        return None

    def _get_transfer_protocol(self) -> str:
        '''
        Returns transfer protocol for configuration file.
        Default implementation returns None which means
        TABPY_TRANSFER_PROTOCOL setting won't be added to config.

        Returns
        -------
        str
            Transfer protocol (e.g 'http' or 'https').
            If None TABPY_TRANSFER_PROTOCOL setting won't be
            added to config.
        '''
        return None

    def _get_certificate_file_name(self) -> str:
        '''
        Returns absolute or relative certificate file name
        for configuration file.
        Default implementation returns None which means
        TABPY_CERTIFICATE_FILE setting won't be added to config.

        Returns
        -------
        str
            Absolute or relative certificate file name.
            If None TABPY_CERTIFICATE_FILE setting won't be
            added to config.
        '''
        return None

    def _get_key_file_name(self) -> str:
        '''
        Returns absolute or relative private key file name
        for configuration file.
        Default implementation returns None which means
        TABPY_KEY_FILE setting won't be added to config.

        Returns
        -------
        str
            Absolute or relative private key file name.
            If None TABPY_KEY_FILE setting won't be
            added to config.
        '''
        return None

    def _get_evaluate_timeout(self) -> str:
        '''
        Returns the configured timeout for the /evaluate method.
        Default implementation returns None, which means that
        the timeout will default to 30.

        Returns
        -------
        str
            Timeout for calling /evaluate.
            If None, defaults TABPY_EVALUATE_TIMEOUT setting
            will default to '30'.
        '''
        return None

    def _get_config_file_name(self) -> str:
        '''
        Generates config file. Overwrite this function for tests to
        run against not default state file.

        Returns
        -------
        str
            Absolute path to config file.
        '''
        config_file = open(os.path.join(self.tmp_dir, 'test.conf'), 'w+')
        config_file.write(
            '[TabPy]\n'
            f'TABPY_QUERY_OBJECT_PATH = ./query_objects\n'
            f'TABPY_PORT = {self._get_port()}\n'
            f'TABPY_STATE_PATH = {self.tmp_dir}\n')

        pwd_file = self._get_pwd_file()
        if pwd_file is not None:
            pwd_file = os.path.abspath(pwd_file)
            config_file.write(f'TABPY_PWD_FILE = {pwd_file}\n')

        transfer_protocol = self._get_transfer_protocol()
        if transfer_protocol is not None:
            config_file.write(
                f'TABPY_TRANSFER_PROTOCOL = {transfer_protocol}\n')

        cert_file_name = self._get_certificate_file_name()
        if cert_file_name is not None:
            cert_file_name = os.path.abspath(cert_file_name)
            config_file.write(f'TABPY_CERTIFICATE_FILE = {cert_file_name}\n')

        key_file_name = self._get_key_file_name()
        if key_file_name is not None:
            key_file_name = os.path.abspath(key_file_name)
            config_file.write(f'TABPY_KEY_FILE = {key_file_name}\n')

        evaluate_timeout = self._get_evaluate_timeout()
        if evaluate_timeout is not None:
            config_file.write(f'TABPY_EVALUATE_TIMEOUT = {evaluate_timeout}\n')

        config_file.close()

        self.delete_config_file = True
        return config_file.name

    def setUp(self):
        super(IntegTestBase, self).setUp()
        prefix = 'TabPy_IntegTest_'
        self.tmp_dir = tempfile.mkdtemp(prefix=prefix)

        # create temporary state.ini
        orig_state_file_name = os.path.abspath(
            self._get_state_file_path() + '/state.ini')
        self.state_file_name = os.path.abspath(self.tmp_dir + '/state.ini')
        if orig_state_file_name != self.state_file_name:
            shutil.copyfile(orig_state_file_name, self.state_file_name)

        # create config file
        orig_config_file_name = os.path.abspath(self._get_config_file_name())
        self.config_file_name = os.path.abspath(
            self.tmp_dir + '/' +
            os.path.basename(orig_config_file_name))
        if orig_config_file_name != self.config_file_name:
            shutil.copyfile(orig_config_file_name, self.config_file_name)

        # Platform specific - for integration tests we want to engage
        # startup script
        with open(self.tmp_dir + '/output.txt', 'w') as outfile:
            cmd = ['tabpy',
                   '--config=' + self.config_file_name]
            preexec_fn = None
            if platform.system() == 'Windows':
                self.py = 'python'
            else:
                self.py = 'python3'
                preexec_fn = os.setsid

            coverage.process_startup()
            self.process = subprocess.Popen(
                cmd,
                preexec_fn=preexec_fn,
                stdout=outfile,
                stderr=outfile)

            # give the app some time to start up...
            time.sleep(5)

    def tearDown(self):
        # stop TabPy
        if self.process is not None:
            if platform.system() == 'Windows':
                subprocess.call(['taskkill', '/F', '/T', '/PID',
                                 str(self.process.pid)])
            else:
                os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
            self.process.kill()

            # after shutting down TabPy and before we start it again
            # for next test give it some time to terminate.
            time.sleep(5)

        # remove temporary files
        if self.delete_temp_folder:
            os.remove(self.state_file_name)
            os.remove(self.config_file_name)
            shutil.rmtree(self.tmp_dir)

        super(IntegTestBase, self).tearDown()

    def _get_connection(self) -> http.client.HTTPConnection:
        protocol = self._get_transfer_protocol()
        url = 'localhost:' + self._get_port()

        if protocol is not None and protocol.lower() == 'https':
            connection = http.client.HTTPSConnection(url)
        else:
            connection = http.client.HTTPConnection(url)

        return connection

    def _get_username(self) -> str:
        return 'user1'

    def _get_password(self) -> str:
        return 'P@ssw0rd'

    def deploy_models(self, username: str, password: str):
        repo_dir = os.path.abspath(os.path.dirname(tabpy.__file__))
        path = os.path.join(repo_dir, 'models', 'deploy_models.py')
        with open(self.tmp_dir + '/deploy_models_output.txt', 'w') as outfile:
            outfile.write(
                f'--<< Running {self.py} {path} '
                f'{self._get_config_file_name()} >>--\n')
            input_string = f'{username}\n{password}\n'
            outfile.write(f'--<< Input = {input_string} >>--')
            coverage.process_startup()
            p = subprocess.run(
                [self.py, path, self._get_config_file_name()],
                input=input_string.encode('utf-8'),
                stdout=outfile,
                stderr=outfile)

    def _get_process(self):
        return self.process


1			import coverage
2			import http.client
3			import os
4			from pathlib import Path
5			import platform
6			import shutil
7			import signal
8			import subprocess
9			import tabpy
10			import tempfile
11			import time
12			import unittest
13
14
15			class IntegTestBase(unittest.TestCase):
16			'''
17			Base class for integration tests.
18			'''
19
20			def __init__(self, methodName="runTest"):
21			super(IntegTestBase, self).__init__(methodName)
22			self.process = None
23			self.delete_temp_folder = True
24
25			def set_delete_temp_folder(self, delete_temp_folder: bool):
26			'''
27			Specify if temporary folder for state, config and log
28			files should be deleted when test is done.
29			By default the folder is deleted.
30
31			Parameters
32			----------
33			delete_test_folder: bool
34			If True temp folder will be deleted.
35			'''
36			self.delete_temp_folder = delete_temp_folder
37
38			def _get_state_file_path(self) -> str:
39			'''
40			Generates state.ini and returns absolute path to it.
41			Overwrite this function for tests to run against not default state
42			file.
43
44			Returns
45			-------
46			str
47			Absolute path to state file folder.
48			'''
49			state_file = open(os.path.join(self.tmp_dir, 'state.ini'), 'w+')
50			state_file.write(
51			'[Service Info]\n'
52			'Name = TabPy Serve\n'
53			'Description = \n'
54			'Creation Time = 0\n'
55			'Access-Control-Allow-Origin = \n'
56			'Access-Control-Allow-Headers = \n'
57			'Access-Control-Allow-Methods = \n'
58			'\n'
59			'[Query Objects Service Versions]\n'
60			'\n'
61			'[Query Objects Docstrings]\n'
62			'\n'
63			'[Meta]\n'
64			'Revision Number = 1\n')
65			state_file.close()
66
67			return self.tmp_dir
68
69			def _get_port(self) -> str:
70			'''
71			Returns port TabPy should run on. Default implementation
72			returns '9004'.
73
74			Returns
75			-------
76			str
77			Port number.
78			'''
79			return '9004'
80
81			def _get_pwd_file(self) -> str:
82			'''
83			Returns absolute or relative path to password file.
84			Overwrite to create and/or specify your own file.
85			Default implementation returns None which means
86			TABPY_PWD_FILE setting won't be added to config.
87
88			Returns
89			-------
90			str
91			Absolute or relative path to password file.
92			If None TABPY_PWD_FILE setting won't be added to
93			config.
94			'''
95			return None
96
97			def _get_transfer_protocol(self) -> str:
98			'''
99			Returns transfer protocol for configuration file.
100			Default implementation returns None which means
101			TABPY_TRANSFER_PROTOCOL setting won't be added to config.
102
103			Returns
104			-------
105			str
106			Transfer protocol (e.g 'http' or 'https').
107			If None TABPY_TRANSFER_PROTOCOL setting won't be
108			added to config.
109			'''
110			return None
111
112			def _get_certificate_file_name(self) -> str:
113			'''
114			Returns absolute or relative certificate file name
115			for configuration file.
116			Default implementation returns None which means
117			TABPY_CERTIFICATE_FILE setting won't be added to config.
118
119			Returns
120			-------
121			str
122			Absolute or relative certificate file name.
123			If None TABPY_CERTIFICATE_FILE setting won't be
124			added to config.
125			'''
126			return None
127
128			def _get_key_file_name(self) -> str:
129			'''
130			Returns absolute or relative private key file name
131			for configuration file.
132			Default implementation returns None which means
133			TABPY_KEY_FILE setting won't be added to config.
134
135			Returns
136			-------
137			str
138			Absolute or relative private key file name.
139			If None TABPY_KEY_FILE setting won't be
140			added to config.
141			'''
142			return None
143
144			def _get_evaluate_timeout(self) -> str:
145			'''
146			Returns the configured timeout for the /evaluate method.
147			Default implementation returns None, which means that
148			the timeout will default to 30.
149
150			Returns
151			-------
152			str
153			Timeout for calling /evaluate.
154			If None, defaults TABPY_EVALUATE_TIMEOUT setting
155			will default to '30'.
156			'''
157			return None
158
159			def _get_config_file_name(self) -> str:
160			'''
161			Generates config file. Overwrite this function for tests to
162			run against not default state file.
163
164			Returns
165			-------
166			str
167			Absolute path to config file.
168			'''
169			config_file = open(os.path.join(self.tmp_dir, 'test.conf'), 'w+')
170			config_file.write(
171			'[TabPy]\n'
172			f'TABPY_QUERY_OBJECT_PATH = ./query_objects\n'
173			f'TABPY_PORT = {self._get_port()}\n'
174			f'TABPY_STATE_PATH = {self.tmp_dir}\n')
175
176			pwd_file = self._get_pwd_file()
177			if pwd_file is not None:
178			pwd_file = os.path.abspath(pwd_file)
179			config_file.write(f'TABPY_PWD_FILE = {pwd_file}\n')
180
181			transfer_protocol = self._get_transfer_protocol()
182			if transfer_protocol is not None:
183			config_file.write(
184			f'TABPY_TRANSFER_PROTOCOL = {transfer_protocol}\n')
185
186			cert_file_name = self._get_certificate_file_name()
187			if cert_file_name is not None:
188			cert_file_name = os.path.abspath(cert_file_name)
189			config_file.write(f'TABPY_CERTIFICATE_FILE = {cert_file_name}\n')
190
191			key_file_name = self._get_key_file_name()
192			if key_file_name is not None:
193			key_file_name = os.path.abspath(key_file_name)
194			config_file.write(f'TABPY_KEY_FILE = {key_file_name}\n')
195
196			evaluate_timeout = self._get_evaluate_timeout()
197			if evaluate_timeout is not None:
198			config_file.write(f'TABPY_EVALUATE_TIMEOUT = {evaluate_timeout}\n')
199
200			config_file.close()
201
202			self.delete_config_file = True
203			return config_file.name
204
205			def setUp(self):
206			super(IntegTestBase, self).setUp()
207			prefix = 'TabPy_IntegTest_'
208			self.tmp_dir = tempfile.mkdtemp(prefix=prefix)
209
210			# create temporary state.ini
211			orig_state_file_name = os.path.abspath(
212			self._get_state_file_path() + '/state.ini')
213			self.state_file_name = os.path.abspath(self.tmp_dir + '/state.ini')
214			if orig_state_file_name != self.state_file_name:
215			shutil.copyfile(orig_state_file_name, self.state_file_name)
216
217			# create config file
218			orig_config_file_name = os.path.abspath(self._get_config_file_name())
219			self.config_file_name = os.path.abspath(
220			self.tmp_dir + '/' +
221			os.path.basename(orig_config_file_name))
222			if orig_config_file_name != self.config_file_name:
223			shutil.copyfile(orig_config_file_name, self.config_file_name)
224
225			# Platform specific - for integration tests we want to engage
226			# startup script
227			with open(self.tmp_dir + '/output.txt', 'w') as outfile:
228			cmd = ['tabpy',
229			'--config=' + self.config_file_name]
230			preexec_fn = None
231			if platform.system() == 'Windows':
232			self.py = 'python'
233			else:
234			self.py = 'python3'
235			preexec_fn = os.setsid
236
237			coverage.process_startup()
238			self.process = subprocess.Popen(
239			cmd,
240			preexec_fn=preexec_fn,
241			stdout=outfile,
242			stderr=outfile)
243
244			# give the app some time to start up...
245			time.sleep(5)
246
247			def tearDown(self):
248			# stop TabPy
249			if self.process is not None:
250			if platform.system() == 'Windows':
251			subprocess.call(['taskkill', '/F', '/T', '/PID',
252			str(self.process.pid)])
253			else:
254			os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
255			self.process.kill()
256
257			# after shutting down TabPy and before we start it again
258			# for next test give it some time to terminate.
259			time.sleep(5)
260
261			# remove temporary files
262			if self.delete_temp_folder:
263			os.remove(self.state_file_name)
264			os.remove(self.config_file_name)
265			shutil.rmtree(self.tmp_dir)
266
267			super(IntegTestBase, self).tearDown()
268
269			def _get_connection(self) -> http.client.HTTPConnection:
270			protocol = self._get_transfer_protocol()
271			url = 'localhost:' + self._get_port()
272
273			if protocol is not None and protocol.lower() == 'https':
274			connection = http.client.HTTPSConnection(url)
275			else:
276			connection = http.client.HTTPConnection(url)
277
278			return connection
279
280			def _get_username(self) -> str:
281			return 'user1'
282
283			def _get_password(self) -> str:
284			return 'P@ssw0rd'
285
286			def deploy_models(self, username: str, password: str):
287			repo_dir = os.path.abspath(os.path.dirname(tabpy.__file__))
288			path = os.path.join(repo_dir, 'models', 'deploy_models.py')
289			with open(self.tmp_dir + '/deploy_models_output.txt', 'w') as outfile:
290			outfile.write(
291			f'--<< Running {self.py} {path} '
292			f'{self._get_config_file_name()} >>--\n')
293			input_string = f'{username}\n{password}\n'
294			outfile.write(f'--<< Input = {input_string} >>--')
295			coverage.process_startup()
296			p = subprocess.run(
297			[self.py, path, self._get_config_file_name()],
298			input=input_string.encode('utf-8'),
299			stdout=outfile,
300			stderr=outfile)
301
302			def _get_process(self):
303			return self.process
304

tableau / TabPy

Push — master ( 305552...00a4d3 )

integ_test_base.IntegTestBase._get_process() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like