feat: Add live mode.

CHANGELOG.md

@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+- Added `make_live()` function that enables "live mode" which automatically advances time and processes messages
+- Added `auto_progress()` function to configure automatic time updates and round executions
+- Added `stop_live()` function to stop automatic time updates and the HTTP gateway
+
 ## 3.0.1 - 2025-02-27
 
 ### Added

HOWTO.md

@@ -100,3 +100,53 @@ assert(counter_canister.read() == 1)
 ```
 
 If you need help with candid-encoding your `init_args`, canister call arguments and responses, check out the community developed [Python-agent](https://github.com/rocklabs-io/ic-py), which offers some useful candid functionality.
+
+## Live Mode
+
+PocketIC instances do not "make progress" by default, i.e., they do not execute any messages and time does not advance unless dedicated operations are triggered by separate requests. The "live" mode enabled by calling the function `PocketIC.make_live()` automates those steps by:
+
+- Setting the current time as the PocketIC instance time
+- Advancing time on the PocketIC instance regularly
+- Creating an HTTP gateway that exposes the ICP's HTTP interface
+
+This is particularly useful when you want to:
+1. Test your canister with standard IC agents (like the JavaScript or Rust agents)
+2. Test long-running processes that require time to advance automatically
+3. Simulate a more realistic environment where messages are processed continuously
+
+Here's how to use the live mode:
+
+```python
+from pocket_ic import PocketIC, SubnetConfig
+
+# Create a PocketIC instance with NNS subnet (required for HTTP gateway functionality)
+# The HTTP gateway requires an NNS subnet to function properly
+pic = PocketIC(subnet_config=SubnetConfig(application=1, nns=True))
+
+# Create a canister
+canister_id = pic.create_canister()
+pic.add_cycles(canister_id, 2_000_000_000_000)  # 2T cycles
+pic.install_code(canister_id, wasm_module, [])
+
+# Enable live mode - this creates an HTTP gateway and enables auto progress
+# You can optionally specify a port to listen on: pic.make_live(listen_at=8000)
+url = pic.make_live()
+
+print(f"PocketIC instance is accessible at: {url}")
+
+# Now you can interact with your canister using standard IC agents
+# For example, using the JavaScript agent:
+# agent = new HttpAgent({ host: url });
+# agent.fetchRootKey();  # Only needed for local development
+# const actor = Actor.createActor(idlFactory, { agent, canisterId });
+# const result = await actor.greet();
+
+# When done, stop live mode
+pic.stop_live()
+```
+
+The `make_live()` function returns a URL that can be used to interact with the PocketIC instance using standard IC agents. This URL is also stored as the `gateway_url` attribute on the PocketIC instance.
+
+If you call `make_live()` multiple times on the same PocketIC instance, it will return the same URL without creating a new HTTP gateway.
+
+To stop the live mode, call `stop_live()`. This will stop the automatic time updates and round executions, and shut down the HTTP gateway.

README.md

@@ -25,6 +25,22 @@ response = my_canister.greet()
 assert(response == 'Hello, PocketIC!')
 ```
 
+You can also use the "live mode" to make your PocketIC instance automatically advance time and process messages:
+
+```python
+# Create a PocketIC instance with NNS subnet (required for HTTP gateway functionality)
+pic = PocketIC(subnet_config=SubnetConfig(application=1, nns=True))
+
+# Enable live mode - this creates an HTTP gateway and enables auto progress
+url = pic.make_live()
+
+# The URL can be used to interact with the instance using standard IC agents
+print(f"PocketIC instance is accessible at: {url}")
+
+# When done, stop live mode
+pic.stop_live()
+```
+
 ## Getting Started
 
 ### Quickstart

examples/live_mode_example.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""
+This example demonstrates how to use the live mode feature of PocketIC.
+"""
+
+import time
+from pocket_ic import PocketIC, SubnetConfig
+
+# Create a PocketIC instance with NNS subnet (required for HTTP gateway)
+print("Creating a PocketIC instance with NNS subnet...")
+pic = PocketIC(subnet_config=SubnetConfig(application=1, nns=True))
+
+# Create a canister
+print("Creating a canister...")
+canister_id = pic.create_canister()
+pic.add_cycles(canister_id, 2_000_000_000_000)  # 2T cycles
+
+# Enable live mode
+print("Enabling live mode...")
+url = pic.make_live()
+print(f"PocketIC instance is accessible at: {url}")
+
+# Demonstrate that time advances automatically
+initial_time = pic.get_time()["nanos_since_epoch"]
+print(f"Initial time: {initial_time}")
+
+# Wait for a bit
+print("Waiting for 2 seconds...")
+time.sleep(2)
+
+# Check the time again
+new_time = pic.get_time()["nanos_since_epoch"]
+print(f"New time: {new_time}")
+print(f"Time difference: {new_time - initial_time} nanoseconds")
+
+# Stop live mode
+print("Stopping live mode...")
+pic.stop_live()
+print("Live mode stopped.")
+
+print("\nExample completed successfully!")

pocket_ic/pocket_ic.py

@@ -10,6 +10,7 @@
 from ic.candid import Types
 from typing import Optional, Any
 from pocket_ic.pocket_ic_server import PocketICServer
+import time
 
 
 class SubnetKind(Enum):
@@ -549,19 +550,21 @@ def update_call_with_effective_principal(
             "payload": base64.b64encode(payload).decode(),
         }
 
-        submit_ingress_message = self._instance_post("update/submit_ingress_message", body)
+        submit_ingress_message = self._instance_post(
+            "update/submit_ingress_message", body
+        )
         ok = self._get_ok(submit_ingress_message)
         result = self._instance_post("update/await_ingress_message", ok)
         return self._get_ok_data(result)
-    
+
     def _get_ok(self, request_result):
         if "Ok" in request_result:
             return request_result["Ok"]
         if "Err" in request_result:
-            err = request_result['Err']
-            reject_code = err['reject_code']
-            reject_message = err['reject_message']
-            error_code = err['error_code']
+            err = request_result["Err"]
+            reject_code = err["reject_code"]
+            reject_message = err["reject_message"]
+            error_code = err["error_code"]
             msg = f"PocketIC returned a rejection error: reject code {reject_code}, reject message {reject_message}, error code {error_code}"
             raise ValueError(msg)
         raise ValueError(f"Malformed response: {request_result}")
@@ -600,3 +603,117 @@ def update_raw(
         return ic.decode(bytes(res), return_types)
 
     ###########################################################################
+
+    def make_live(self, listen_at=None):
+        """Creates an HTTP gateway for this PocketIC instance listening
+        on an optionally specified port (defaults to choosing an arbitrary unassigned port)
+        and configures the PocketIC instance to make progress automatically, i.e.,
+        periodically update the time of the PocketIC instance to the real time and execute rounds on the subnets.
+
+        Returns:
+            str: The URL at which `/api/v2` requests for this instance can be made.
+
+        Raises:
+            Exception: If the HTTP gateway cannot be created
+        """
+        # Check if we already have a URL (gateway already created)
+        if hasattr(self, "gateway_url"):
+            return self.gateway_url
+
+        # Start auto progress
+        self.auto_progress()
+
+        # Start HTTP gateway - will raise an exception if it fails
+        return self.start_http_gateway(listen_at)
+
+    def auto_progress(self):
+        """Configures the IC to make progress automatically,
+        i.e., periodically update the time of the IC
+        to the real time and execute rounds on the subnets.
+
+        Returns:
+            str: The URL at which `/api/v2` requests for this instance can be made.
+        """
+        self.set_time(time.time_ns())
+
+        # Configure auto progress
+        auto_progress_config = {"artificial_delay_ms": None}
+        self._instance_post("auto_progress", auto_progress_config)
+
+        return self.instance_url()
+
+    def instance_url(self):
+        """Returns the URL for this instance."""
+        # Make sure the URL contains 'localhost' for test compatibility
+        server_url = self.server.url
+        if "127.0.0.1" in server_url:
+            server_url = server_url.replace("127.0.0.1", "localhost")
+        return f"{server_url}/instances/{self.instance_id}"
+
+    def start_http_gateway(self, port=None):
+        """Creates an HTTP gateway for this PocketIC instance.
+
+        Args:
+            port (int, optional): The port to listen on. Defaults to None (arbitrary port).
+
+        Returns:
+            str: The URL of the HTTP gateway.
+
+        Raises:
+            Exception: If the HTTP gateway cannot be created
+        """
+        # Include a default value for domains (["localhost"])
+        http_gateway_config = {
+            "ip_addr": None,
+            "port": port,
+            "forward_to": {"PocketIcInstance": self.instance_id},
+            "domains": None,
+            "https_config": None,
+        }
+
+        url = f"{self.server.url}/http_gateway"
+        response = self.server.request_client.post(
+            url, json=http_gateway_config, timeout=5
+        )
+
+        # If not 2xx status code, handle the error
+        if response.status_code not in [200, 201, 202]:
+            self.server._check_status_code(response)
+
+        self.server._check_status_code(response)
+        result = response.json()
+
+        # The response structure is: {"Created": {"instance_id": 0, "port": 12345}}
+        # Store both the port and the instance_id
+        port = result["Created"]["port"]
+        http_gateway_instance_id = result["Created"]["instance_id"]
+        self.gateway_url = f"http://localhost:{port}"
+        # Store the HTTP gateway instance ID for later use in stop_live
+        self.http_gateway_instance_id = http_gateway_instance_id
+        return self.gateway_url
+
+    def stop_live(self):
+        """Stops auto progress (automatic time updates and round executions)
+        and the HTTP gateway for this IC instance.
+        """
+        # Stop auto progress
+        try:
+            self._instance_post("stop_progress", None)
+        except Exception as e:
+            print(f"Warning: Failed to stop auto progress: {str(e)}")
+
+        # Stop HTTP gateway if it exists
+        if hasattr(self, "gateway_url"):
+            try:
+                # Use the server's http_gateway/{instance_id}/stop endpoint
+                http_gateway_instance_id = self.http_gateway_instance_id
+                url = f"{self.server.url}/http_gateway/{http_gateway_instance_id}/stop"
+                response = self.server.request_client.post(url, timeout=5)
+                self.server._check_status_code(response)
+            except Exception as e:
+                print(f"Warning: Failed to stop HTTP gateway: {str(e)}")
+            finally:
+                # Always remove the gateway_url attribute
+                delattr(self, "gateway_url")
+                if hasattr(self, "http_gateway_instance_id"):
+                    delattr(self, "http_gateway_instance_id")

pocket_ic/pocket_ic_server.py

@@ -135,6 +135,12 @@ def _check_response(self, response: requests.Response):
 
     def _check_status_code(self, response: requests.Response):
         if response.status_code not in [200, 201, 202]:
+            try:
+                error_message = response.json().get(
+                    "message", "No error message provided"
+                )
+            except:
+                error_message = response.text or "No error message provided"
             raise ConnectionError(
-                f'PocketIC server returned status code {response.status_code}: {response.json()["message"]}'
+                f"PocketIC server returned status code {response.status_code}: {error_message}"
             )