integ_test_base.IntegTestBase._get_evaluate_timeout() - Code Metrics - Inspection of "Migrate to Tornado 6 and other improvements (#329)" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d71f34...c856ab )

by Oleksandr

created 2019-07-31 20:42 UTC

IntegTestBase._get_evaluate_timeout() A

↳ Parent: integ_test_base

Complexity

Conditions

Size

Total Lines	14
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import http.client
import os
import platform
import shutil
import signal
import subprocess
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:
            if platform.system() == 'Windows':
                self.py = 'python'
                self.process = subprocess.Popen(
                    ['startup.cmd', self.config_file_name],
                    stdout=outfile,
                    stderr=outfile)
            else:
                self.py = 'python3'
                self.process = subprocess.Popen(
                    ['./startup.sh',
                     '--config=' + self.config_file_name],
                    preexec_fn=os.setsid,
                    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


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

tableau / TabPy

Push — master ( d71f34...c856ab )

IntegTestBase._get_evaluate_timeout() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like