revert writer changes

Author: gnufede

ddtrace/internal/ci_visibility/writer.py

@@ -1,8 +1,9 @@
 from http.client import RemoteDisconnected
 import os
 import socket
+from typing import TYPE_CHECKING  # noqa:F401
 from typing import Dict
-from typing import Optional
+from typing import Optional  # noqa:F401
 
 import ddtrace
 from ddtrace import config
@@ -11,10 +12,8 @@
 from ddtrace.internal.ci_visibility.constants import MODULE_TYPE
 from ddtrace.internal.ci_visibility.constants import SESSION_TYPE
 from ddtrace.internal.ci_visibility.constants import SUITE_TYPE
-from ddtrace.internal.logger import get_logger
-from ddtrace.internal.utils.http import Response
 from ddtrace.internal.utils.time import StopWatch
-from ddtrace.vendor.dogstatsd import DogStatsd
+from ddtrace.vendor.dogstatsd import DogStatsd  # noqa:F401
 
 from .. import agent
 from .. import service
@@ -39,11 +38,14 @@
 from .telemetry.payload import record_endpoint_payload_request_time
 
 
-log = get_logger(__name__)
+if TYPE_CHECKING:  # pragma: no cover
+    from typing import List  # noqa:F401
+
+    from ddtrace.internal.utils.http import Response  # noqa:F401
 
 
 class CIVisibilityEventClient(WriterClientBase):
-    def __init__(self) -> None:
+    def __init__(self):
         encoder = CIVisibilityEncoderV01(0, 0)
         encoder.set_metadata(
             "*",
@@ -55,7 +57,7 @@ def __init__(self) -> None:
                 "_dd.test.is_user_provided_service": "true" if config._is_user_provided_service else "false",
             },
         )
-        super().__init__(encoder)
+        super(CIVisibilityEventClient, self).__init__(encoder)
 
     def set_metadata(self, event_type: str, metadata: Dict[str, str]) -> None:
         if isinstance(self.encoder, CIVisibilityEncoderV01):
@@ -67,16 +69,14 @@ def set_test_session_name(self, test_session_name: str) -> None:
 
 
 class CIVisibilityCoverageClient(WriterClientBase):
-    def __init__(
-        self, intake_url: str, headers: Optional[Dict[str, str]] = None, itr_suite_skipping_mode: bool = False
-    ) -> None:
+    def __init__(self, intake_url, headers=None, itr_suite_skipping_mode=False):
         encoder = CIVisibilityCoverageEncoderV02(0, 0)
         if itr_suite_skipping_mode:
             encoder._set_itr_suite_skipping_mode(itr_suite_skipping_mode)
         self._intake_url = intake_url
         if headers:
             self._headers = headers
-        super().__init__(encoder)
+        super(CIVisibilityCoverageClient, self).__init__(encoder)
 
 
 class CIVisibilityProxiedCoverageClient(CIVisibilityCoverageClient):
@@ -102,18 +102,18 @@ class CIVisibilityWriter(HTTPWriter):
 
     def __init__(
         self,
-        intake_url: str = "",
-        processing_interval: Optional[float] = None,
-        timeout: Optional[float] = None,
-        dogstatsd: Optional[DogStatsd] = None,
-        sync_mode: bool = False,
-        report_metrics: bool = False,  # unused, but required for inheritance
-        api_version: Optional[str] = None,  # unused, but required for inheritance
-        reuse_connections: Optional[bool] = None,
-        headers: Optional[Dict[str, str]] = None,
-        use_evp: bool = False,
-        coverage_enabled: bool = False,
-        itr_suite_skipping_mode: bool = False,
+        intake_url="",  # type: str
+        processing_interval=None,  # type: Optional[float]
+        timeout=None,  # type: Optional[float]
+        dogstatsd=None,  # type: Optional[DogStatsd]
+        sync_mode=False,  # type: bool
+        report_metrics=False,  # type: bool
+        api_version=None,  # type: Optional[str]
+        reuse_connections=None,  # type: Optional[bool]
+        headers=None,  # type: Optional[Dict[str, str]]
+        use_evp=False,  # type: bool
+        coverage_enabled=False,  # type: bool
+        itr_suite_skipping_mode=False,  # type: bool
     ):
         if processing_interval is None:
             processing_interval = config._trace_writer_interval_seconds
@@ -129,15 +129,9 @@ def __init__(
         if not intake_url:
             intake_url = "%s.%s" % (AGENTLESS_BASE_URL, os.getenv("DD_SITE", AGENTLESS_DEFAULT_SITE))
 
-        ## Validate API key if headers are provided
-        # if headers and "dd-api-key" in headers and not headers["dd-api-key"]:
-        #     log.warning("Empty API key provided to CIVisibilityWriter. This may cause authentication issues.")
-        #     # Remove empty API key to prevent sending empty credentials
-        #     headers = {k: v for k, v in headers.items() if k != "dd-api-key"}
-
-        clients: list[WriterClientBase] = (
+        clients = (
             [CIVisibilityProxiedEventClient()] if use_evp else [CIVisibilityAgentlessEventClient()]
-        )
+        )  # type: List[WriterClientBase]
         if coverage_enabled:
             if not intake_cov_url:
                 intake_cov_url = "%s.%s" % (AGENTLESS_COVERAGE_BASE_URL, os.getenv("DD_SITE", AGENTLESS_DEFAULT_SITE))
@@ -153,23 +147,23 @@ def __init__(
                 )
             )
 
-        super().__init__(
+        super(CIVisibilityWriter, self).__init__(
             intake_url=intake_url,
             clients=clients,
             processing_interval=processing_interval,
             timeout=timeout,
             dogstatsd=dogstatsd,
             sync_mode=sync_mode,
-            report_metrics=report_metrics,
             reuse_connections=reuse_connections,
             headers=headers,
         )
 
-    def stop(self, timeout: Optional[float] = None) -> None:
+    def stop(self, timeout=None):
         if self.status != service.ServiceStatus.STOPPED:
-            super().stop(timeout=timeout)
+            super(CIVisibilityWriter, self).stop(timeout=timeout)
 
-    def recreate(self) -> "CIVisibilityWriter":
+    def recreate(self):
+        # type: () -> HTTPWriter
         return self.__class__(
             intake_url=self.intake_url,
             processing_interval=self._interval,
@@ -178,8 +172,9 @@ def recreate(self) -> "CIVisibilityWriter":
             sync_mode=self._sync_mode,
         )
 
-    def _put(self, data: bytes, headers: Dict[str, str], client: WriterClientBase, no_trace: bool) -> Response:
-        request_error: Optional[REQUEST_ERROR_TYPE] = None
+    def _put(self, data, headers, client, no_trace):
+        # type: (bytes, Dict[str, str], WriterClientBase, bool) -> Response
+        request_error = None  # type: Optional[REQUEST_ERROR_TYPE]
 
         with StopWatch() as sw:
             try: