refactor: [AAP-20044] ContainerRequest defines data

src/aap_eda/core/models/event_stream.py

@@ -12,8 +12,6 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-import typing as tp
-
 from django.db import models
 
 from aap_eda.core.enums import (
@@ -112,8 +110,6 @@ def __str__(self) -> str:
         return f"EventStream {self.name} ({self.id})"
 
     # Implementation of the ContainerableMixin.
-    def get_command_line_parameters(self) -> dict[str, tp.Any]:
-        params = super().get_command_line_parameters()
-        return params | {
-            "skip_audit_events": True,
-        }
+    def _get_skip_audit_events(self) -> bool:
+        """Event stream skips audit events."""
+        return True

src/aap_eda/services/activation/engine/common.py

@@ -147,56 +147,22 @@ class ContainerableNoLatestInstanceError(ContainerableMixinError):
 #   rulebook_rulesets       str
 #
 class ContainerableMixin:
-    def get_command_line_parameters(self) -> dict[str, tp.Any]:
-        """Return parameters for running ansible-rulebook."""
-        self.validate()
-
-        access_token, refresh_token = create_jwt_token()
-        return {
-            "id": str(self.latest_instance.id),
-            "ws_url": self._get_ws_url(),
-            "log_level": self._get_log_level(),
-            "ws_ssl_verify": settings.WEBSOCKET_SSL_VERIFY,
-            "ws_token_url": self._get_ws_token_url(),
-            "ws_access_token": access_token,
-            "ws_refresh_token": refresh_token,
-            "heartbeat": settings.RULEBOOK_LIVENESS_CHECK_SECONDS,
-            "skip_audit_events": False,
-        }
-
-    def get_container_parameters(self) -> dict[str, tp.Any]:
-        """Return parameters used to create a ContainerRquest."""
-        self.validate()
-
-        return {
-            "credential": self._get_image_credential(),
-            "name": self._get_container_name(),
-            "image_url": self.decision_environment.image_url,
-            "ports": self._get_ports(),
-            "env_vars": settings.PODMAN_ENV_VARS,
-            "extra_args": settings.PODMAN_EXTRA_ARGS,
-            "mem_limit": settings.PODMAN_MEM_LIMIT,
-            "mounts": settings.PODMAN_MOUNTS,
-            "process_parent_id": self.id,
-            "rulebook_process_id": self.latest_instance.id,
-            "cmdline": self._build_cmdline(),
-        }
-
     def get_container_request(self) -> ContainerRequest:
         """Return ContainerRequest used for creation."""
-        params = self.get_container_parameters()
+        self.validate()
+
         return ContainerRequest(
-            credential=params["credential"],
-            name=params["name"],
-            image_url=params["image_url"],
-            ports=params["ports"],
-            process_parent_id=params["process_parent_id"],
-            rulebook_process_id=params["rulebook_process_id"],
-            env_vars=params["env_vars"],
-            extra_args=params["extra_args"],
-            mem_limit=params["mem_limit"],
-            mounts=params["mounts"],
-            cmdline=params["cmdline"],
+            credential=self._get_image_credential(),
+            name=self._get_container_name(),
+            image_url=self.decision_environment.image_url,
+            ports=self._get_ports(),
+            process_parent_id=self.id,
+            rulebook_process_id=self.latest_instance.id,
+            env_vars=settings.PODMAN_ENV_VARS,
+            extra_args=settings.PODMAN_EXTRA_ARGS,
+            mem_limit=settings.PODMAN_MEM_LIMIT,
+            mounts=settings.PODMAN_MOUNTS,
+            cmdline=self._build_cmdline(),
         )
 
     def get_restart_policy(self) -> str:
@@ -216,17 +182,17 @@ def validate(self):
             raise ContainerableInvalidError from e
 
     def _build_cmdline(self) -> AnsibleRulebookCmdLine:
-        params = self.get_command_line_parameters()
+        access_token, refresh_token = create_jwt_token()
         return AnsibleRulebookCmdLine(
-            ws_url=params["ws_url"],
-            log_level=params["log_level"],
-            ws_ssl_verify=params["ws_ssl_verify"],
-            ws_access_token=params["ws_access_token"],
-            ws_refresh_token=params["ws_refresh_token"],
-            ws_token_url=params["ws_token_url"],
-            heartbeat=params["heartbeat"],
-            id=params["id"],
-            skip_audit_events=params["skip_audit_events"],
+            ws_url=self._get_ws_url(),
+            log_level=self._get_log_level(),
+            ws_ssl_verify=settings.WEBSOCKET_SSL_VERIFY,
+            ws_access_token=access_token,
+            ws_refresh_token=refresh_token,
+            ws_token_url=self._get_ws_token_url(),
+            heartbeat=settings.RULEBOOK_LIVENESS_CHECK_SECONDS,
+            id=str(self.latest_instance.id),
+            skip_audit_events=self._get_skip_audit_events(),
         )
 
     def _get_log_level(self) -> tp.Optional[str]:
@@ -266,6 +232,10 @@ def _get_image_credential(self) -> tp.Optional[Credential]:
     def _get_ports(self) -> list[tuple]:
         return find_ports(self.rulebook_rulesets, self._get_context())
 
+    def _get_skip_audit_events(self) -> bool:
+        """Return not skipping audit events as default."""
+        return False
+
     def _get_ws_url(self) -> str:
         return f"{settings.WEBSOCKET_BASE_URL}{self._get_ws_url_subpath()}"
 

tests/integration/services/activation/test_activation.py

@@ -17,6 +17,7 @@
 
 from aap_eda.core import models
 from aap_eda.services.activation.engine.common import (
+    AnsibleRulebookCmdLine,
     ContainerableInvalidError,
     ContainerRequest,
 )
@@ -53,47 +54,42 @@ def activation(activation_no_instance) -> models.Activation:
 
 
 @pytest.mark.django_db
-def test_command_line_parameters(activation):
-    params = activation.get_command_line_parameters()
-    assert params["ws_url"] is not None
-    assert params["log_level"] is None
-    assert params["ws_ssl_verify"] is not None
-    assert params["ws_token_url"] is not None
-    assert params["ws_access_token"] is not None
-    assert params["ws_refresh_token"] is not None
-    assert params["heartbeat"] is not None
-    assert not params["skip_audit_events"]
-    assert params["id"] == str(activation.latest_instance.id)
-
-
-@pytest.mark.django_db
-def test_container_parameters(activation):
-    params = activation.get_container_parameters()
-    assert params["name"] is not None
-    assert params["image_url"] is not None
-    assert params["ports"] is not None
-    assert params["env_vars"] is not None
-    assert params["extra_args"] is not None
-    assert params["mem_limit"] is not None
-    assert params["mounts"] is not None
-    assert params["process_parent_id"] == activation.id
-    assert params["rulebook_process_id"] == activation.latest_instance.id
-
-
-@pytest.mark.django_db
-def test_container_parameters_no_credential(activation):
+def test_container_request_no_credential(activation):
     """Test container params when no credential exists."""
-    params = activation.get_container_parameters()
-    assert params["credential"] is None
+    request = activation.get_container_request()
+    assert request.credential is None
 
 
 @pytest.mark.django_db
 def test_get_container_request(activation):
     """Test the construction of a ContainerRequest."""
     request = activation.get_container_request()
+
     assert isinstance(request, ContainerRequest)
-    assert not request.cmdline.skip_audit_events
-    assert "--skip-audit-events" not in request.cmdline.get_args()
+    assert request.name is not None
+    assert request.image_url is not None
+    assert request.ports is not None
+    assert request.env_vars is not None
+    assert request.extra_args is not None
+    assert request.mem_limit is not None
+    assert request.mounts is not None
+    assert request.process_parent_id == activation.id
+    assert request.rulebook_process_id == activation.latest_instance.id
+
+    cmdline = request.cmdline
+    assert (cmdline is not None) and isinstance(
+        cmdline, AnsibleRulebookCmdLine
+    )
+    assert cmdline.id == str(activation.latest_instance.id)
+    assert cmdline.ws_url is not None
+    assert cmdline.log_level is None
+    assert cmdline.ws_ssl_verify is not None
+    assert cmdline.ws_token_url is not None
+    assert cmdline.ws_access_token is not None
+    assert cmdline.ws_refresh_token is not None
+    assert cmdline.heartbeat is not None
+    assert not cmdline.skip_audit_events
+    assert "--skip-audit-events" not in cmdline.get_args()
 
 
 @pytest.mark.django_db

tests/integration/services/activation/test_event_stream.py

@@ -17,6 +17,7 @@
 
 from aap_eda.core import models
 from aap_eda.services.activation.engine.common import (
+    AnsibleRulebookCmdLine,
     ContainerableInvalidError,
     ContainerRequest,
 )
@@ -43,7 +44,7 @@ def event_stream_no_instance(
 
 @pytest.fixture
 def event_stream(event_stream_no_instance) -> models.EventStream:
-    """Return an activation with associated RulebookProcess."""
+    """Return an event stream with associated RulebookProcess."""
     models.RulebookProcess.objects.create(
         name="event-stream-instance-1",
         event_stream=event_stream_no_instance,
@@ -53,47 +54,42 @@ def event_stream(event_stream_no_instance) -> models.EventStream:
 
 
 @pytest.mark.django_db
-def test_command_line_parameters(event_stream):
-    params = event_stream.get_command_line_parameters()
-    assert params["ws_url"] is not None
-    assert params["log_level"] is None
-    assert params["ws_ssl_verify"] is not None
-    assert params["ws_token_url"] is not None
-    assert params["ws_access_token"] is not None
-    assert params["ws_refresh_token"] is not None
-    assert params["heartbeat"] is not None
-    assert params["skip_audit_events"]
-    assert params["id"] == str(event_stream.latest_instance.id)
-
-
-@pytest.mark.django_db
-def test_container_parameters(event_stream):
-    params = event_stream.get_container_parameters()
-    assert params["name"] is not None
-    assert params["image_url"] is not None
-    assert params["ports"] is not None
-    assert params["env_vars"] is not None
-    assert params["extra_args"] is not None
-    assert params["mem_limit"] is not None
-    assert params["mounts"] is not None
-    assert params["process_parent_id"] == event_stream.id
-    assert params["rulebook_process_id"] == event_stream.latest_instance.id
-
-
-@pytest.mark.django_db
-def test_container_parameters_no_credential(event_stream):
+def test_container_request_no_credential(event_stream):
     """Test container params when no credential exists."""
-    params = event_stream.get_container_parameters()
-    assert params["credential"] is None
+    request = event_stream.get_container_request()
+    assert request.credential is None
 
 
 @pytest.mark.django_db
 def test_get_container_request(event_stream):
     """Test the construction of a ContainerRequest."""
     request = event_stream.get_container_request()
+
     assert isinstance(request, ContainerRequest)
-    assert request.cmdline.skip_audit_events
-    assert "--skip-audit-events" in request.cmdline.get_args()
+    assert request.name is not None
+    assert request.image_url is not None
+    assert request.ports is not None
+    assert request.env_vars is not None
+    assert request.extra_args is not None
+    assert request.mem_limit is not None
+    assert request.mounts is not None
+    assert request.process_parent_id == event_stream.id
+    assert request.rulebook_process_id == event_stream.latest_instance.id
+
+    cmdline = request.cmdline
+    assert (cmdline is not None) and isinstance(
+        cmdline, AnsibleRulebookCmdLine
+    )
+    assert cmdline.id == str(event_stream.latest_instance.id)
+    assert cmdline.ws_url is not None
+    assert cmdline.log_level is None
+    assert cmdline.ws_ssl_verify is not None
+    assert cmdline.ws_token_url is not None
+    assert cmdline.ws_access_token is not None
+    assert cmdline.ws_refresh_token is not None
+    assert cmdline.heartbeat is not None
+    assert cmdline.skip_audit_events
+    assert "--skip-audit-events" in cmdline.get_args()
 
 
 @pytest.mark.django_db