test: adding backend/e2e tests (#97)

* missing e2e tests
* all evetnt types as strs -> EventType; updated usage places
* updated tests - tests for happy paths are using pydantic models from now on
* resource cleaner fix: getting k8s stuff from DI
* tests update - removed kafka prefix, simplified multithread testing in conftest
* removed integration tests, moved all of them to e2e, also updated docs
* sse status enum, fixed tests
* fixes + no if status=200 checks, using asserts
* routing notif service same as sse bridge and event store consumer - via Lifecycle
* fixed problem with sagas

Author: HardMax71

.github/workflows/stack-tests.yml

@@ -185,76 +185,7 @@ jobs:
           path: /tmp/all-images.tar.zst
           retention-days: 1
 
-  # Three parallel test jobs
-  backend-integration:
-    name: Backend Integration Tests
-    needs: [build-images]
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v6
-
-      - name: Cache and load Docker images
-        uses: ./.github/actions/docker-cache
-        with:
-          images: ${{ env.MONGO_IMAGE }} ${{ env.REDIS_IMAGE }} ${{ env.KAFKA_IMAGE }} ${{ env.ZOOKEEPER_IMAGE }} ${{ env.SCHEMA_REGISTRY_IMAGE }}
-
-      - name: Download built images
-        uses: actions/download-artifact@v7
-        with:
-          name: docker-images
-          path: /tmp
-
-      - name: Load built images
-        run: zstd -d -c /tmp/all-images.tar.zst | docker load
-
-      - name: Setup k3s
-        uses: ./.github/actions/k3s-setup
-
-      - name: Use test environment config
-        run: cp backend/.env.test backend/.env
-
-      - name: Start stack
-        run: ./deploy.sh dev --wait
-
-      - name: Run integration tests
-        timeout-minutes: 10
-        run: |
-          docker compose exec -T -e TEST_RUN_ID=integration backend \
-            uv run pytest tests/integration -v -rs \
-            --durations=0 \
-            --cov=app \
-            --cov-report=xml:coverage-integration.xml \
-            --cov-report=term
-
-      - name: Copy coverage
-        if: always()
-        run: docker compose cp backend:/app/coverage-integration.xml backend/coverage-integration.xml || true
-
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v5
-        if: always()
-        with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          files: backend/coverage-integration.xml
-          flags: backend-integration
-          name: backend-integration-coverage
-          fail_ci_if_error: false
-
-      - name: Collect logs on failure
-        if: failure()
-        run: |
-          mkdir -p logs
-          docker compose logs > logs/docker-compose.log 2>&1
-          docker compose logs backend > logs/backend.log 2>&1
-          docker compose logs kafka > logs/kafka.log 2>&1
-
-      - name: Upload logs
-        if: failure()
-        uses: actions/upload-artifact@v6
-        with:
-          name: backend-integration-logs
-          path: logs/
-
+  # Parallel test jobs (backend-e2e, frontend-e2e)
   backend-e2e:
     name: Backend E2E Tests
     needs: [build-images]
@@ -289,9 +220,9 @@ jobs:
         run: docker compose exec -T backend uv run python scripts/seed_users.py
 
       - name: Run E2E tests
-        timeout-minutes: 10
+        timeout-minutes: 15
         run: |
-          docker compose exec -T -e TEST_RUN_ID=e2e backend \
+          docker compose exec -T backend \
             uv run pytest tests/e2e -v -rs \
             --durations=0 \
             --cov=app \

backend/app/api/routes/admin/events.py

@@ -77,7 +77,7 @@ async def export_events_csv(
 ) -> StreamingResponse:
     try:
         export_filter = EventFilter(
-            event_types=[str(et) for et in event_types] if event_types else None,
+            event_types=event_types,
             start_time=start_time,
             end_time=end_time,
         )
@@ -107,7 +107,7 @@ async def export_events_json(
     """Export events as JSON with comprehensive filtering."""
     try:
         export_filter = EventFilter(
-            event_types=[str(et) for et in event_types] if event_types else None,
+            event_types=event_types,
             aggregate_id=aggregate_id,
             correlation_id=correlation_id,
             user_id=user_id,

backend/app/api/routes/admin/users.py

@@ -7,18 +7,19 @@
 from app.api.dependencies import admin_user
 from app.db.repositories.admin.admin_user_repository import AdminUserRepository
 from app.domain.enums.user import UserRole
-from app.domain.rate_limit import UserRateLimit
+from app.domain.rate_limit import RateLimitRule, UserRateLimit
 from app.domain.user import UserUpdate as DomainUserUpdate
 from app.schemas_pydantic.admin_user_overview import (
     AdminUserOverview,
     DerivedCounts,
     RateLimitSummary,
 )
-from app.schemas_pydantic.events import EventResponse, EventStatistics
+from app.schemas_pydantic.events import EventStatistics
 from app.schemas_pydantic.user import (
     DeleteUserResponse,
     MessageResponse,
     PasswordResetRequest,
+    RateLimitUpdateRequest,
     RateLimitUpdateResponse,
     UserCreate,
     UserListResponse,
@@ -27,7 +28,6 @@
     UserUpdate,
 )
 from app.services.admin import AdminUserService
-from app.services.rate_limit_service import RateLimitService
 
 router = APIRouter(
     prefix="/admin/users", tags=["admin", "users"], route_class=DishkaRoute, dependencies=[Depends(admin_user)]
@@ -38,7 +38,6 @@
 async def list_users(
     admin: Annotated[UserResponse, Depends(admin_user)],
     admin_user_service: FromDishka[AdminUserService],
-    rate_limit_service: FromDishka[RateLimitService],
     limit: int = Query(default=100, le=1000),
     offset: int = Query(default=0, ge=0),
     search: str | None = None,
@@ -51,24 +50,8 @@ async def list_users(
         search=search,
         role=role,
     )
-
-    summaries = await rate_limit_service.get_user_rate_limit_summaries([u.user_id for u in result.users])
-    user_responses: list[UserResponse] = []
-    for user in result.users:
-        user_response = UserResponse.model_validate(user)
-        summary = summaries.get(user.user_id)
-        if summary:
-            user_response = user_response.model_copy(
-                update={
-                    "bypass_rate_limit": summary.bypass_rate_limit,
-                    "global_multiplier": summary.global_multiplier,
-                    "has_custom_limits": summary.has_custom_limits,
-                }
-            )
-        user_responses.append(user_response)
-
     return UserListResponse(
-        users=user_responses,
+        users=[UserResponse.model_validate(u) for u in result.users],
         total=result.total,
         offset=result.offset,
         limit=result.limit,
@@ -119,7 +102,7 @@ async def get_user_overview(
         stats=EventStatistics.model_validate(domain.stats),
         derived_counts=DerivedCounts.model_validate(domain.derived_counts),
         rate_limit_summary=RateLimitSummary.model_validate(domain.rate_limit_summary),
-        recent_events=[EventResponse.model_validate(e).model_dump() for e in domain.recent_events],
+        recent_events=domain.recent_events,
     )
 
 
@@ -165,13 +148,19 @@ async def delete_user(
     if admin.user_id == user_id:
         raise HTTPException(status_code=400, detail="Cannot delete your own account")
 
-    deleted_counts = await admin_user_service.delete_user(
+    result = await admin_user_service.delete_user(
         admin_username=admin.username, user_id=user_id, cascade=cascade
     )
-    if deleted_counts.get("user", 0) == 0:
-        raise HTTPException(status_code=500, detail="Failed to delete user")
-
-    return DeleteUserResponse(message=f"User {user_id} deleted successfully", deleted_counts=deleted_counts)
+    return DeleteUserResponse(
+        message=f"User {user_id} deleted successfully",
+        user_deleted=result.user_deleted,
+        executions=result.executions,
+        saved_scripts=result.saved_scripts,
+        notifications=result.notifications,
+        user_settings=result.user_settings,
+        events=result.events,
+        sagas=result.sagas,
+    )
 
 
 @router.post("/{user_id}/reset-password", response_model=MessageResponse)
@@ -204,10 +193,15 @@ async def update_user_rate_limits(
     admin: Annotated[UserResponse, Depends(admin_user)],
     admin_user_service: FromDishka[AdminUserService],
     user_id: str,
-    rate_limit_config: UserRateLimit,
+    request: RateLimitUpdateRequest,
 ) -> RateLimitUpdateResponse:
+    config = UserRateLimit(
+        user_id=user_id,
+        rules=[RateLimitRule(**r.model_dump()) for r in request.rules],
+        **request.model_dump(exclude={"rules"}),
+    )
     result = await admin_user_service.update_user_rate_limits(
-        admin_username=admin.username, user_id=user_id, config=rate_limit_config
+        admin_username=admin.username, user_id=user_id, config=config
     )
     return RateLimitUpdateResponse.model_validate(result)
 

backend/app/api/routes/events.py

@@ -6,13 +6,16 @@
 from dishka import FromDishka
 from dishka.integrations.fastapi import DishkaRoute
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from pydantic import TypeAdapter
 
 from app.api.dependencies import admin_user, current_user
 from app.core.correlation import CorrelationContext
 from app.core.utils import get_client_ip
 from app.domain.enums.common import SortOrder
+from app.domain.enums.events import EventType
+from app.domain.enums.user import UserRole
 from app.domain.events.event_models import EventFilter
-from app.domain.events.typed import BaseEvent, EventMetadata
+from app.domain.events.typed import BaseEvent, DomainEvent, EventMetadata
 from app.schemas_pydantic.events import (
     DeleteEventResponse,
     EventAggregationRequest,
@@ -26,9 +29,12 @@
 )
 from app.schemas_pydantic.user import UserResponse
 from app.services.event_service import EventService
+from app.services.execution_service import ExecutionService
 from app.services.kafka_event_service import KafkaEventService
 from app.settings import Settings
 
+_event_response_list_adapter: TypeAdapter[list[EventResponse]] = TypeAdapter(list[EventResponse])
+
 router = APIRouter(prefix="/events", tags=["events"], route_class=DishkaRoute)
 
 
@@ -37,10 +43,16 @@ async def get_execution_events(
     execution_id: str,
     current_user: Annotated[UserResponse, Depends(current_user)],
     event_service: FromDishka[EventService],
+    execution_service: FromDishka[ExecutionService],
     include_system_events: bool = Query(False, description="Include system-generated events"),
     limit: int = Query(100, ge=1, le=1000),
     skip: int = Query(0, ge=0),
 ) -> EventListResponse:
+    # Check execution ownership first (before checking events)
+    execution = await execution_service.get_execution_result(execution_id)
+    if execution.user_id and execution.user_id != current_user.user_id and current_user.role != UserRole.ADMIN:
+        raise HTTPException(status_code=403, detail="Access denied")
+
     result = await event_service.get_execution_events(
         execution_id=execution_id,
         user_id=current_user.user_id,
@@ -53,10 +65,8 @@ async def get_execution_events(
     if result is None:
         raise HTTPException(status_code=403, detail="Access denied")
 
-    event_responses = [EventResponse.model_validate(event) for event in result.events]
-
     return EventListResponse(
-        events=event_responses,
+        events=_event_response_list_adapter.validate_python(result.events, from_attributes=True),
         total=result.total,
         limit=limit,
         skip=skip,
@@ -68,7 +78,7 @@ async def get_execution_events(
 async def get_user_events(
     current_user: Annotated[UserResponse, Depends(current_user)],
     event_service: FromDishka[EventService],
-    event_types: List[str] | None = Query(None),
+    event_types: List[EventType] | None = Query(None),
     start_time: datetime | None = Query(None),
     end_time: datetime | None = Query(None),
     limit: int = Query(100, ge=1, le=1000),
@@ -86,10 +96,12 @@ async def get_user_events(
         sort_order=sort_order,
     )
 
-    event_responses = [EventResponse.model_validate(event) for event in result.events]
-
     return EventListResponse(
-        events=event_responses, total=result.total, limit=limit, skip=skip, has_more=result.has_more
+        events=_event_response_list_adapter.validate_python(result.events, from_attributes=True),
+        total=result.total,
+        limit=limit,
+        skip=skip,
+        has_more=result.has_more,
     )
 
 
@@ -100,7 +112,7 @@ async def query_events(
     event_service: FromDishka[EventService],
 ) -> EventListResponse:
     event_filter = EventFilter(
-        event_types=[str(et) for et in filter_request.event_types] if filter_request.event_types else None,
+        event_types=filter_request.event_types,
         aggregate_id=filter_request.aggregate_id,
         correlation_id=filter_request.correlation_id,
         user_id=filter_request.user_id,
@@ -121,10 +133,12 @@ async def query_events(
     if result is None:
         raise HTTPException(status_code=403, detail="Cannot query other users' events")
 
-    event_responses = [EventResponse.model_validate(event) for event in result.events]
-
     return EventListResponse(
-        events=event_responses, total=result.total, limit=result.limit, skip=result.skip, has_more=result.has_more
+        events=_event_response_list_adapter.validate_python(result.events, from_attributes=True),
+        total=result.total,
+        limit=result.limit,
+        skip=result.skip,
+        has_more=result.has_more,
     )
 
 
@@ -146,10 +160,8 @@ async def get_events_by_correlation(
         skip=skip,
     )
 
-    event_responses = [EventResponse.model_validate(event) for event in result.events]
-
     return EventListResponse(
-        events=event_responses,
+        events=_event_response_list_adapter.validate_python(result.events, from_attributes=True),
         total=result.total,
         limit=limit,
         skip=skip,
@@ -177,10 +189,8 @@ async def get_current_request_events(
         skip=skip,
     )
 
-    event_responses = [EventResponse.model_validate(event) for event in result.events]
-
     return EventListResponse(
-        events=event_responses,
+        events=_event_response_list_adapter.validate_python(result.events, from_attributes=True),
         total=result.total,
         limit=limit,
         skip=skip,
@@ -212,15 +222,15 @@ async def get_event_statistics(
     return EventStatistics.model_validate(stats)
 
 
-@router.get("/{event_id}", response_model=EventResponse)
+@router.get("/{event_id}", response_model=DomainEvent)
 async def get_event(
     event_id: str, current_user: Annotated[UserResponse, Depends(current_user)], event_service: FromDishka[EventService]
-) -> EventResponse:
+) -> DomainEvent:
     """Get a specific event by ID"""
     event = await event_service.get_event(event_id=event_id, user_id=current_user.user_id, user_role=current_user.role)
     if event is None:
         raise HTTPException(status_code=404, detail="Event not found")
-    return EventResponse.model_validate(event)
+    return event
 
 
 @router.post("/publish", response_model=PublishEventResponse)