Code Duplication    Length = 45-46 lines in 2 locations

tests/integration/test_gzip.py 1 location

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

tests/integration/integ_test_base.py 1 location

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