tests: simplify by removing unneeded Python code
No binary (build failed)
—
(log yok)
| @@ -795,7 +795,7 @@ def run_score(args, protocol): | |
| socks_args = None | |
| if args.socks4 and args.socks5: | |
| raise ScoreCardError('unable to run --socks4 and --socks5 together') | |
| - | elif args.socks4 or args.socks5: |
| + | if args.socks4 or args.socks5: |
| sockd = Dante(env=env) | |
| if sockd: | |
| assert sockd.initial_start() | |
| @@ -701,10 +701,6 @@ class TestDownload: | |
| if url_junk <= 1024: | |
| r.check_exit_code(0) | |
| r.check_response(http_status=200) | |
| - | elif url_junk <= 16 * 1024: |
| - | r.check_exit_code(0) |
| - | # server replies with 414, Request URL too long |
| - | r.check_response(http_status=414) |
| elif url_junk <= 32 * 1024: | |
| r.check_exit_code(0) | |
| # server replies with 414, Request URL too long | |
| @@ -71,12 +71,12 @@ class UDSFaker: | |
| c, client_address = self._socket.accept() | |
| try: | |
| c.recv(16) | |
| - | c.sendall("""HTTP/1.1 200 Ok |
| + | c.sendall(b"""HTTP/1.1 200 Ok |
| Server: UdsFaker | |
| Content-Type: application/json | |
| Content-Length: 19 | |
| - | { "host": "faked" }""".encode()) |
| + | { "host": "faked" }""") |
| finally: | |
| c.close() | |
| @@ -80,9 +80,8 @@ class TestSSLUse: | |
| count = 3 | |
| exp_resumed = 'Resumed' | |
| xargs = ['--sessionid', '--tls-max', tls_max, f'--tlsv{tls_max}'] | |
| - | if env.curl_uses_lib('libressl'): |
| - | if tls_max == '1.3': |
| - | exp_resumed = 'Initial' # 1.2 works in LibreSSL, but 1.3 does not, TODO |
| + | if env.curl_uses_lib('libressl') and tls_max == '1.3': |
| + | exp_resumed = 'Initial' # 1.2 works in LibreSSL, but 1.3 does not, TODO |
| if env.curl_uses_lib('rustls-ffi'): | |
| exp_resumed = 'Initial' # Rustls does not support sessions, TODO | |
| if env.curl_uses_lib('mbedtls') and tls_max == '1.3' and \ | |
| @@ -265,9 +264,9 @@ class TestSSLUse: | |
| elif env.curl_uses_lib('schannel'): # not in CI, so untested | |
| if ciphers12 is not None: | |
| pytest.skip('Schannel does not support setting TLSv1.2 ciphers by name') | |
| - | elif env.curl_uses_lib('mbedtls') and not env.curl_lib_version_at_least('mbedtls', '3.6.0'): |
| - | if tls_proto == 'TLSv1.3': |
| - | pytest.skip('mbedTLS < 3.6.0 does not support TLSv1.3') |
| + | elif (env.curl_uses_lib('mbedtls') and not env.curl_lib_version_at_least('mbedtls', '3.6.0') |
| + | and tls_proto == 'TLSv1.3'): |
| + | pytest.skip('mbedTLS < 3.6.0 does not support TLSv1.3') |
| # test | |
| extra_args = ['--tls13-ciphers', ':'.join(ciphers13)] if ciphers13 else [] | |
| extra_args += ['--ciphers', ':'.join(ciphers12)] if ciphers12 else [] | |
| @@ -317,13 +316,12 @@ class TestSSLUse: | |
| ]) | |
| httpd.reload_if_config_changed() | |
| # curl's TLS backend supported version | |
| - | if env.curl_uses_lib('gnutls') or \ |
| - | env.curl_uses_lib('quiche') or \ |
| - | env.curl_uses_lib('aws-lc') or \ |
| - | env.curl_uses_lib('boringssl'): |
| - | curl_supported = [0x301, 0x302, 0x303, 0x304] |
| - | elif env.curl_uses_lib('openssl') and \ |
| - | env.curl_lib_version_before('openssl', '3.0.0'): |
| + | if (env.curl_uses_lib('gnutls') or |
| + | env.curl_uses_lib('quiche') or |
| + | env.curl_uses_lib('aws-lc') or |
| + | env.curl_uses_lib('boringssl') or |
| + | (env.curl_uses_lib('openssl') and |
| + | env.curl_lib_version_before('openssl', '3.0.0'))): |
| curl_supported = [0x301, 0x302, 0x303, 0x304] | |
| else: # most SSL backends dropped support for TLSv1.0, TLSv1.1 | |
| curl_supported = [0x303, 0x304] | |
| @@ -389,20 +389,18 @@ class TestCA: | |
| :returns: the certificate and private key PEM file paths | |
| """ | |
| if spec.domains and len(spec.domains): | |
| - | creds = TestCA._make_server_credentials(name=spec.name, domains=spec.domains, |
| - | issuer=issuer, valid_from=valid_from, |
| - | valid_to=valid_to, key_type=key_type) |
| - | elif spec.client: |
| - | creds = TestCA._make_client_credentials(name=spec.name, issuer=issuer, |
| - | email=spec.email, valid_from=valid_from, |
| - | valid_to=valid_to, key_type=key_type) |
| - | elif spec.name: |
| - | creds = TestCA._make_ca_credentials(name=spec.name, issuer=issuer, |
| - | valid_from=valid_from, valid_to=valid_to, |
| - | key_type=key_type) |
| - | else: |
| - | raise CertError(f"unrecognized certificate specification: {spec}") |
| - | return creds |
| + | return TestCA._make_server_credentials(name=spec.name, domains=spec.domains, |
| + | issuer=issuer, valid_from=valid_from, |
| + | valid_to=valid_to, key_type=key_type) |
| + | if spec.client: |
| + | return TestCA._make_client_credentials(name=spec.name, issuer=issuer, |
| + | email=spec.email, valid_from=valid_from, |
| + | valid_to=valid_to, key_type=key_type) |
| + | if spec.name: |
| + | return TestCA._make_ca_credentials(name=spec.name, issuer=issuer, |
| + | valid_from=valid_from, valid_to=valid_to, |
| + | key_type=key_type) |
| + | raise CertError(f"unrecognized certificate specification: {spec}") |
| @staticmethod | |
| def _make_x509_name(org_name: Optional[str] = None, common_name: Optional[str] = None, parent: x509.Name = None) -> x509.Name: | |
| @@ -95,7 +95,6 @@ class H2o: | |
| def _rmf(self, path): | |
| if os.path.isfile(path): | |
| os.remove(path) | |
| - | return |
| def _dump_file(self, path, lines): | |
| if os.path.isfile(path): | |
| @@ -106,7 +105,6 @@ class H2o: | |
| def _mkpath(self, path): | |
| if not os.path.exists(path): | |
| os.makedirs(path) | |
| - | return |
| def _log(self, level, msg): | |
| getattr(log, level)(f"[{self._name}] {msg}") | |
| @@ -293,15 +293,13 @@ class TestSmbServer(imp_smbserver.SMBSERVER): | |
| # If we have a rootFid, the path is relative to that fid | |
| path = conn_data["OpenedFiles"][root_fid]["FileName"] | |
| log.debug(f'RootFid present {path}!') | |
| - | else: |
| - | if "path" in conn_shares[tid]: |
| - | path = conn_shares[tid]["path"] |
| - | else: |
| - | raise SmbError(STATUS_ACCESS_DENIED, "Connection share had no path") |
| - | else: |
| - | raise SmbError(imp_smbserver.STATUS_SMB_BAD_TID, "TID was invalid") |
| - | |
| - | return path |
| + | return path |
| + | |
| + | if "path" in conn_shares[tid]: |
| + | return conn_shares[tid]["path"] |
| + | |
| + | raise SmbError(STATUS_ACCESS_DENIED, "Connection share had no path") |
| + | raise SmbError(imp_smbserver.STATUS_SMB_BAD_TID, "TID was invalid") |
| def get_server_path(self, requested_filename): | |
| log.debug("[SMB] Get server path '%s'", requested_filename) | |
| @@ -48,15 +48,15 @@ class ClosingFileHandler(logging.StreamHandler): | |
| if callable(setStream): | |
| return setStream(stream) | |
| if stream is self.stream: | |
| - | result = None |
| - | else: |
| - | result = self.stream |
| - | self.acquire() |
| - | try: |
| - | self.flush() |
| - | self.stream = stream |
| - | finally: |
| - | self.release() |
| + | return None |
| + | |
| + | result = self.stream |
| + | self.acquire() |
| + | try: |
| + | self.flush() |
| + | self.stream = stream |
| + | finally: |
| + | self.release() |
| return result | |