tests.integration.integ_test_base.IntegTestBase.tearDown() - Code Metrics - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

IntegTestBase.tearDown() A
last analyzed 2024-11-25 22:00 UTC

↳ Parent: tests.integration.integ_test_base

Complexity

Conditions

Size

Total Lines	20
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	20
rs	9.8
c	0
b	0
f	0
cc	4
nop	1

import coverage
import http.client
import os
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 = {self.tmp_dir}/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
        self.log_file_path = os.path.join(self.tmp_dir, "output.txt")
        with open(self.log_file_path, "w") as outfile:
            cmd = ["tabpy", "--config=" + self.config_file_name, "--disable-auth-warning"]
            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_url(self) -> str:
        protocol = self._get_transfer_protocol()
        url = ""
        if protocol is not None and protocol.lower() == "https":
            url = "https://"
        else:
            url = "http://"
        url += "localhost:" + self._get_port()
        return url

    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()
            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		import platform
5		import shutil
6		import signal
7		import subprocess
8		import tabpy
9		import tempfile
10		import time
11		import unittest
12
13
14		class IntegTestBase(unittest.TestCase):
15		"""
16		Base class for integration tests.
17		"""
18
19		def __init__(self, methodName="runTest"):
20		super(IntegTestBase, self).__init__(methodName)
21		self.process = None
22		self.delete_temp_folder = True
23
24		def set_delete_temp_folder(self, delete_temp_folder: bool):
25		"""
26		Specify if temporary folder for state, config and log
27		files should be deleted when test is done.
28		By default the folder is deleted.
29
30		Parameters
31		----------
32		delete_test_folder: bool
33		If True temp folder will be deleted.
34		"""
35		self.delete_temp_folder = delete_temp_folder
36
37		def _get_state_file_path(self) -> str:
38		"""
39		Generates state.ini and returns absolute path to it.
40		Overwrite this function for tests to run against not default state
41		file.
42
43		Returns
44		-------
45		str
46		Absolute path to state file folder.
47		"""
48		state_file = open(os.path.join(self.tmp_dir, "state.ini"), "w+")
49		state_file.write(
50		"[Service Info]\n"
51		"Name = TabPy Serve\n"
52		"Description = \n"
53		"Creation Time = 0\n"
54		"Access-Control-Allow-Origin = \n"
55		"Access-Control-Allow-Headers = \n"
56		"Access-Control-Allow-Methods = \n"
57		"\n"
58		"[Query Objects Service Versions]\n"
59		"\n"
60		"[Query Objects Docstrings]\n"
61		"\n"
62		"[Meta]\n"
63		"Revision Number = 1\n"
64		)
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	View Code Duplication	def _get_config_file_name(self) -> str:
		0 ignored issues – show Duplication introduced 2022-02-22 19:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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 = {self.tmp_dir}/query_objects\n"
173		f"TABPY_PORT = {self._get_port()}\n"
174		f"TABPY_STATE_PATH = {self.tmp_dir}\n"
175		)
176
177		pwd_file = self._get_pwd_file()
178		if pwd_file is not None:
179		pwd_file = os.path.abspath(pwd_file)
180		config_file.write(f"TABPY_PWD_FILE = {pwd_file}\n")
181
182		transfer_protocol = self._get_transfer_protocol()
183		if transfer_protocol is not None:
184		config_file.write(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		)
214		self.state_file_name = os.path.abspath(self.tmp_dir + "/state.ini")
215		if orig_state_file_name != self.state_file_name:
216		shutil.copyfile(orig_state_file_name, self.state_file_name)
217
218		# create config file
219		orig_config_file_name = os.path.abspath(self._get_config_file_name())
220		self.config_file_name = os.path.abspath(
221		self.tmp_dir + "/" + os.path.basename(orig_config_file_name)
222		)
223		if orig_config_file_name != self.config_file_name:
224		shutil.copyfile(orig_config_file_name, self.config_file_name)
225
226		# Platform specific - for integration tests we want to engage
227		# startup script
228		self.log_file_path = os.path.join(self.tmp_dir, "output.txt")
229		with open(self.log_file_path, "w") as outfile:
230		cmd = ["tabpy", "--config=" + self.config_file_name, "--disable-auth-warning"]
231		preexec_fn = None
232		if platform.system() == "Windows":
233		self.py = "python"
234		else:
235		self.py = "python3"
236		preexec_fn = os.setsid
237
238		coverage.process_startup()
239		self.process = subprocess.Popen(
240		cmd, preexec_fn=preexec_fn, stdout=outfile, stderr=outfile
241		)
242
243		# give the app some time to start up...
244		time.sleep(5)
245
246		def tearDown(self):
247		# stop TabPy
248		if self.process is not None:
249		if platform.system() == "Windows":
250		subprocess.call(["taskkill", "/F", "/T", "/PID", str(self.process.pid)])
251		else:
252		os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
253		self.process.kill()
254
255		# after shutting down TabPy and before we start it again
256		# for next test give it some time to terminate.
257		time.sleep(5)
258
259		# remove temporary files
260		if self.delete_temp_folder:
261		os.remove(self.state_file_name)
262		os.remove(self.config_file_name)
263		shutil.rmtree(self.tmp_dir)
264
265		super(IntegTestBase, self).tearDown()
266
267		def _get_url(self) -> str:
268		protocol = self._get_transfer_protocol()
269		url = ""
270		if protocol is not None and protocol.lower() == "https":
271		url = "https://"
272		else:
273		url = "http://"
274		url += "localhost:" + self._get_port()
275		return url
276
277		def _get_connection(self) -> http.client.HTTPConnection:
278		protocol = self._get_transfer_protocol()
279		url = "localhost:" + self._get_port()
280
281		if protocol is not None and protocol.lower() == "https":
282		connection = http.client.HTTPSConnection(url)
283		else:
284		connection = http.client.HTTPConnection(url)
285
286		return connection
287
288		def _get_username(self) -> str:
289		return "user1"
290
291		def _get_password(self) -> str:
292		return "P@ssw0rd"
293
294		def deploy_models(self, username: str, password: str):
295		repo_dir = os.path.abspath(os.path.dirname(tabpy.__file__))
296		path = os.path.join(repo_dir, "models", "deploy_models.py")
297		with open(self.tmp_dir + "/deploy_models_output.txt", "w") as outfile:
298		outfile.write(
299		f"--<< Running {self.py} {path} "
300		f"{self._get_config_file_name()} >>--\n"
301		)
302		input_string = f"{username}\n{password}\n"
303		outfile.write(f"--<< Input = {input_string} >>--")
304		coverage.process_startup()
305		subprocess.run(
306		[self.py, path, self._get_config_file_name()],
307		input=input_string.encode("utf-8"),
308		stdout=outfile,
309		stderr=outfile,
310		)
311
312		def _get_process(self):
313		return self.process
314

tableau / TabPy

IntegTestBase.tearDown() A last analyzed 2024-11-25 22:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

IntegTestBase.tearDown() A
last analyzed 2024-11-25 22:00 UTC