add separate tests for container and external arkiv nodes

Author: Matthias Zimmermann

src/arkiv/node.py

@@ -87,7 +87,6 @@ def __init__(
         image: str | None = None,
         http_port: int | None = None,
         ws_port: int | None = None,
-        auto_start: bool = False,
         http_url: str | None = None,
         ws_url: str | None = None,
     ) -> None:
@@ -98,7 +97,6 @@ def __init__(
             image: Docker image to use (default: golemnetwork/golembase-op-geth:latest)
             http_port: Internal HTTP port (default: 8545)
             ws_port: Internal WebSocket port (default: 8546)
-            auto_start: Automatically start the node on initialization (default: False)
             http_url: External HTTP RPC URL (for external nodes, disables container)
             ws_url: External WebSocket RPC URL (for external nodes, disables container)
 
@@ -142,9 +140,6 @@ def __init__(
         self._http_port = http_port or self.DEFAULT_HTTP_PORT
         self._ws_port = ws_port or self.DEFAULT_WS_PORT
 
-        if auto_start:
-            self.start()
-
     @property
     def http_url(self) -> str:
         """

tests/conftest.py

@@ -50,6 +50,37 @@ def _load_env_if_available() -> None:
 _load_env_if_available()
 
 
+@pytest.fixture(scope="session")
+def arkiv_container() -> Generator[ArkivNode, None, None]:
+    """
+    Provide a containerized ArkivNode for testing.
+
+    Creates and manages a local Docker-based Arkiv node.
+    Use this fixture for tests that need container-specific features.
+    """
+    logger.info("Creating containerized test ArkivNode...")
+    with ArkivNode() as node:
+        logger.info(f"Test node ready at {node.http_url}")
+        yield node
+
+
+@pytest.fixture(scope="session")
+def arkiv_testnet() -> ArkivNode:
+    """
+    Provide Kaolin testnet ArkivNode for testing.
+
+    Always points to the Kaolin testnet.
+    Use this fixture for tests that work with the public Kaolin testnet.
+    """
+    from arkiv.provider import HTTP, KAOLIN, NETWORK_URL, WS
+
+    http_url = NETWORK_URL[KAOLIN][HTTP]
+    ws_url = NETWORK_URL[KAOLIN][WS]
+
+    logger.info(f"Using Kaolin testnet: {http_url}")
+    return ArkivNode(http_url=http_url, ws_url=ws_url)
+
+
 @pytest.fixture(scope="session")
 def arkiv_node() -> Generator[ArkivNode, None, None]:
     """
@@ -58,6 +89,7 @@ def arkiv_node() -> Generator[ArkivNode, None, None]:
     If RPC_URL and WS_URL are set in environment, creates an ArkivNode
     configured for the external node (no container management).
     Otherwise, creates and manages a local containerized node.
+    Use this fixture for tests that work with either node type.
     """
     # Check for external node configuration
     rpc_url = os.getenv(RPC_URL_ENV)

tests/test_node.py

@@ -12,78 +12,241 @@
 logger = logging.getLogger(__name__)
 
 
-def test_node_connection_http(arkiv_node: ArkivNode) -> None:
-    """Check if the Arkiv node is available and responsive via JSON-RPC."""
-    # Use JSON-RPC call - works for both dev and production nodes
-    rpc_payload = {"jsonrpc": "2.0", "method": "eth_chainId", "params": [], "id": 1}
-
-    response = requests.post(
-        arkiv_node.http_url,
-        json=rpc_payload,
-        headers={"Content-Type": "application/json"},
-        timeout=10,
-    )
-
-    assert response.status_code == 200, (
-        f"Arkiv node should respond with 200 OK, got {response.status_code}"
-    )
-
-    # Verify it's a proper JSON-RPC response
-    json_response = response.json()
-    assert "result" in json_response or "error" in json_response, (
-        "Response should contain either 'result' or 'error' field"
-    )
-    assert json_response.get("jsonrpc") == "2.0", (
-        "Response should have jsonrpc version 2.0"
-    )
-
-    logger.info(f"HTTP connection successful: {arkiv_node.http_url}")
-    logger.info(f"Request response: {json_response}")
-
-
-def test_node_connection_ws(arkiv_node: ArkivNode) -> None:
-    """Check if the Arkiv node WebSocket endpoint is available and responsive."""
-    # Try to import websockets, skip test if not available
-    try:
-        import websockets
-    except ImportError:
-        pytest.skip("websockets package not available")
-
-    async def test_ws_connection() -> dict[str, object]:
-        """Test WebSocket connection and send a JSON-RPC request."""
-        async with websockets.connect(arkiv_node.ws_url, open_timeout=5) as websocket:
-            # Send a JSON-RPC request over WebSocket
-            request = {"jsonrpc": "2.0", "method": "eth_chainId", "params": [], "id": 1}
-
-            await websocket.send(json.dumps(request))
-            response = await websocket.recv()
-
-            return json.loads(response)  # type: ignore[no-any-return]
-
-    # Run the async test
-    response = asyncio.run(test_ws_connection())
-    logger.info(f"WebSocket response: {response}")
-
-    # Verify the WebSocket JSON-RPC response
-    assert "result" in response or "error" in response, (
-        "WebSocket response should contain either 'result' or 'error' field"
-    )
-    assert response.get("jsonrpc") == "2.0", (
-        "WebSocket response should have jsonrpc version 2.0"
-    )
-    assert response.get("id") == 1, "WebSocket response should have matching request id"
-
-    logger.info(f"WebSocket connection successful: {arkiv_node.ws_url}")
-    logger.info(f"Chain ID response: {response.get('result', 'N/A')}")
-
-
-def test_node_arkiv_help(arkiv_node: ArkivNode) -> None:
-    """Check if the Arkiv node help command is available and responsive."""
-    if arkiv_node.is_external():
-        pytest.skip("Skipping help command test for external nodes")
-
-    help_command = ["golembase", "account", "help"]
-    exit_code, output = arkiv_node.container.exec(help_command)
-    logger.info(
-        f"Account help command: {help_command}, exit_code: {exit_code}, output:\n{output.decode()}"
-    )
+class TestArkivNodeConnections:
+    """Test basic node connectivity - works for both containerized and external nodes."""
+
+    def test_node_connection_http(self, arkiv_node: ArkivNode) -> None:
+        """Check if the Arkiv node is available and responsive via JSON-RPC."""
+        # Use JSON-RPC call - works for both dev and production nodes
+        rpc_payload = {"jsonrpc": "2.0", "method": "eth_chainId", "params": [], "id": 1}
+
+        response = requests.post(
+            arkiv_node.http_url,
+            json=rpc_payload,
+            headers={"Content-Type": "application/json"},
+            timeout=10,
+        )
+
+        assert response.status_code == 200, (
+            f"Arkiv node should respond with 200 OK, got {response.status_code}"
+        )
+
+        # Verify it's a proper JSON-RPC response
+        json_response = response.json()
+        assert "result" in json_response or "error" in json_response, (
+            "Response should contain either 'result' or 'error' field"
+        )
+        assert json_response.get("jsonrpc") == "2.0", (
+            "Response should have jsonrpc version 2.0"
+        )
+
+        logger.info(f"HTTP connection successful: {arkiv_node.http_url}")
+        logger.info(f"Request response: {json_response}")
+
+    def test_node_connection_ws(self, arkiv_node: ArkivNode) -> None:
+        """Check if the Arkiv node WebSocket endpoint is available and responsive."""
+        # Try to import websockets, skip test if not available
+        try:
+            import websockets
+        except ImportError:
+            pytest.skip("websockets package not available")
+
+        async def test_ws_connection() -> dict[str, object]:
+            """Test WebSocket connection and send a JSON-RPC request."""
+            async with websockets.connect(
+                arkiv_node.ws_url, open_timeout=5
+            ) as websocket:
+                # Send a JSON-RPC request over WebSocket
+                request = {
+                    "jsonrpc": "2.0",
+                    "method": "eth_chainId",
+                    "params": [],
+                    "id": 1,
+                }
+
+                await websocket.send(json.dumps(request))
+                response = await websocket.recv()
+
+                return json.loads(response)  # type: ignore[no-any-return]
+
+        # Run the async test
+        response = asyncio.run(test_ws_connection())
+        logger.info(f"WebSocket response: {response}")
+
+        # Verify the WebSocket JSON-RPC response
+        assert "result" in response or "error" in response, (
+            "WebSocket response should contain either 'result' or 'error' field"
+        )
+        assert response.get("jsonrpc") == "2.0", (
+            "WebSocket response should have jsonrpc version 2.0"
+        )
+        assert response.get("id") == 1, (
+            "WebSocket response should have matching request id"
+        )
+
+        logger.info(f"WebSocket connection successful: {arkiv_node.ws_url}")
+        logger.info(f"Chain ID response: {response.get('result', 'N/A')}")
+
+
+class TestContainerizedNode:
+    """Tests specific to containerized (local Docker) nodes."""
+
+    def test_node_container_properties(self, arkiv_container: ArkivNode) -> None:
+        """Test that containerized nodes expose container properties."""
+        # Should have access to container
+        assert arkiv_container.container is not None
+        assert arkiv_container.http_port > 0
+        assert arkiv_container.ws_port > 0
+
+    def test_node_container_stop_start(self, arkiv_container: ArkivNode) -> None:
+        """Test that containerized nodes can be stopped and started."""
+        # Should be running initially
+        assert arkiv_container.is_running()
+
+        # Stop the node
+        arkiv_container.stop()
+        assert not arkiv_container.is_running()
+        with pytest.raises(
+            RuntimeError,
+            match=r"Node is not running\. Call start\(\) first or use context manager\.",
+        ):
+            _ = arkiv_container.http_url
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"Node is not running\. Call start\(\) first or use context manager\.",
+        ):
+            _ = arkiv_container.ws_url
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"Node is not running\. Call start\(\) first or use context manager\.",
+        ):
+            _ = arkiv_container.http_port
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"Node is not running\. Call start\(\) first or use context manager\.",
+        ):
+            _ = arkiv_container.ws_port
+
+        # Stop a 2nd time should be no-op
+        arkiv_container.stop()
+        assert not arkiv_container.is_running()
+
+        # Start the node again
+        arkiv_container.start()
+        assert arkiv_container.is_running()
+        assert arkiv_container.http_port > 0
+        assert arkiv_container.ws_port > 0
+
+        # Start a 2nd time should be no-op
+        arkiv_container.start()
+        assert arkiv_container.is_running()
+        assert arkiv_container.http_port > 0
+        assert arkiv_container.ws_port > 0
+
+    def test_node_container_arkiv_help_command(
+        self, arkiv_container: ArkivNode
+    ) -> None:
+        """Check if the Arkiv node help command is available via container CLI."""
+        help_command = ["golembase", "account", "help"]
+        exit_code, output = arkiv_container.container.exec(help_command)
+
+        assert exit_code == 0, f"Help command should succeed, got exit code {exit_code}"
+        assert b"help" in output.lower() or b"usage" in output.lower(), (
+            "Help output should contain help or usage information"
+        )
+
+        logger.info(
+            f"Account help command: {help_command}, exit_code: {exit_code}, output:\n{output.decode()}"
+        )
+
+    def test_node_container_fund_account(self, arkiv_container: ArkivNode) -> None:
+        """Test that fund_account works for containerized nodes."""
+        from arkiv.account import NamedAccount
+
+        # Create a test account
+        account = NamedAccount.create("test_fund")
+
+        # Should be able to fund it
+        arkiv_container.fund_account(account)
+
+        # No exception means success
+        logger.info(f"Successfully funded account {account.name}")
+
+    def test_node_container_context_manager(self, arkiv_container: ArkivNode) -> None:
+        """Test that containerized nodes work with context managers."""
+        # arkiv_container fixture already uses context manager, just verify it's running
+        assert arkiv_container.is_running()
+        assert not arkiv_container.is_external()
+
+
+class TestExternalNode:
+    """Tests specific to external (remote) nodes - uses Kaolin testnet."""
+
+    def test_node_external_urls(self, arkiv_testnet: ArkivNode) -> None:
+        """Test that external nodes have proper URLs configured."""
+
+        # Should have URLs
+        assert arkiv_testnet.http_url.startswith("http")
+        assert arkiv_testnet.ws_url.startswith("ws")
+
+        logger.info(f"External node HTTP URL: {arkiv_testnet.http_url}")
+        logger.info(f"External node WS URL: {arkiv_testnet.ws_url}")
+
+    def test_node_external_no_container(self, arkiv_testnet: ArkivNode) -> None:
+        """Test that external nodes raise errors when accessing container properties."""
+
+        # Should raise error when accessing container
+        with pytest.raises(RuntimeError, match="External nodes do not have containers"):
+            _ = arkiv_testnet.container
+
+        # Should raise error when accessing ports
+        with pytest.raises(
+            RuntimeError, match="External nodes do not expose port information"
+        ):
+            _ = arkiv_testnet.http_port
+
+        with pytest.raises(
+            RuntimeError, match="External nodes do not expose port information"
+        ):
+            _ = arkiv_testnet.ws_port
+
+    def test_node_external_cannot_start(self, arkiv_testnet: ArkivNode) -> None:
+        """Test that external nodes cannot be started."""
+
+        with pytest.raises(RuntimeError, match="Cannot start external node"):
+            arkiv_testnet.start()
+
+    def test_node_external_cannot_stop(self, arkiv_testnet: ArkivNode) -> None:
+        """Test that external nodes cannot be stopped."""
+
+        with pytest.raises(RuntimeError, match="Cannot stop external node"):
+            arkiv_testnet.stop()
+
+    def test_node_external_cannot_fund(self, arkiv_testnet: ArkivNode) -> None:
+        """Test that external nodes cannot fund accounts."""
+
+        from arkiv.account import NamedAccount
+
+        account = NamedAccount.create("test_external")
+
+        with pytest.raises(RuntimeError, match="Cannot fund account on external node"):
+            arkiv_testnet.fund_account(account)
+
+    def test_node_external_context_manager(self, arkiv_testnet: ArkivNode) -> None:
+        """Test that external nodes work safely with context managers."""
+
+        # Should work without errors (no-op)
+        with arkiv_testnet as node:
+            assert node.is_running()
+            assert node.http_url == arkiv_testnet.http_url
+
+        # Node should still be running after context exit
+        assert arkiv_testnet.is_running()
+
+    def test_node_external_is_external(self, arkiv_testnet: ArkivNode) -> None:
+        """Verify that external nodes are properly identified."""
+
+        assert arkiv_testnet.is_external()