Activate ruff rules on SSI/auto-injection

pyproject.toml

@@ -203,8 +203,6 @@ ignore = [
     "SIM108",    # if-else-block-instead-of-if-exp: ternary operator nay be harder to read on long statements
 ]
 "tests/auto_inject/*" = [
-    "ANN001",  # missing-type-function-argument: TODO
-    "RUF059",   # unused-unpacked-variable: TODO
     "SLF001",  # private-member-access: TODO
 ]
 "tests/fuzzer/*" = [
@@ -224,11 +222,9 @@ ignore = [
     "S605",      # start-process-with-a-shell: test the test needs to run shell commands
 ]
 "utils/_context/_scenarios/auto_injection.py" = [
-    "ANN001",   # missing-type-function-argument: TODO
     "PLC0415",  # import-outside-top-level TODO
 ]
 "utils/{_context/_scenarios/docker_ssi.py,docker_ssi/docker_ssi_matrix_utils.py}" = [
-    "ANN001",   # missing-type-function-argument: TODO
     "E501",     # line-too-long: TODO
     "T201",     # print: TODO
     "RET504",   # unnecessary-assign: TODO
@@ -242,11 +238,6 @@ ignore = [
     "ANN205"  # missing-return-type-static-method: obvious decorators
 ]
 "utils/build/*"  = ["ALL"]  # mostly python weblog code. it may be a good idea to enable rules here
-"utils/docker_ssi/*" = [
-    "ANN001",   # missing-type-function-argument: TODO
-    "ANN201",  # missing-return-type-undocumented-public-function: TODO
-    "ANN205",  # issing-return-type-static-method: TODO
-]
 
 "utils/{k8s_lib_injection/*,_context/_scenarios/k8s_lib_injection.py}" = [
     "ANN003",   # missing-type-kwargs: TODO
@@ -288,7 +279,6 @@ ignore = [
     "UP017",   # datetime-timezone-utc: TODO
 ]
 "utils/scripts/ssi_wizards/aws_onboarding_wizard_utils.py" = [
-    "ANN001",   # missing-type-function-argument: TODO
     "PLR2004",  # magic-value-comparison: TODO
 ]
 

requirements.txt

@@ -42,6 +42,7 @@ setuptools==75.8.0
 shellcheck-py==0.11.0.1
 types-aiofiles==24.1.0.20241221
 types-mock==5.2.0.20250924
+types-paramiko==4.0.0.20250822
 types-protobuf==5.29.1.20241207
 types-python-dateutil==2.9.0.20241206
 types-PyYAML==6.0.12.20241230

tests/auto_inject/test_auto_inject_chaos.py

@@ -7,7 +7,7 @@
 
 
 class BaseAutoInjectChaos(base.AutoInjectBaseTest):
-    def _test_removing_things(self, virtual_machine: _VirtualMachine, evil_command):
+    def _test_removing_things(self, virtual_machine: _VirtualMachine, evil_command: str):
         """Test break the installation and restore it.
         After breaking the installation, the app should be still working (but no sending traces to the backend).
         After breaking the installation, we can restart the app
@@ -64,7 +64,7 @@ def _test_removing_things(self, virtual_machine: _VirtualMachine, evil_command):
         logger.info("Restoring installation using the command:: ")
         apm_inject_restore = f"{prefix_env} {apm_inject_restore}"
         logger.info(apm_inject_restore)
-        _, stdout, stderr = virtual_machine.get_ssh_connection().exec_command(apm_inject_restore)
+        _, stdout, _stderr = virtual_machine.get_ssh_connection().exec_command(apm_inject_restore)
         stdout.channel.set_combine_stderr(True)
 
         # Read the output line by line

tests/auto_inject/test_blocklist_auto_inject.py

@@ -1,4 +1,5 @@
 import uuid
+import paramiko
 from scp import SCPClient
 
 from utils import scenarios, context, features, irrelevant, logger
@@ -8,7 +9,7 @@
 class _AutoInjectWorkloadSelectionBaseTest:
     """Base class to test workload selection policies on auto instrumentation."""
 
-    def _execute_remote_command(self, ssh_client, command):
+    def _execute_remote_command(self, ssh_client: paramiko.SSHClient, command: str):
         """Execute remote command and get remote log file from the vm. You can use this method using env variables or using injection config file"""
 
         unique_log_name = f"host_injection_{uuid.uuid4()}.log"

tests/auto_inject/utils.py

@@ -1,4 +1,5 @@
 import time
+import paramiko
 from utils.onboarding.weblog_interface import make_get_request, warmup_weblog, make_internal_get_request
 from utils.onboarding.backend_interface import wait_backend_trace_id
 from utils.onboarding.wait_for_tcp_port import wait_for_port
@@ -10,7 +11,12 @@
 
 class AutoInjectBaseTest:
     def _test_install(
-        self, virtual_machine, *, profile: bool = False, appsec: bool = False, origin_detection: bool = False
+        self,
+        virtual_machine: _VirtualMachine,
+        *,
+        profile: bool = False,
+        appsec: bool = False,
+        origin_detection: bool = False,
     ):
         """If there is a multicontainer app, we need to make a request to each app"""
 
@@ -32,7 +38,7 @@ def _test_install(
     def _check_install(
         self,
         virtual_machine: _VirtualMachine,
-        vm_context_url,
+        vm_context_url: str,
         *,
         profile: bool = False,
         appsec: bool = False,
@@ -75,7 +81,7 @@ def _check_install(
             self._log_trace_debug_message(e, request_uuid)
             raise
 
-    def _appsec_validator(self, _, trace_data):
+    def _appsec_validator(self, _: str, trace_data: dict):
         """Validator for Appsec traces that checks if the trace contains an Appsec event."""
         root_id = trace_data["trace"]["root_id"]
         root_span = trace_data["trace"]["spans"][root_id]
@@ -102,7 +108,7 @@ def _appsec_validator(self, _, trace_data):
         logger.error("expected 'appsec.event' to be true in trace meta or at least one rule triggered")
         return False
 
-    def _container_tags_validator(self, _, trace_data):
+    def _container_tags_validator(self, _: str, trace_data: dict):
         root_id = trace_data["trace"]["root_id"]
         root_span = trace_data["trace"]["spans"][root_id]
 
@@ -127,14 +133,14 @@ def _log_trace_debug_message(self, exc: Exception, request_uuid: str) -> None:
             f"- A problem processing the intake in the backend (manually locate the trace id [{request_uuid}] in the DD console, using the system-tests organization)\n"
         )
 
-    def close_channel(self, channel) -> None:
+    def close_channel(self, channel: paramiko.Channel) -> None:
         try:
             if not channel.eof_received:
                 channel.close()
         except Exception as e:
             logger.error(f"Error closing the channel: {e}")
 
-    def execute_command(self, virtual_machine, command) -> str:
+    def execute_command(self, virtual_machine: _VirtualMachine, command: str) -> str:
         # Env for the command
         prefix_env = ""
         for key, value in virtual_machine.get_command_environment().items():
@@ -165,7 +171,12 @@ def execute_command(self, virtual_machine, command) -> str:
             return command_output
 
     def _test_uninstall_commands(
-        self, virtual_machine, stop_weblog_command, start_weblog_command, uninstall_command, install_command
+        self,
+        virtual_machine: _VirtualMachine,
+        stop_weblog_command: str,
+        start_weblog_command: str,
+        uninstall_command: str,
+        install_command: str,
     ):
         """We can unistall the auto injection software. We can start the app again
         The weblog app should work but no sending traces to the backend.
@@ -228,7 +239,7 @@ def _test_uninstall_commands(
         self._test_install(virtual_machine)
         logger.info(f"Success _test_uninstall for : [{virtual_machine.name}]")
 
-    def _test_uninstall(self, virtual_machine):
+    def _test_uninstall(self, virtual_machine: _VirtualMachine):
         header = "----------------------------------------------------------------------"
         vm_logger(context.scenario.host_log_folder, virtual_machine.name).info(
             f"{header} \n {header}  \n  Launching the uninstall for VM: {virtual_machine.name}  \n {header} \n {header}"
@@ -250,7 +261,7 @@ def _test_uninstall(self, virtual_machine):
             virtual_machine, stop_weblog_command, start_weblog_command, uninstall_command, install_command
         )
 
-    def _test_no_world_writeable(self, virtual_machine):
+    def _test_no_world_writeable(self, virtual_machine: _VirtualMachine):
         """Checks that there are no world writeable files in /opt/datadog-packages/datadog-apm*"""
         logger = vm_logger(context.scenario.host_log_folder, virtual_machine.name)
         logger.info(

utils/_context/_scenarios/__init__.py

@@ -1094,7 +1094,7 @@ class _Scenarios:
         "DOCKER_SSI_APPSEC",
         doc="Validates the installer and the ssi on a docker environment",
         extra_env_vars={"DD_SERVICE": "payments-service"},
-        appsec_enabled="true",
+        appsec_enabled=True,
         scenario_groups=[scenario_groups.all, scenario_groups.docker_ssi],
     )
     docker_ssi_crashtracking = DockerSSIScenario(

utils/_context/_scenarios/auto_injection.py

@@ -7,7 +7,7 @@
 from utils._logger import logger
 from utils.onboarding.debug_vm import download_vm_logs
 from utils.virtual_machine.virtual_machines import _VirtualMachine, load_virtual_machines
-from .core import Scenario
+from .core import Scenario, ScenarioGroup
 
 
 class _VirtualMachineScenario(Scenario):
@@ -19,10 +19,10 @@ def __init__(
         *,
         github_workflow: str,
         doc: str,
-        vm_provision=None,
-        agent_env=None,
-        app_env=None,
-        scenario_groups=None,
+        vm_provision: str | None = None,
+        agent_env: dict | None = None,
+        app_env: dict | None = None,
+        scenario_groups: list[ScenarioGroup] | None = None,
     ) -> None:
         super().__init__(name, doc=doc, github_workflow=github_workflow, scenario_groups=scenario_groups)
         self.vm_provision_name = vm_provision
@@ -153,7 +153,7 @@ def fill_context(self):
                 # different version
                 del self.components[key]
 
-    def pytest_sessionfinish(self, session, exitstatus):  # noqa: ARG002
+    def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None:  # noqa: ARG002
         self.close_targets()
 
     def close_targets(self):
@@ -227,13 +227,13 @@ def customize_feature_parity_dashboard(self, result: dict):
 class InstallerAutoInjectionScenario(_VirtualMachineScenario):
     def __init__(
         self,
-        name,
-        doc,
-        vm_provision="installer-auto-inject",
-        agent_env=None,
-        app_env=None,
-        scenario_groups=None,
-        github_workflow=None,
+        name: str,
+        doc: str,
+        vm_provision: str = "installer-auto-inject",
+        agent_env: dict | None = None,
+        app_env: dict | None = None,
+        scenario_groups: list[ScenarioGroup] | None = None,
+        github_workflow: str | None = None,
     ) -> None:
         # Force full tracing without limits
         app_env_defaults = {