Add dummy runtime/deployment

docs/api/deployments/dummy.md

@@ -0,0 +1 @@
+::: swerex.deployment.dummy.DummyDeployment

docs/api/runtimes/dummy.md

@@ -0,0 +1 @@
+::: swerex.runtime.dummy.DummyRuntime

mkdocs.yml

@@ -61,11 +61,13 @@ nav:
       - Docker: api/deployments/docker.md
       - Modal: api/deployments/modal.md
       - Fargate: api/deployments/fargate.md
+      - Dummy: api/deployments/dummy.md
     - Runtimes:
       - Abstract: api/runtimes/abstract.md
       - Local: api/runtimes/local.md
       - Remote: api/runtimes/remote.md
       - Data: api/runtimes/data.md
+      - Dummy: api/runtimes/dummy.md
     - Server: api/server.md
 plugins:
    - glightbox

src/swerex/deployment/__init__.py

@@ -6,6 +6,10 @@
 def get_deployment(
     deployment_type: Literal["local", "docker", "modal", "fargate", "remote"], **kwargs
 ) -> AbstractDeployment:
+    if deployment_type == "dummy":
+        from swerex.deployment.dummy import DummyDeployment
+
+        return DummyDeployment(**kwargs)
     if deployment_type == "local":
         from swerex.deployment.local import LocalDeployment
 

src/swerex/deployment/dummy.py

@@ -0,0 +1,29 @@
+from swerex.deployment.abstract import AbstractDeployment
+from swerex.runtime.abstract import IsAliveResponse
+from swerex.runtime.dummy import DummyRuntime
+
+
+class DummyDeployment(AbstractDeployment):
+    """This deployment does nothing.
+    Useful for testing.
+    """
+
+    def __init__(self):
+        self._runtime = DummyRuntime()  # type: ignore
+
+    async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
+        return IsAliveResponse(is_alive=True)
+
+    async def start(self):
+        pass
+
+    async def stop(self):
+        pass
+
+    @property
+    def runtime(self) -> DummyRuntime:
+        return self._runtime
+
+    @runtime.setter
+    def runtime(self, runtime: DummyRuntime):
+        self._runtime = runtime

src/swerex/runtime/abstract.py

@@ -238,12 +238,12 @@ async def create_session(self, request: CreateSessionRequest) -> CreateSessionRe
         pass
 
     @abstractmethod
-    async def run_in_session(self, action: BashAction) -> BashObservation:
+    async def run_in_session(self, action: Action) -> Observation:
         """Runs a command in a session."""
         pass
 
     @abstractmethod
-    async def close_session(self, request: CloseSessionRequest):
+    async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse:
         """Closes a shell session."""
         pass
 

src/swerex/runtime/dummy.py

@@ -0,0 +1,64 @@
+from swerex.runtime.abstract import (
+    AbstractRuntime,
+    Action,
+    BashObservation,
+    CloseBashSessionResponse,
+    CloseResponse,
+    CloseSessionRequest,
+    CloseSessionResponse,
+    Command,
+    CommandResponse,
+    CreateBashSessionResponse,
+    CreateSessionRequest,
+    CreateSessionResponse,
+    IsAliveResponse,
+    Observation,
+    ReadFileRequest,
+    ReadFileResponse,
+    UploadRequest,
+    UploadResponse,
+    WriteFileRequest,
+    WriteFileResponse,
+)
+
+
+class DummyRuntime(AbstractRuntime):
+    """This runtime does nothing.
+    Useful for testing.
+    """
+
+    async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
+        return IsAliveResponse(is_alive=True)
+
+    async def create_session(self, request: CreateSessionRequest) -> CreateSessionResponse:
+        if request.session_type == "bash":
+            return CreateBashSessionResponse()
+        msg = f"Unknown session type: {request.session_type}"
+        raise ValueError(msg)
+
+    async def run_in_session(self, action: Action) -> Observation:
+        if action.session_type == "bash":
+            return BashObservation(exit_code=0)
+        msg = f"Unknown session type: {action.session_type}"
+        raise ValueError(msg)
+
+    async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse:
+        if request.session_type == "bash":
+            return CloseBashSessionResponse()
+        msg = f"Unknown session type: {request.session_type}"
+        raise ValueError(msg)
+
+    async def execute(self, command: Command) -> CommandResponse:
+        return CommandResponse(exit_code=0)
+
+    async def read_file(self, request: ReadFileRequest) -> ReadFileResponse:
+        return ReadFileResponse()
+
+    async def write_file(self, request: WriteFileRequest) -> WriteFileResponse:
+        return WriteFileResponse()
+
+    async def upload(self, request: UploadRequest) -> UploadResponse:
+        return UploadResponse()
+
+    async def close(self) -> CloseResponse:
+        return CloseResponse()