refactor: Remove all usage of custom OSName class

pyproject.toml

@@ -176,8 +176,8 @@ source = [
 ]
 
 [tool.coverage.coverage_conditional_plugin.rules]
-is-windows = "sys_platform == 'win32'"
-is-posix = "sys_platform != 'win32'"
+skip-coverage-windows = "sys_platform == 'win32'"
+skip-coverage-posix = "sys_platform != 'win32'"
 
 [tool.coverage.report]
 show_missing = true

src/openjd/adaptor_runtime/_background/backend_runner.py

@@ -10,19 +10,21 @@
 from threading import Thread, Event
 import traceback
 from types import FrameType
-from typing import Callable, List, Optional, Union
+from typing import Callable, List, Optional
 
 from .server_response import ServerResponseGenerator
-from .._osname import OSName
 from ..adaptors import AdaptorRunner
-from .._http import SocketPaths
 from .._utils import secure_open
 
-if OSName.is_posix():
-    from .http_server import BackgroundHTTPServer
-if OSName.is_windows():
+if os.name == "nt":  # pragma: skip-coverage-posix
     from ...adaptor_runtime_client.named_pipe.named_pipe_helper import NamedPipeHelper
-    from .backend_named_pipe_server import WinBackgroundNamedPipeServer
+    from .backend_named_pipe_server import (
+        WinBackgroundNamedPipeServer,
+        WinBackgroundNamedPipeServer as BackgroundServerType,
+    )
+else:  # pragma: skip-coverage-windows
+    from .._http import SocketPaths
+    from .http_server import BackgroundHTTPServer, BackgroundHTTPServer as BackgroundServerType  # type: ignore
 from .log_buffers import LogBuffer
 from .model import ConnectionSettings
 from .model import DataclassJSONEncoder
@@ -46,12 +48,12 @@ def __init__(
         self._connection_file_path = connection_file_path
 
         self._log_buffer = log_buffer
-        self._server: Optional[Union[BackgroundHTTPServer, WinBackgroundNamedPipeServer]] = None
+        self._server: Optional[BackgroundServerType] = None
         signal.signal(signal.SIGINT, self._sigint_handler)
-        if OSName.is_posix():  # pragma: is-windows
-            signal.signal(signal.SIGTERM, self._sigint_handler)
-        else:  # pragma: is-posix
+        if os.name == "nt":  # pragma: skip-coverage-posix
             signal.signal(signal.SIGBREAK, self._sigint_handler)  # type: ignore[attr-defined]
+        else:  # pragma: skip-coverage-windows
+            signal.signal(signal.SIGTERM, self._sigint_handler)
 
     def _sigint_handler(self, signum: int, frame: Optional[FrameType]) -> None:
         """
@@ -81,24 +83,24 @@ def run(self, *, on_connection_file_written: List[Callable[[], None]] | None = N
         _logger.info("Running in background daemon mode.")
         shutdown_event: Event = Event()
 
-        if OSName.is_posix():  # pragma: is-windows
-            server_path = SocketPaths.for_os().get_process_socket_path(
+        if os.name == "nt":  # pragma: skip-coverage-posix
+            server_path = NamedPipeHelper.generate_pipe_name("AdaptorNamedPipe")
+        else:  # pragma: skip-coverage-windows
+            server_path = SocketPaths().get_process_socket_path(
                 ".openjd_adaptor_runtime",
                 create_dir=True,
             )
-        else:  # pragma: is-posix
-            server_path = NamedPipeHelper.generate_pipe_name("AdaptorNamedPipe")
 
         try:
-            if OSName.is_windows():  # pragma: is-posix
+            if os.name == "nt":  # pragma: skip-coverage-posix
                 self._server = WinBackgroundNamedPipeServer(
                     server_path,
                     self._adaptor_runner,
                     shutdown_event=shutdown_event,
                     log_buffer=self._log_buffer,
                 )
-            else:  # pragma: is-windows
-                self._server = BackgroundHTTPServer(
+            else:  # pragma: skip-coverage-windows
+                self._server = BackgroundHTTPServer(  # type: ignore
                     server_path,
                     self._adaptor_runner,
                     shutdown_event=shutdown_event,
@@ -153,8 +155,8 @@ def run(self, *, on_connection_file_written: List[Callable[[], None]] | None = N
             # NamedPipe servers are managed by Named Pipe File System it is not a regular file.
             # Once all handles are closed, the system automatically cleans up the named pipe.
             files_for_deletion = [self._connection_file_path]
-            if OSName.is_posix():  # pragma: is-windows
-                files_for_deletion.append(server_path)
+            if os.name != "nt":  # pragma: skip-coverage-windows
+                files_for_deletion.append(server_path)  # type: ignore
             for path in files_for_deletion:
                 try:
                     os.remove(path)

src/openjd/adaptor_runtime/_background/frontend_runner.py

@@ -20,7 +20,6 @@
 from types import ModuleType
 from typing import Optional, Callable, Dict
 
-from .._osname import OSName
 from ..process._logging import _ADAPTOR_OUTPUT_LEVEL
 from .._utils._constants import _OPENJD_ENV_STDOUT_PREFIX, _OPENJD_ADAPTOR_SOCKET_ENV
 from .loaders import ConnectionSettingsFileLoader
@@ -36,7 +35,7 @@
 
 _FRONTEND_RUNNER_REQUEST_TIMEOUT: float = 5.0
 
-if OSName.is_windows():
+if os.name == "nt":  # pragma: skip-coverage-posix
     from ...adaptor_runtime_client.named_pipe.named_pipe_helper import NamedPipeHelper
     import pywintypes
 
@@ -78,10 +77,10 @@ def __init__(
 
         self._canceled = Event()
         signal.signal(signal.SIGINT, self._sigint_handler)
-        if OSName.is_posix():  # pragma: is-windows
-            signal.signal(signal.SIGTERM, self._sigint_handler)
-        else:  # pragma: is-posix
+        if os.name == "nt":  # pragma: skip-coverage-posix
             signal.signal(signal.SIGBREAK, self._sigint_handler)  # type: ignore[attr-defined]
+        else:  # pragma: skip-coverage-windows
+            signal.signal(signal.SIGTERM, self._sigint_handler)
 
     def init(
         self,
@@ -270,7 +269,7 @@ def _heartbeat(self, ack_id: str | None = None) -> HeartbeatResponse:
         """
         params: dict[str, str] | None = {"ack_id": ack_id} if ack_id else None
         response = self._send_request("GET", "/heartbeat", params=params)
-        body = json.load(response.fp) if OSName.is_posix() else json.loads(response["body"])  # type: ignore
+        body = json.loads(response["body"]) if os.name == "nt" else json.load(response.fp)  # type: ignore
         return DataclassMapper(HeartbeatResponse).map(body)
 
     def _heartbeat_until_state_complete(self, state: AdaptorState) -> None:
@@ -330,7 +329,7 @@ def _send_request(
                 "Connection settings are required to send requests, but none were provided"
             )
 
-        if OSName.is_windows():  # pragma: is-posix
+        if os.name == "nt":  # pragma: skip-coverage-posix
             if params:
                 # This is used for aligning to the Linux's behavior in order to reuse the code in handler.
                 # In linux, query string params will always be put in a list.
@@ -353,7 +352,7 @@ def _send_request(
                 _logger.error(f"Failed to send {path} request: {e}")
                 raise
             return response
-        else:  # pragma: is-windows
+        else:  # pragma: skip-coverage-windows
             return self._send_linux_request(
                 method,
                 path,
@@ -368,7 +367,7 @@ def _send_linux_request(
         *,
         params: dict | None = None,
         json_body: dict | None = None,
-    ) -> http_client.HTTPResponse:  # pragma: is-windows
+    ) -> http_client.HTTPResponse:  # pragma: skip-coverage-windows
         if not self.connection_settings:
             raise ConnectionSettingsNotProvidedError(
                 "Connection settings are required to send requests, but none were provided"

src/openjd/adaptor_runtime/_entrypoint.py

@@ -5,6 +5,7 @@
 import contextlib
 import logging
 import os
+import platform
 import signal
 import sys
 import tempfile
@@ -40,7 +41,6 @@
     RuntimeConfiguration,
     ConfigurationManager,
 )
-from ._osname import OSName
 from ._utils._constants import _OPENJD_ADAPTOR_SOCKET_ENV, _OPENJD_LOG_REGEX
 from ._utils._logging import (
     ConditionalFormatter,
@@ -84,7 +84,7 @@
 _DIR = os.path.dirname(os.path.realpath(__file__))
 # Keyword args to init the ConfigurationManager for the runtime.
 _ENV_CONFIG_PATH_PREFIX = "RUNTIME_CONFIG_PATH"
-_system_config_path_prefix = "/etc" if OSName.is_posix() else os.environ["PROGRAMDATA"]
+_system_config_path_prefix = os.environ["PROGRAMDATA"] if os.name == "nt" else "/etc"
 _system_config_path = os.path.abspath(
     os.path.join(
         _system_config_path_prefix,
@@ -226,7 +226,7 @@ def _init_config(self) -> None:
             raise
         except NotImplementedError as e:
             _logger.warning(
-                f"The current system ({OSName()}) is not supported for runtime "
+                f"The current system ({platform.platform()}) is not supported for runtime "
                 f"configuration. Only the default configuration will be loaded. Full error: {e}"
             )
             # The above call to build_config() would have already successfully retrieved the
@@ -354,10 +354,10 @@ def _handle_run(
         self._adaptor_runner = AdaptorRunner(adaptor=adaptor)
         # To be able to handle cancelation via signals
         signal.signal(signal.SIGINT, self._sigint_handler)
-        if OSName.is_posix():  # pragma: is-windows
-            signal.signal(signal.SIGTERM, self._sigint_handler)
-        else:  # pragma: is-posix
+        if os.name == "nt":  # pragma: skip-coverage-posix
             signal.signal(signal.SIGBREAK, self._sigint_handler)  # type: ignore[attr-defined]
+        else:  # pragma: skip-coverage-windows
+            signal.signal(signal.SIGTERM, self._sigint_handler)
         try:
             self._adaptor_runner._start()
             self._adaptor_runner._run(integration_data.run_data)

src/openjd/adaptor_runtime/_http/request_handler.py

@@ -12,8 +12,8 @@
 from dataclasses import dataclass
 from http import HTTPStatus, server
 from typing import Any, Callable, Type
+import sys
 
-from .._osname import OSName
 from .exceptions import UnsupportedPlatformException
 
 _logger = logging.getLogger(__name__)
@@ -125,8 +125,8 @@ def _authenticate(self) -> bool:
         peercred_opt_level: Any
         peercred_opt: Any
         cred_cls: Any
-        if OSName.is_macos():  # pragma: no cover
-            # SOL_LOCAL is not defined in Python's socket module, need to hardcode it
+        if sys.platform == "darwin":
+            # On MacOS, SOL_LOCAL is not defined in Python's socket module, need to hardcode it
             # source: https://github.com/apple-oss-distributions/xnu/blob/1031c584a5e37aff177559b9f69dbd3c8c3fd30a/bsd/sys/un.h#L85
             peercred_opt_level = 0  # type: ignore[attr-defined]
             peercred_opt = socket.LOCAL_PEERCRED  # type: ignore[attr-defined]

src/openjd/adaptor_runtime/_http/sockets.py

@@ -5,11 +5,11 @@
 import abc
 import os
 import stat
+import sys
 import tempfile
+from typing import Type
 
-from .._osname import OSName
 from .exceptions import (
-    UnsupportedPlatformException,
     NonvalidSocketPathException,
     NoSocketPathFoundException,
 )
@@ -18,29 +18,11 @@
 _PID_MAX_LENGTH = 7
 
 
-class SocketPaths(abc.ABC):
+class SocketPathsBase(abc.ABC):
     """
     Base class for determining the paths for sockets used in the Adaptor Runtime.
     """
 
-    @staticmethod
-    def for_os(osname: OSName = OSName()):  # pragma: no cover
-        """
-        Gets the SocketPaths class for a specific OS.
-
-        Args:
-            osname (OSName, optional): The OS to get socket paths for.
-                Defaults to the current OS.
-
-        Raises:
-            UnsupportedPlatformException: Raised when this class is requested for an unsupported
-                platform.
-        """
-        klass = _get_socket_paths_cls(osname)
-        if not klass:
-            raise UnsupportedPlatformException(osname)
-        return klass()
-
     def get_process_socket_path(
         self,
         namespace: str | None = None,
@@ -143,7 +125,7 @@ def verify_socket_path(self, path: str) -> None:  # pragma: no cover
         pass
 
 
-class WindowsSocketPaths(SocketPaths):
+class WindowsSocketPaths(SocketPathsBase):
     """
     Specialization for verifying socket paths on Windows systems.
     """
@@ -153,7 +135,7 @@ def verify_socket_path(self, path: str) -> None:
         pass
 
 
-class UnixSocketPaths(SocketPaths):
+class UnixSocketPaths(SocketPathsBase):
     """
     Specialization for verifying socket paths on Unix systems.
     """
@@ -215,14 +197,12 @@ def verify_socket_path(self, path: str) -> None:
             )
 
 
-_os_map: dict[str, type[SocketPaths]] = {
-    OSName.LINUX: LinuxSocketPaths,
-    OSName.MACOS: MacOSSocketPaths,
-    OSName.WINDOWS: WindowsSocketPaths,
-}
-
+SocketPaths: Type
 
-def _get_socket_paths_cls(
-    osname: OSName,
-) -> type[SocketPaths] | None:  # pragma: no cover
-    return _os_map.get(osname, None)
+# Make SocketPaths the correct operating-specific subclass of SocketPathsBase
+if sys.platform == "darwin":  # pragma: skip-coverage-windows
+    SocketPaths = MacOSSocketPaths
+elif os.name == "nt":  # pragma: skip-coverage-posix
+    SocketPaths = WindowsSocketPaths
+else:  # pragma: skip-coverage-windows
+    SocketPaths = LinuxSocketPaths

src/openjd/adaptor_runtime/_named_pipe/named_pipe_request_handler.py

@@ -1,6 +1,8 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
 import json
+import os
+import platform
 from typing import TYPE_CHECKING, Dict, List
 
 if TYPE_CHECKING:  # pragma: no cover because pytest will think we should test for this.
@@ -18,8 +20,6 @@
 import traceback
 from abc import ABC, abstractmethod
 
-from openjd.adaptor_runtime._osname import OSName
-
 
 _logger = logging.getLogger(__name__)
 
@@ -42,10 +42,10 @@ def __init__(self, server: "NamedPipeServer", pipe_handle: HANDLE):
                 Utilized for message read/write operations.
         """
         self._handler_type_name = self.__class__.__name__
-        if not OSName.is_windows():
+        if os.name != "nt":  # pragma: skip-coverage-windows
             raise OSError(
                 f"{self._handler_type_name} can be only used on Windows Operating Systems. "
-                f"Current Operating System is {OSName._get_os_name()}"
+                f"Current Operating System is {platform.platform()}"
             )
         self.server = server
         self.pipe_handle = pipe_handle

src/openjd/adaptor_runtime/_named_pipe/named_pipe_server.py

@@ -1,8 +1,10 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
 from __future__ import annotations
-import logging
 
+import logging
+import os
+import platform
 import threading
 from threading import Event
 
@@ -20,7 +22,6 @@
 
 if TYPE_CHECKING:
     from .._named_pipe import ResourceRequestHandler
-from .._osname import OSName
 
 import win32pipe
 import win32file
@@ -83,10 +84,10 @@ def __init__(self, pipe_name: str, shutdown_event: Event):  # pragma: no cover
             shutdown_event (Event): An Event used for signaling server shutdown.
         """
         self._server_type_name = self.__class__.__name__
-        if not OSName.is_windows():
+        if os.name != "nt":  # pragma: skip-coverage-windows
             raise OSError(
                 f"{self._server_type_name} can be only used on Windows Operating Systems. "
-                f"Current Operating System is {OSName._get_os_name()}"
+                f"Current Operating System is {platform.platform()}"
             )
         self._named_pipe_instances: List[HANDLE] = []
         self._pipe_name = pipe_name