Code Duplication - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 45-47 lines in 3 locations

tests/integration/test_max_request_size.py 1 location




class TestMaxRequestSize(integ_test_base.IntegTestBase):
    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_GZIP_ENABLE = TRUE\n"
            f"TABPY_STATE_PATH = {self.tmp_dir}\n"
            "TABPY_MAX_REQUEST_SIZE_MB = 1\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 create_large_payload(self):
        size_mb = 2

tests/integration/test_gzip.py 1 location




class TestEvaluate(integ_test_base.IntegTestBase):
    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_GZIP_ENABLE = TRUE\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 test_single_value_returned(self):
        payload = """

tests/integration/integ_test_base.py 1 location


        """
        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()

		@@ 14-60 (lines=47) @@
11
12
13		class TestMaxRequestSize(integ_test_base.IntegTestBase):
14		def _get_config_file_name(self) -> str:
15		"""
16		Generates config file. Overwrite this function for tests to
17		run against not default state file.
18
19		Returns
20		-------
21		str
22		Absolute path to config file.
23		"""
24		config_file = open(os.path.join(self.tmp_dir, "test.conf"), "w+")
25		config_file.write(
26		"[TabPy]\n"
27		f"TABPY_QUERY_OBJECT_PATH = {self.tmp_dir}/query_objects\n"
28		f"TABPY_PORT = {self._get_port()}\n"
29		f"TABPY_GZIP_ENABLE = TRUE\n"
30		f"TABPY_STATE_PATH = {self.tmp_dir}\n"
31		"TABPY_MAX_REQUEST_SIZE_MB = 1\n"
32		)
33
34		pwd_file = self._get_pwd_file()
35		if pwd_file is not None:
36		pwd_file = os.path.abspath(pwd_file)
37		config_file.write(f"TABPY_PWD_FILE = {pwd_file}\n")
38
39		transfer_protocol = self._get_transfer_protocol()
40		if transfer_protocol is not None:
41		config_file.write(f"TABPY_TRANSFER_PROTOCOL = {transfer_protocol}\n")
42
43		cert_file_name = self._get_certificate_file_name()
44		if cert_file_name is not None:
45		cert_file_name = os.path.abspath(cert_file_name)
46		config_file.write(f"TABPY_CERTIFICATE_FILE = {cert_file_name}\n")
47
48		key_file_name = self._get_key_file_name()
49		if key_file_name is not None:
50		key_file_name = os.path.abspath(key_file_name)
51		config_file.write(f"TABPY_KEY_FILE = {key_file_name}\n")
52
53		evaluate_timeout = self._get_evaluate_timeout()
54		if evaluate_timeout is not None:
55		config_file.write(f"TABPY_EVALUATE_TIMEOUT = {evaluate_timeout}\n")
56
57		config_file.close()
58
59		self.delete_config_file = True
60		return config_file.name
61
62		def create_large_payload(self):
63		size_mb = 2

		@@ 13-58 (lines=46) @@
10
11
12		class TestEvaluate(integ_test_base.IntegTestBase):
13		def _get_config_file_name(self) -> str:
14		"""
15		Generates config file. Overwrite this function for tests to
16		run against not default state file.
17
18		Returns
19		-------
20		str
21		Absolute path to config file.
22		"""
23		config_file = open(os.path.join(self.tmp_dir, "test.conf"), "w+")
24		config_file.write(
25		"[TabPy]\n"
26		f"TABPY_QUERY_OBJECT_PATH = {self.tmp_dir}/query_objects\n"
27		f"TABPY_PORT = {self._get_port()}\n"
28		f"TABPY_GZIP_ENABLE = TRUE\n"
29		f"TABPY_STATE_PATH = {self.tmp_dir}\n"
30		)
31
32		pwd_file = self._get_pwd_file()
33		if pwd_file is not None:
34		pwd_file = os.path.abspath(pwd_file)
35		config_file.write(f"TABPY_PWD_FILE = {pwd_file}\n")
36
37		transfer_protocol = self._get_transfer_protocol()
38		if transfer_protocol is not None:
39		config_file.write(f"TABPY_TRANSFER_PROTOCOL = {transfer_protocol}\n")
40
41		cert_file_name = self._get_certificate_file_name()
42		if cert_file_name is not None:
43		cert_file_name = os.path.abspath(cert_file_name)
44		config_file.write(f"TABPY_CERTIFICATE_FILE = {cert_file_name}\n")
45
46		key_file_name = self._get_key_file_name()
47		if key_file_name is not None:
48		key_file_name = os.path.abspath(key_file_name)
49		config_file.write(f"TABPY_KEY_FILE = {key_file_name}\n")
50
51		evaluate_timeout = self._get_evaluate_timeout()
52		if evaluate_timeout is not None:
53		config_file.write(f"TABPY_EVALUATE_TIMEOUT = {evaluate_timeout}\n")
54
55		config_file.close()
56
57		self.delete_config_file = True
58		return config_file.name
59
60		def test_single_value_returned(self):
61		payload = """

		@@ 159-203 (lines=45) @@
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 = {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()

tableau / TabPy

Code Duplication Length = 45-47 lines in 3 locations

tests/integration/test_max_request_size.py 1 location

tests/integration/test_gzip.py 1 location

tests/integration/integ_test_base.py 1 location