Make the --insecure CLI option test less hacky (cvat-ai#8601)

Test it by starting an HTTPS server (that forwards all requests to CVAT)
and attempting to access that server via the CLI. This allows us to test
that the option actually works end-to-end, and avoids dependencies on
CLI internals.

The new test requires `main` to catch the exception that is raised in
the case of a certificate error, so implement that.

changelog.d/20241029_134200_roman_cli_refactor.md

@@ -0,0 +1,4 @@
+### Changed
+
+- CLI no longer prints the stack trace in case of HTTP errors
+  (<https://github.com/cvat-ai/cvat/pull/8601>)

cvat-cli/src/cvat_cli/__main__.py

@@ -9,6 +9,7 @@
 from types import SimpleNamespace
 from typing import List
 
+import urllib3.exceptions
 from cvat_sdk import exceptions
 from cvat_sdk.core.client import Client, Config
 
@@ -70,7 +71,7 @@ def main(args: List[str] = None):
         try:
             cli = CLI(client=client, credentials=parsed_args.auth)
             actions[parsed_args.action](cli, **vars(action_args))
-        except exceptions.ApiException as e:
+        except (exceptions.ApiException, urllib3.exceptions.HTTPError) as e:
             logger.critical(e)
             return 1
 

tests/python/cli/self-signed.crt

@@ -0,0 +1,9 @@
+-----BEGIN CERTIFICATE-----
+MIIBPDCB76ADAgECAhQksQwFGcyVwF0+gIOPMPBB+/NjNTAFBgMrZXAwFDESMBAG
+A1UEAwwJbG9jYWxob3N0MB4XDTI0MTAyODEyMTkyNFoXDTI0MTAyOTEyMTkyNFow
+FDESMBAGA1UEAwwJbG9jYWxob3N0MCowBQYDK2VwAyEAzGOv96vkrHr0GPcWL7vN
+8mgR4XMg9ItNpJ2nbMmjYCKjUzBRMB0GA1UdDgQWBBR6Hn0aG/ZGAJjY9HIUK7El
+84qAgzAfBgNVHSMEGDAWgBR6Hn0aG/ZGAJjY9HIUK7El84qAgzAPBgNVHRMBAf8E
+BTADAQH/MAUGAytlcANBAMj2zWdIa8oOiEtUWFMv+KYf1kyP1lUnlcC2xUpOj8d3
+kRYtlRX4E7F5zzzgKgNpbanRAg72qnqPiFAFCGVAhgY=
+-----END CERTIFICATE-----

tests/python/cli/self-signed.key

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIKe5zj/UrVJ/LySjKm9BBVHXziqFIwJ6w+HuTHnldCLo
+-----END PRIVATE KEY-----

tests/python/cli/test_cli.py

@@ -10,7 +10,6 @@
 
 import packaging.version as pv
 import pytest
-from cvat_cli.cli import CLI
 from cvat_sdk import Client, make_client
 from cvat_sdk.api_client import exceptions, models
 from cvat_sdk.core.proxies.tasks import ResourceType, Task
@@ -20,7 +19,7 @@
 from shared.utils.config import BASE_URL, USER_PASS
 from shared.utils.helpers import generate_image_file
 
-from .util import generate_images, run_cli
+from .util import generate_images, https_reverse_proxy, run_cli
 
 
 class TestCLI:
@@ -243,23 +242,26 @@ def mocked_version(_):
         assert "Server version '0' is not compatible with SDK version" in caplog.text
 
     @pytest.mark.parametrize("verify", [True, False])
-    def test_can_control_ssl_verification_with_arg(self, monkeypatch, verify: bool):
-        # TODO: Very hacky implementation, improve it, if possible
-        class MyException(Exception):
-            pass
-
-        normal_init = CLI.__init__
-
-        def my_init(self, *args, **kwargs):
-            normal_init(self, *args, **kwargs)
-            raise MyException(self.client.api_client.configuration.verify_ssl)
-
-        monkeypatch.setattr(CLI, "__init__", my_init)
-
-        with pytest.raises(MyException) as capture:
-            self.run_cli(*(["--insecure"] if not verify else []), "ls")
-
-        assert capture.value.args[0] == verify
+    def test_can_control_ssl_verification_with_arg(self, verify: bool):
+        with https_reverse_proxy() as proxy_url:
+            if verify:
+                insecure_args = []
+            else:
+                insecure_args = ["--insecure"]
+
+            run_cli(
+                self,
+                f"--auth={self.user}:{self.password}",
+                f"--server-host={proxy_url}",
+                *insecure_args,
+                "ls",
+                expected_code=1 if verify else 0,
+            )
+            stdout = self.stdout.getvalue()
+
+        if not verify:
+            for line in stdout.splitlines():
+                int(line)
 
     def test_can_control_organization_context(self):
         org = "cli-test-org"

tests/python/cli/util.py

@@ -3,10 +3,17 @@
 # SPDX-License-Identifier: MIT
 
 
+import contextlib
+import http.server
+import ssl
+import threading
 import unittest
 from pathlib import Path
-from typing import Any, List, Union
+from typing import Any, Dict, Iterator, List, Union
 
+import requests
+
+from shared.utils.config import BASE_URL
 from shared.utils.helpers import generate_image_file
 
 
@@ -29,3 +36,50 @@ def generate_images(dst_dir: Path, count: int) -> List[Path]:
         filename.write_bytes(generate_image_file(filename.name).getvalue())
         filenames.append(filename)
     return filenames
+
+
+@contextlib.contextmanager
+def https_reverse_proxy() -> Iterator[str]:
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
+    cert_dir = Path(__file__).parent
+    ssl_context.load_cert_chain(cert_dir / "self-signed.crt", cert_dir / "self-signed.key")
+
+    with http.server.HTTPServer(("localhost", 0), _ProxyHttpRequestHandler) as proxy_server:
+        proxy_server.socket = ssl_context.wrap_socket(
+            proxy_server.socket,
+            server_side=True,
+        )
+        server_thread = threading.Thread(target=proxy_server.serve_forever)
+        server_thread.start()
+        try:
+            yield f"https://localhost:{proxy_server.server_port}"
+        finally:
+            proxy_server.shutdown()
+            server_thread.join()
+
+
+class _ProxyHttpRequestHandler(http.server.BaseHTTPRequestHandler):
+    def do_GET(self):
+        response = requests.get(**self._shared_request_args())
+        self._translate_response(response)
+
+    def do_POST(self):
+        body_length = int(self.headers["Content-Length"])
+
+        response = requests.post(data=self.rfile.read(body_length), **self._shared_request_args())
+        self._translate_response(response)
+
+    def _shared_request_args(self) -> Dict[str, Any]:
+        headers = {k.lower(): v for k, v in self.headers.items()}
+        del headers["host"]
+
+        return {"url": BASE_URL + self.path, "headers": headers, "timeout": 60, "stream": True}
+
+    def _translate_response(self, response: requests.Response) -> None:
+        self.send_response(response.status_code)
+        for key, value in response.headers.items():
+            self.send_header(key, value)
+        self.end_headers()
+        # Need to use raw here to prevent requests from handling Content-Encoding.
+        self.wfile.write(response.raw.read())