Cardano cli chain context

integration-test/docker-compose.yml

@@ -13,6 +13,7 @@ services:
     entrypoint: bash
     environment:
       NETWORK: "${NETWORK:-local-alonzo}"
+      CARDANO_NODE_SOCKET_PATH: "/ipc/node.socket"
     command: /code/run_node.sh
 
     networks:
@@ -21,6 +22,7 @@ services:
 
     volumes:
       - .:/code
+      - /tmp:/tmp
       - node-db:/data/db
       - node-ipc:/ipc
     ports:

integration-test/test/test_cardano_cli.py

@@ -0,0 +1,67 @@
+import os
+from pathlib import Path
+
+from pycardano import (
+    CardanoCliChainContext,
+    CardanoCliNetwork,
+    GenesisParameters,
+    Network,
+    ProtocolParameters,
+)
+from pycardano.backend.cardano_cli import DockerConfig
+
+
+class TestCardanoCli:
+    network_env = os.getenv("NETWORK", "local-alonzo")
+    host_socket = os.getenv("DOCKER_HOST", None)
+    network_magic = os.getenv("NETWORK_MAGIC", 42)
+
+    configs_dir = Path(__file__).parent.parent / "configs"
+
+    chain_context = CardanoCliChainContext(
+        binary=Path("cardano-cli"),
+        socket=Path("/ipc/node.socket"),
+        config_file=configs_dir / network_env / "config.json",
+        network=CardanoCliNetwork.CUSTOM,
+        docker_config=DockerConfig(
+            container_name="integration-test_cardano-node_1",
+            host_socket=Path(host_socket) if host_socket else None,
+        ),
+        network_magic_number=int(network_magic),
+    )
+
+    def test_protocol_param(self):
+        protocol_param = self.chain_context.protocol_param
+
+        assert protocol_param is not None
+        assert isinstance(protocol_param, ProtocolParameters)
+
+        cost_models = protocol_param.cost_models
+        for _, cost_model in cost_models.items():
+            assert "addInteger-cpu-arguments-intercept" in cost_model
+            assert "addInteger-cpu-arguments-slope" in cost_model
+
+    def test_genesis_param(self):
+        genesis_param = self.chain_context.genesis_param
+
+        assert genesis_param is not None
+        assert isinstance(genesis_param, GenesisParameters)
+
+    def test_network(self):
+        network = self.chain_context.network
+
+        assert network is not None
+        assert isinstance(network, Network)
+
+    def test_epoch(self):
+        epoch = self.chain_context.epoch
+
+        assert epoch is not None
+        assert isinstance(epoch, int)
+        assert epoch > 0
+
+    def test_last_block_slot(self):
+        last_block_slot = self.chain_context.last_block_slot
+
+        assert isinstance(last_block_slot, int)
+        assert last_block_slot > 0

integration-test/test/test_plutus.py

@@ -1,4 +1,6 @@
+import collections
 import time
+from typing import Dict, Union
 
 import cbor2
 import pytest
@@ -7,6 +9,7 @@
 from pycardano import *
 
 from .base import TEST_RETRIES, TestBase
+from .test_cardano_cli import TestCardanoCli
 
 
 class TestPlutus(TestBase):
@@ -371,3 +374,22 @@ class TestPlutusOgmiosOnly(TestPlutus):
     @classmethod
     def setup_class(cls):
         cls.chain_context._kupo_url = None
+
+
+def evaluate_tx(tx: Transaction) -> Dict[str, ExecutionUnits]:
+    redeemers = tx.transaction_witness_set.redeemer
+    execution_units = {}
+
+    if redeemers:
+        for r in redeemers:
+            k = f"{r.tag.name.lower()}:{r.index}"
+            execution_units[k] = ExecutionUnits(1000000, 1000000000)
+
+    return execution_units
+
+
+class TestPlutusCardanoCLI(TestPlutus):
+    @classmethod
+    def setup_class(cls):
+        cls.chain_context = TestCardanoCli.chain_context
+        cls.chain_context.evaluate_tx = evaluate_tx

pycardano/backend/__init__.py

@@ -2,4 +2,5 @@
 
 from .base import *
 from .blockfrost import *
+from .cardano_cli import *
 from .ogmios import *

pycardano/backend/base.py

@@ -171,7 +171,7 @@ def submit_tx(self, tx: Union[Transaction, bytes, str]):
         """
         if isinstance(tx, Transaction):
             return self.submit_tx_cbor(tx.to_cbor())
-        elif isinstance(tx, bytes):
+        elif isinstance(tx, bytes) or isinstance(tx, str):
             return self.submit_tx_cbor(tx)
         else:
             raise InvalidArgumentException(