refactoring dummy event service, DRY principal; adding bulk_download_event to existing invoker tests

tests/app/services/download/test_download_queue.py

@@ -2,19 +2,17 @@
 import re
 import time
 from pathlib import Path
-from typing import Any, Dict, List
 
 import pytest
-from pydantic import BaseModel
 from pydantic.networks import AnyHttpUrl
 from requests.sessions import Session
 from requests_testadapter import TestAdapter, TestSession
 
 from invokeai.app.services.download import DownloadJob, DownloadJobStatus, DownloadQueueService
-from invokeai.app.services.events.events_base import EventServiceBase
+from tests.fixtures.event_service import DummyEventService
 
 # Prevent pytest deprecation warnings
-TestAdapter.__test__ = False
+TestAdapter.__test__ = False  # type: ignore
 
 
 @pytest.fixture
@@ -52,28 +50,6 @@ def session() -> Session:
     return sess
 
 
-class DummyEvent(BaseModel):
-    """Dummy Event to use with Dummy Event service."""
-
-    event_name: str
-    payload: Dict[str, Any]
-
-
-# A dummy event service for testing event issuing
-class DummyEventService(EventServiceBase):
-    """Dummy event service for testing."""
-
-    events: List[DummyEvent]
-
-    def __init__(self) -> None:
-        super().__init__()
-        self.events = []
-
-    def dispatch(self, event_name: str, payload: Any) -> None:
-        """Dispatch an event by appending it to self.events."""
-        self.events.append(DummyEvent(event_name=payload["event"], payload=payload["data"]))
-
-
 def test_basic_queue_download(tmp_path: Path, session: Session) -> None:
     events = set()
 

tests/app/services/model_install/test_model_install.py

@@ -10,7 +10,6 @@ from pydantic import ValidationError
 from pydantic.networks import Url
 
 from invokeai.app.services.config import InvokeAIAppConfig
-from invokeai.app.services.events.events_base import EventServiceBase
 from invokeai.app.services.model_install import (
     InstallStatus,
     LocalModelSource,

tests/fixtures/event_service.py

@@ -0,0 +1,34 @@
+from typing import Any, Dict, List
+
+import pytest
+from pydantic import BaseModel
+
+from invokeai.app.services.events.events_base import EventServiceBase
+
+
+class DummyEvent(BaseModel):
+    """Dummy Event to use with Dummy Event service."""
+
+    event_name: str
+    payload: Dict[str, Any]
+
+
+# A dummy event service for testing event issuing
+class DummyEventService(EventServiceBase):
+    """Dummy event service for testing."""
+
+    events: List[DummyEvent]
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.events = []
+
+    def dispatch(self, event_name: str, payload: Any) -> None:
+        """Dispatch an event by appending it to self.events."""
+        self.events.append(DummyEvent(event_name=payload["event"], payload=payload["data"]))
+
+
+@pytest.fixture
+def mock_event_service() -> EventServiceBase:
+    """Create a dummy event service."""
+    return DummyEventService()

tests/test_graph_execution_state.py

@@ -50,6 +50,7 @@ def mock_services() -> InvocationServices:
         board_images=None,  # type: ignore
         board_records=None,  # type: ignore
         boards=None,  # type: ignore
+        bulk_download=None,  # type: ignore
         configuration=configuration,
         events=TestEventService(),
         image_files=None,  # type: ignore