moved k8s tests to e2e, fixed inconsistencies in user settings

Author: HardMax71

.github/actions/docker-cache/action.yml

@@ -12,9 +12,12 @@ runs:
     - name: Generate cache key from images
       id: cache-key
       shell: bash
+      env:
+        IMAGES_INPUT: ${{ inputs.images }}
       run: |
         # Create a stable hash from the sorted image list
-        IMAGES_HASH=$(echo "${{ inputs.images }}" | tr ' ' '\n' | sort | md5sum | cut -d' ' -f1)
+        # Using env var to prevent script injection
+        IMAGES_HASH=$(echo "$IMAGES_INPUT" | tr ' ' '\n' | sort | md5sum | cut -d' ' -f1)
         echo "key=docker-${{ runner.os }}-${IMAGES_HASH}" >> $GITHUB_OUTPUT
 
     - name: Cache Docker images
@@ -38,17 +41,19 @@ runs:
     - name: Pull and save Docker images
       if: steps.docker-cache.outputs.cache-hit != 'true'
       shell: bash
+      env:
+        IMAGES_INPUT: ${{ inputs.images }}
       run: |
         mkdir -p /tmp/docker-cache
 
         echo "Pulling images in parallel..."
-        for img in ${{ inputs.images }}; do
+        for img in $IMAGES_INPUT; do
           docker pull "$img" &
         done
         wait
 
         echo "Saving images with zstd compression..."
-        for img in ${{ inputs.images }}; do
+        for img in $IMAGES_INPUT; do
           # Create filename from image name (replace special chars)
           filename=$(echo "$img" | tr '/:' '_')
           docker save "$img" | zstd -T0 -3 > "/tmp/docker-cache/${filename}.tar.zst" &

backend/app/infrastructure/kafka/events/user.py

@@ -51,7 +51,7 @@ class UserSettingsUpdatedEvent(BaseEvent):
     topic: ClassVar[KafkaTopic] = KafkaTopic.USER_SETTINGS_EVENTS
     user_id: str
     settings_type: SettingsType
-    changes: dict[str, str]
+    updated: dict[str, str]
 
 
 class UserThemeChangedEvent(BaseEvent):

backend/app/services/user_settings_service.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 from datetime import datetime, timedelta, timezone
 from typing import Any, List
 
@@ -182,17 +183,17 @@ async def update_user_settings(
                 settings_type = SettingsType.DISPLAY
             else:
                 settings_type = SettingsType.PREFERENCES
-            # Flatten changes to string map for the generic event
-            changes: dict[str, str] = {}
-            for k, v in updated.items():
-                changes[k] = str(v)
+            # Stringify all values for Avro compatibility (nested dicts become JSON strings)
+            updated_stringified: dict[str, str] = {
+                k: json.dumps(v) if isinstance(v, dict) else str(v) for k, v in updated.items()
+            }
             await self.event_service.publish_event(
                 event_type=EventType.USER_SETTINGS_UPDATED,
                 aggregate_id=f"user_settings_{user_id}",
                 payload={
                     "user_id": user_id,
                     "settings_type": settings_type,
-                    "changes": changes,
+                    "updated": updated_stringified,
                     "reason": reason,
                 },
                 metadata=None,
@@ -297,7 +298,7 @@ async def restore_settings_to_point(self, user_id: str, timestamp: datetime) ->
             payload={
                 "user_id": user_id,
                 "settings_type": SettingsType.PREFERENCES,
-                "changes": {"restored_to": timestamp.isoformat()},
+                "updated": {"restored_to": timestamp.isoformat()},
             },
             metadata=None,
         )
@@ -339,32 +340,22 @@ def _apply_event(self, settings: DomainUserSettings, event: DomainSettingsEvent)
                 settings.theme = Theme(new_theme)
             return settings
 
-        upd = event.payload.get("updated")
-        if not upd:
-            return settings
-
-        # Top-level
-        if "theme" in upd:
-            settings.theme = Theme(upd["theme"])
-        if "timezone" in upd:
-            settings.timezone = upd["timezone"]
-        if "date_format" in upd:
-            settings.date_format = upd["date_format"]
-        if "time_format" in upd:
-            settings.time_format = upd["time_format"]
-        # Nested
-        if "notifications" in upd and isinstance(upd["notifications"], dict):
-            n = upd["notifications"]
-            channels: list[NotificationChannel] = [NotificationChannel(c) for c in n.get("channels", [])]
+        if event.event_type == EventType.USER_NOTIFICATION_SETTINGS_UPDATED:
+            n = event.payload.get("settings", {})
+            channels_raw = event.payload.get("channels", [])
+            channels: list[NotificationChannel] = [NotificationChannel(c) for c in channels_raw] if channels_raw else []
             settings.notifications = DomainNotificationSettings(
                 execution_completed=n.get("execution_completed", settings.notifications.execution_completed),
                 execution_failed=n.get("execution_failed", settings.notifications.execution_failed),
                 system_updates=n.get("system_updates", settings.notifications.system_updates),
                 security_alerts=n.get("security_alerts", settings.notifications.security_alerts),
                 channels=channels or settings.notifications.channels,
             )
-        if "editor" in upd and isinstance(upd["editor"], dict):
-            e = upd["editor"]
+            settings.updated_at = event.timestamp
+            return settings
+
+        if event.event_type == EventType.USER_EDITOR_SETTINGS_UPDATED:
+            e = event.payload.get("settings", {})
             settings.editor = DomainEditorSettings(
                 theme=e.get("theme", settings.editor.theme),
                 font_size=e.get("font_size", settings.editor.font_size),
@@ -373,8 +364,58 @@ def _apply_event(self, settings: DomainUserSettings, event: DomainSettingsEvent)
                 word_wrap=e.get("word_wrap", settings.editor.word_wrap),
                 show_line_numbers=e.get("show_line_numbers", settings.editor.show_line_numbers),
             )
-        if "custom_settings" in upd and isinstance(upd["custom_settings"], dict):
-            settings.custom_settings = upd["custom_settings"]
+            settings.updated_at = event.timestamp
+            return settings
+
+        upd = event.payload.get("updated")
+        if not upd:
+            return settings
+
+        # Helper to parse JSON strings or return dict as-is
+        def parse_value(val: object) -> object:
+            if isinstance(val, str):
+                try:
+                    return json.loads(val)
+                except (json.JSONDecodeError, ValueError):
+                    return val
+            return val
+
+        # Top-level
+        if "theme" in upd:
+            settings.theme = Theme(str(upd["theme"]))
+        if "timezone" in upd:
+            settings.timezone = str(upd["timezone"])
+        if "date_format" in upd:
+            settings.date_format = str(upd["date_format"])
+        if "time_format" in upd:
+            settings.time_format = str(upd["time_format"])
+        # Nested (may be JSON strings or dicts)
+        if "notifications" in upd:
+            n = parse_value(upd["notifications"])
+            if isinstance(n, dict):
+                channels: list[NotificationChannel] = [NotificationChannel(c) for c in n.get("channels", [])]
+                settings.notifications = DomainNotificationSettings(
+                    execution_completed=n.get("execution_completed", settings.notifications.execution_completed),
+                    execution_failed=n.get("execution_failed", settings.notifications.execution_failed),
+                    system_updates=n.get("system_updates", settings.notifications.system_updates),
+                    security_alerts=n.get("security_alerts", settings.notifications.security_alerts),
+                    channels=channels or settings.notifications.channels,
+                )
+        if "editor" in upd:
+            e = parse_value(upd["editor"])
+            if isinstance(e, dict):
+                settings.editor = DomainEditorSettings(
+                    theme=e.get("theme", settings.editor.theme),
+                    font_size=e.get("font_size", settings.editor.font_size),
+                    tab_size=e.get("tab_size", settings.editor.tab_size),
+                    use_tabs=e.get("use_tabs", settings.editor.use_tabs),
+                    word_wrap=e.get("word_wrap", settings.editor.word_wrap),
+                    show_line_numbers=e.get("show_line_numbers", settings.editor.show_line_numbers),
+                )
+        if "custom_settings" in upd:
+            cs = parse_value(upd["custom_settings"])
+            if isinstance(cs, dict):
+                settings.custom_settings = cs
         settings.version = event.payload.get("version", settings.version)
         settings.updated_at = event.timestamp
         return settings

…sts/integration/test_execution_routes.py …integration/k8s/test_execution_routes.pybackend/tests/integration/test_execution_routes.py renamed to backend/tests/integration/k8s/test_execution_routes.py backend/tests/integration/test_execution_routes.py renamed to backend/tests/integration/k8s/test_execution_routes.py

@@ -14,23 +14,7 @@
 )
 
 
-def has_k8s_workers() -> bool:
-    """Check if K8s workers are available for execution."""
-    # Check if K8s worker container is running
-    import subprocess
-    try:
-        result = subprocess.run(
-            ["docker", "ps", "--filter", "name=k8s-worker", "--format", "{{.Names}}"],
-            capture_output=True,
-            text=True,
-            timeout=2
-        )
-        return "k8s-worker" in result.stdout
-    except Exception:
-        return False
-
-
-@pytest.mark.integration
+@pytest.mark.k8s
 class TestExecution:
     """Test execution endpoints against real backend."""
 
@@ -52,7 +36,6 @@ async def test_execute_requires_authentication(self, client: AsyncClient) -> Non
                    for word in ["not authenticated", "unauthorized", "login"])
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execute_simple_python_script(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test executing a simple Python script."""
         # Login first
@@ -96,7 +79,6 @@ async def test_execute_simple_python_script(self, client: AsyncClient, test_user
         ]
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_get_execution_result(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test getting execution result after completion using SSE (event-driven)."""
         # Login first
@@ -137,7 +119,6 @@ async def test_get_execution_result(self, client: AsyncClient, test_user: Dict[s
             assert "Line 2" in execution_result.stdout
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execute_with_error(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test executing a script that produces an error."""
         # Login first
@@ -163,7 +144,6 @@ async def test_execute_with_error(self, client: AsyncClient, test_user: Dict[str
         # No waiting - execution was accepted, error will be processed asynchronously
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execute_with_resource_tracking(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test that execution tracks resource usage."""
         # Login first
@@ -205,7 +185,6 @@ async def test_execute_with_resource_tracking(self, client: AsyncClient, test_us
                 assert resource_usage.peak_memory_kb >= 0
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execute_with_different_language_versions(self, client: AsyncClient,
                                                             test_user: Dict[str, str]) -> None:
         """Test execution with different Python versions."""
@@ -240,7 +219,6 @@ async def test_execute_with_different_language_versions(self, client: AsyncClien
                 assert "execution_id" in data
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execute_with_large_output(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test execution with large output."""
         # Login first
@@ -279,7 +257,6 @@ async def test_execute_with_large_output(self, client: AsyncClient, test_user: D
                 assert "End of output" in result_data["stdout"] or len(result_data["stdout"]) > 10000
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_cancel_running_execution(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test cancelling a running execution."""
         # Login first
@@ -326,7 +303,6 @@ async def test_cancel_running_execution(self, client: AsyncClient, test_user: Di
         # Cancel response of 200 means cancellation was accepted
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execution_with_timeout(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Bounded check: long-running executions don't finish immediately.
 
@@ -364,7 +340,6 @@ async def test_execution_with_timeout(self, client: AsyncClient, test_user: Dict
         # No need to wait or observe states
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_sandbox_restrictions(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test that dangerous operations are blocked by sandbox."""
         # Login first
@@ -421,7 +396,6 @@ async def test_sandbox_restrictions(self, client: AsyncClient, test_user: Dict[s
                 assert exec_response.status_code in [400, 422]
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_concurrent_executions_by_same_user(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """Test running multiple executions concurrently."""
         # Login first
@@ -489,7 +463,6 @@ async def test_get_k8s_resource_limits(self, client: AsyncClient) -> None:
             assert key in limits
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_get_user_executions_list(self, client: AsyncClient, test_user: Dict[str, str]) -> None:
         """User executions list returns paginated executions for current user."""
         # Login first
@@ -504,7 +477,6 @@ async def test_get_user_executions_list(self, client: AsyncClient, test_user: Di
         assert set(["executions", "total", "limit", "skip", "has_more"]).issubset(payload.keys())
 
     @pytest.mark.asyncio
-    @pytest.mark.skipif(not has_k8s_workers(), reason="K8s workers not available")
     async def test_execution_idempotency_same_key_returns_same_execution(self, client: AsyncClient,
                                                                          test_user: Dict[str, str]) -> None:
         """Submitting the same request with the same Idempotency-Key yields the same execution_id."""