chore(live-tests): deps updates + code read-through cleanup (#46731)

Co-authored-by: Octavia Squidington III <[email protected]>

airbyte-ci/connectors/live-tests/pyproject.toml

@@ -46,7 +46,7 @@ python-slugify = ">=8.0.4"
 
 
 [tool.poetry.group.dev.dependencies]
-ruff = "^0.3.0"
+ruff = "^0.6"
 mypy = "^1.8.0"
 types-cachetools = "^5.3.0.7"
 pandas-stubs = "^2.2.0.240218"
@@ -55,6 +55,8 @@ types-pyyaml = "^6.0.12.20240311"
 
 [tool.ruff.lint]
 select = ["I", "F"]
+target-version = "py310"
+line-length = 140
 
 [tool.ruff.lint.isort]
 known-first-party = ["connection-retriever"]

airbyte-ci/connectors/live-tests/src/live_tests/commons/connector_runner.py

@@ -1,16 +1,13 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-from __future__ import annotations
-
 import datetime
 import json
 import logging
 import os
 import subprocess
 import uuid
 from pathlib import Path
-from typing import Optional
 
 import anyio
 import asyncer
@@ -32,7 +29,7 @@ def __init__(
         dagger_client: dagger.Client,
         execution_inputs: ExecutionInputs,
         is_airbyte_ci: bool,
-        http_proxy: Optional[Proxy] = None,
+        http_proxy: Proxy | None = None,
     ):
         self.connector_under_test = execution_inputs.connector_under_test
         self.command = execution_inputs.command
@@ -49,6 +46,9 @@ def __init__(
         self.http_proxy = http_proxy
         self.logger = logging.getLogger(f"{self.connector_under_test.name}-{self.connector_under_test.version}")
         self.dagger_client = dagger_client.pipeline(f"{self.connector_under_test.name}-{self.connector_under_test.version}")
+
+        # When invoked via airbyte-ci, record_obfuscator.py is copied over to /tmp
+        # but when running locally, it's in Airbyte repo root.
         if is_airbyte_ci:
             self.host_obfuscator_path = "/tmp/record_obfuscator.py"
         else:
@@ -68,6 +68,7 @@ def stderr_file_path(self) -> Path:
         return (self.output_dir / "stderr.log").resolve()
 
     def _get_full_command(self, command: Command) -> list[str]:
+        """Returns a list with a full Airbyte command invocation and all it's arguments and options."""
         if command is Command.SPEC:
             return ["spec"]
         elif command is Command.CHECK:
@@ -95,10 +96,10 @@ def _get_full_command(self, command: Command) -> list[str]:
         else:
             raise NotImplementedError(f"The connector runner does not support the {command} command")
 
-    async def get_container_env_variable_value(self, name: str) -> Optional[str]:
+    async def get_container_env_variable_value(self, name: str) -> str | None:
         return await self._connector_under_test_container.env_variable(name)
 
-    async def get_container_label(self, label: str) -> Optional[str]:
+    async def get_container_label(self, label: str) -> str | None:
         return await self._connector_under_test_container.label(label)
 
     async def get_container_entrypoint(self) -> str:
@@ -118,12 +119,14 @@ async def _run(
         container = self._connector_under_test_container
         # Do not cache downstream dagger layers
         container = container.with_env_variable("CACHEBUSTER", str(uuid.uuid4()))
-        expanded_host_executable_path = os.path.expanduser(self.host_obfuscator_path)
 
+        # When running locally, it's likely that record_obfuscator is within the user's home directory, so we expand it.
+        expanded_host_executable_path = os.path.expanduser(self.host_obfuscator_path)
         container = container.with_file(
             self.IN_CONTAINER_OBFUSCATOR_PATH,
             self.dagger_client.host().file(expanded_host_executable_path),
         )
+
         for env_var_name, env_var_value in self.environment_variables.items():
             container = container.with_env_variable(env_var_name, env_var_value)
         if self.config:
@@ -144,7 +147,7 @@ async def _run(
             entrypoint = await container.entrypoint()
             assert entrypoint, "The connector container has no entrypoint"
             airbyte_command = entrypoint + self.full_command
-            # We are piping the output to a file to avoidQueryError: file size exceeds limit 134217728
+            # We are piping the output to a file to avoid QueryError: file size exceeds limit 134217728
             container = container.with_exec(
                 [
                     "sh",
@@ -174,6 +177,7 @@ async def _run(
             self.logger.error(f"❌ Failed to run {self.command.value} command")
         else:
             self.logger.info(f"⌛ Finished running {self.command.value} command")
+
         execution_result = await ExecutionResult.load(
             command=self.command,
             connector_under_test=self.connector_under_test,

airbyte-ci/connectors/live-tests/src/live_tests/commons/errors.py

@@ -1,5 +1,4 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
 
 
 class ExportError(Exception):

airbyte-ci/connectors/live-tests/src/live_tests/commons/evaluation_modes.py

@@ -1,6 +1,4 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 from enum import Enum
 
 

airbyte-ci/connectors/live-tests/src/live_tests/commons/mitm_addons.py

@@ -1,6 +1,4 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 from urllib.parse import parse_qs, urlencode, urlparse
 
 from mitmproxy import http

airbyte-ci/connectors/live-tests/src/live_tests/commons/models.py

@@ -1,6 +1,4 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 import json
 import logging
 import tempfile
@@ -149,12 +147,20 @@ class ActorType(Enum):
 
 
 class ConnectionSubset(Enum):
+    """Signals which connection pool to consider for this live test — just the Airbyte sandboxes, or all possible connctions on Cloud."""
+
     SANDBOXES = "sandboxes"
     ALL = "all"
 
 
 @dataclass
 class ConnectorUnderTest:
+    """Represents a connector being tested.
+    In validation tests, there would be one connector under test.
+    When running regression tests, there would be two connectors under test: the target and the control versions of the same connector.
+    """
+
+    # connector image, assuming it's in the format "airbyte/{actor_type}-{connector_name}:{version}"
     image_name: str
     container: dagger.Container
     target_or_control: TargetOrControl

airbyte-ci/connectors/live-tests/src/live_tests/commons/proxy.py

@@ -1,9 +1,6 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 import logging
 import uuid
-from typing import Optional
 
 import dagger
 
@@ -12,7 +9,8 @@
 
 class Proxy:
     """
-    This class is a wrapper around a mitmproxy container. It allows to declare a mitmproxy container, bind it as a service to a different container and retrieve the mitmproxy stream file.
+    This class is a wrapper around a mitmproxy container. It allows to declare a mitmproxy container,
+    bind it as a service to a different container and retrieve the mitmproxy stream file.
     """
 
     MITMPROXY_IMAGE = "mitmproxy/mitmproxy:10.2.4"
@@ -25,7 +23,7 @@ def __init__(
         dagger_client: dagger.Client,
         hostname: str,
         session_id: str,
-        stream_for_server_replay: Optional[dagger.File] = None,
+        stream_for_server_replay: dagger.File | None = None,
     ) -> None:
         self.dagger_client = dagger_client
         self.hostname = hostname
@@ -142,14 +140,14 @@ async def bind_container(self, container: dagger.Container) -> dagger.Container:
             logging.warn(f"Failed to bind container to proxy: {e}")
             return container
 
-    async def retrieve_http_dump(self) -> Optional[dagger.File]:
+    async def retrieve_http_dump(self) -> dagger.File | None:
         """We mount the cache volume, where the mitmproxy container saves the stream file, to a fresh container.
         We then copy the stream file to a new directory and return it as a dagger.File.
         The copy operation to /to_export is required as Dagger does not support direct access to files in cache volumes.
 
 
         Returns:
-            Optional[dagger.File]: The mitmproxy stream file if it exists, None otherwise.
+            dagger.File | None: The mitmproxy stream file if it exists, None otherwise.
         """
         container = (
             self.dagger_client.container()

airbyte-ci/connectors/live-tests/src/live_tests/commons/secret_access.py

@@ -1,6 +1,4 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 import logging
 
 from google.api_core.exceptions import PermissionDenied

airbyte-ci/connectors/live-tests/src/live_tests/commons/segment_tracking.py

@@ -1,6 +1,4 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 import logging
 import os
 from importlib.metadata import version

airbyte-ci/connectors/live-tests/src/live_tests/conftest.py

@@ -1,5 +1,4 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
 
 import logging
 import os
@@ -316,7 +315,7 @@ def prompt_for_read_with_or_without_state() -> bool:
     📖 Do you want to run the read command with or without state?
     1. Run the read command with state
     2. Run the read command without state
-                              
+
     We recommend reading with state to properly test incremental sync.
     But if the target version introduces a breaking change in the state, you might want to run without state.
     """

airbyte-ci/connectors/live-tests/src/live_tests/report.py

@@ -1,7 +1,5 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 
-from __future__ import annotations
-
 import datetime
 import json
 from collections import defaultdict

airbyte-ci/connectors/live-tests/src/live_tests/stash_keys.py

@@ -1,6 +1,4 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
-from __future__ import annotations
-
 from pathlib import Path
 
 import pytest

airbyte-ci/connectors/live-tests/src/live_tests/utils.py

@@ -34,6 +34,8 @@ def filter_records(messages: Iterable[AirbyteMessage]) -> Iterable[AirbyteMessag
 
 
 def write_string_to_test_artifact(request: SubRequest, content: str, filename: str, subdir: Optional[Path] = None) -> Path:
+    # StashKey (in this case TEST_ARTIFACT_DIRECTORY) defines the output class of this,
+    # so this is already a Path.
     test_artifact_directory = request.config.stash[stash_keys.TEST_ARTIFACT_DIRECTORY]
     if subdir:
         test_artifact_directory = test_artifact_directory / subdir
@@ -69,13 +71,13 @@ def get_and_write_diff(
             request,
             formatted_diff_json,
             f"{filepath}_text.txt",
-            subdir=request.node.name,
+            subdir=Path(request.node.name),
         )
         diff_path_pretty = write_string_to_test_artifact(
             request,
             str(diff.pretty()),
             f"{filepath}_pretty.txt",
-            subdir=request.node.name,
+            subdir=Path(request.node.name),
         )
 
         logger.info(f"Diff file are stored in {diff_path_tree}, {diff_path_text}, and {diff_path_pretty}.")
@@ -128,7 +130,7 @@ def tail_file(file_path: Path, n: int = MAX_LINES_IN_REPORT) -> list[str]:
 
 def is_successful_check(execution_result: ExecutionResult) -> bool:
     for message in execution_result.airbyte_messages:
-        if message.type is Type.CONNECTION_STATUS and message.connectionStatus.status is Status.SUCCEEDED:
+        if message.type is Type.CONNECTION_STATUS and message.connectionStatus and message.connectionStatus.status is Status.SUCCEEDED:
             return True
     return False