Add an initial support for Conway governance commands

cardano_clusterlib/clusterlib_helpers.py

@@ -22,7 +22,7 @@ class EpochInfo(tp.NamedTuple):
 
 
 def _find_genesis_json(clusterlib_obj: "itp.ClusterLib") -> pl.Path:
-    """Find shelley genesis JSON file in state dir."""
+    """Find Shelley genesis JSON file in state dir."""
     default = clusterlib_obj.state_dir / "shelley" / "genesis.json"
     if default.exists():
         return default
@@ -41,6 +41,26 @@ def _find_genesis_json(clusterlib_obj: "itp.ClusterLib") -> pl.Path:
     return genesis_json
 
 
+def _find_conway_genesis_json(clusterlib_obj: "itp.ClusterLib") -> pl.Path:
+    """Find Conway genesis JSON file in state dir."""
+    default = clusterlib_obj.state_dir / "shelley" / "genesis.conway.json"
+    if default.exists():
+        return default
+
+    potential = [
+        *clusterlib_obj.state_dir.glob("*conway*genesis.json"),
+        *clusterlib_obj.state_dir.glob("*genesis*conway.json"),
+    ]
+    if not potential:
+        raise exceptions.CLIError(
+            f"Conway genesis JSON file not found in `{clusterlib_obj.state_dir}`."
+        )
+
+    genesis_json = potential[0]
+    LOGGER.debug(f"Using Conway genesis JSON file `{genesis_json}")
+    return genesis_json
+
+
 def _check_protocol(clusterlib_obj: "itp.ClusterLib") -> None:
     """Check that the cluster is running with the expected protocol."""
     try:

cardano_clusterlib/clusterlib_klass.py

@@ -9,6 +9,7 @@
 from cardano_clusterlib import address_group
 from cardano_clusterlib import clusterlib_helpers
 from cardano_clusterlib import consts
+from cardano_clusterlib import conway_gov_group
 from cardano_clusterlib import coverage
 from cardano_clusterlib import exceptions
 from cardano_clusterlib import genesis_group
@@ -53,6 +54,7 @@ def __init__(
         command_era: str = "",
         tx_era: str = "",  # deprecated - use `command_era` instead
     ):
+        # pylint: disable=too-many-statements
         self.cluster_id = 0  # can be used for identifying cluster instance
         self.cli_coverage: dict = {}
         self._rand_str = helpers.get_rand_str(4)
@@ -95,6 +97,19 @@ def __init__(
         # Ignore the `tx_era` if `command_era` is set
         self.tx_era = "" if self.command_era else tx_era.lower()
 
+        # Conway+ era
+        self.conway_genesis_json: tp.Optional[pl.Path] = None
+        self.conway_genesis: dict = {}
+        if consts.Eras[(self.command_era or "DEFAULT").upper()].value >= consts.Eras.CONWAY.value:
+            # Ignore the `tx_era`
+            self.tx_era = ""
+            # Conway genesis
+            self.conway_genesis_json = clusterlib_helpers._find_conway_genesis_json(
+                clusterlib_obj=self
+            )
+            with open(self.conway_genesis_json, encoding="utf-8") as in_json:
+                self.conway_genesis = json.load(in_json)
+
         self.overwrite_outfiles = True
 
         # Groups of commands
@@ -107,6 +122,7 @@ def __init__(
         self._key_group: tp.Optional[key_group.KeyGroup] = None
         self._genesis_group: tp.Optional[genesis_group.GenesisGroup] = None
         self._governance_group: tp.Optional[governance_group.GovernanceGroup] = None
+        self._conway_gov_group: tp.Optional[conway_gov_group.ConwayGovGroup] = None
 
         clusterlib_helpers._check_protocol(clusterlib_obj=self)
 
@@ -187,6 +203,20 @@ def g_governance(self) -> governance_group.GovernanceGroup:
             self._governance_group = governance_group.GovernanceGroup(clusterlib_obj=self)
         return self._governance_group
 
+    @property
+    def g_conway_governance(self) -> conway_gov_group.ConwayGovGroup:
+        """Conway governance group."""
+        if self._conway_gov_group:
+            return self._conway_gov_group
+
+        if not self.conway_genesis:
+            raise exceptions.CLIError(
+                "Conway governance group can be used only with Command era >= Conway."
+            )
+
+        self._conway_gov_group = conway_gov_group.ConwayGovGroup(clusterlib_obj=self)
+        return self._conway_gov_group
+
     def cli(
         self,
         cli_args: tp.List[str],

cardano_clusterlib/conway_gov_action_group.py

@@ -0,0 +1,271 @@
+"""Group of methods for Conway governance action commands."""
+import logging
+import pathlib as pl
+import typing as tp
+
+from cardano_clusterlib import clusterlib_helpers
+from cardano_clusterlib import consts
+from cardano_clusterlib import helpers
+from cardano_clusterlib import types as itp
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class ConwayGovActionGroup:
+    def __init__(self, clusterlib_obj: "itp.ClusterLib") -> None:
+        self._clusterlib_obj = clusterlib_obj
+        self._group_args = ("governance", "action")
+
+        if self._clusterlib_obj.network_magic == consts.MAINNET_MAGIC:
+            self.magic_args = ["--mainnet"]
+        else:
+            self.magic_args = ["--testnet"]
+
+    def _get_key_args(
+        self,
+        stake_vkey: str = "",
+        stake_vkey_file: tp.Optional[itp.FileType] = None,
+        stake_key_hash: str = "",
+    ) -> tp.List[str]:
+        """Get arguments for verification key."""
+        if stake_vkey:
+            key_args = ["--stake-verification-key", str(stake_vkey)]
+        elif stake_vkey_file:
+            key_args = ["--stake-verification-key-file", str(stake_vkey_file)]
+        elif stake_key_hash:
+            key_args = ["--stake-key-hash", str(stake_key_hash)]
+        else:
+            raise AssertionError("Either stake verification key or stake key hash must be set.")
+
+        return key_args
+
+    def _get_proposal_anchor_args(
+        self,
+        proposal_anchor_url: str,
+        proposal_anchor_metadata: str = "",
+        proposal_anchor_metadata_file: tp.Optional[itp.FileType] = None,
+        proposal_anchor_metadata_hash: str = "",
+    ) -> tp.List[str]:
+        """Get arguments for proposal anchor."""
+        proposal_anchor_args = ["--proposal-anchor-url", str(proposal_anchor_url)]
+        if proposal_anchor_metadata:
+            proposal_anchor_args.extend(
+                ["--proposal-anchor-metadata", str(proposal_anchor_metadata)]
+            )
+        elif proposal_anchor_metadata_file:
+            proposal_anchor_args.extend(
+                [
+                    "--proposal-anchor-metadata-file",
+                    str(proposal_anchor_metadata_file),
+                ]
+            )
+        elif proposal_anchor_metadata_hash:
+            proposal_anchor_args.extend(
+                [
+                    "--proposal-anchor-metadata-hash",
+                    str(proposal_anchor_metadata_hash),
+                ]
+            )
+        else:
+            raise AssertionError("Either proposal anchor metadata or its hash must be set.")
+
+        return proposal_anchor_args
+
+    def create_constitution(
+        self,
+        action_name: str,
+        deposit_amt: int,
+        proposal_anchor_url: str,
+        constitution_anchor_url: str,
+        stake_vkey: str = "",
+        stake_vkey_file: tp.Optional[itp.FileType] = None,
+        stake_key_hash: str = "",
+        prev_action_txid: str = "",
+        prev_action_ix: str = "",
+        proposal_anchor_metadata: str = "",
+        proposal_anchor_metadata_file: tp.Optional[itp.FileType] = None,
+        proposal_anchor_metadata_hash: str = "",
+        constitution_anchor_metadata: str = "",
+        constitution_anchor_metadata_file: tp.Optional[itp.FileType] = None,
+        constitution_anchor_metadata_hash: str = "",
+        destination_dir: itp.FileType = ".",
+    ) -> pl.Path:
+        """Create a constitution."""
+        # pylint: disable=too-many-arguments
+        destination_dir = pl.Path(destination_dir).expanduser()
+        out_file = destination_dir / f"{action_name}_constitution.action"
+        clusterlib_helpers._check_files_exist(out_file, clusterlib_obj=self._clusterlib_obj)
+
+        key_args = self._get_key_args(
+            stake_vkey=stake_vkey,
+            stake_vkey_file=stake_vkey_file,
+            stake_key_hash=stake_key_hash,
+        )
+
+        proposal_anchor_args = self._get_proposal_anchor_args(
+            proposal_anchor_url=proposal_anchor_url,
+            proposal_anchor_metadata=proposal_anchor_metadata,
+            proposal_anchor_metadata_file=proposal_anchor_metadata_file,
+            proposal_anchor_metadata_hash=proposal_anchor_metadata_hash,
+        )
+
+        prev_action_args = []
+        if prev_action_txid:
+            if not prev_action_ix:
+                raise AssertionError("Previous action index must be set.")
+            prev_action_args = [
+                "--governance-action-tx-id",
+                str(prev_action_txid),
+                "--governance-action-index",
+                str(prev_action_ix),
+            ]
+
+        constitution_anchor_args = ["--constitution-anchor-url", str(constitution_anchor_url)]
+        if constitution_anchor_metadata:
+            constitution_anchor_args.extend(
+                ["--constitution-anchor-metadata", str(constitution_anchor_metadata)]
+            )
+        elif constitution_anchor_metadata_file:
+            constitution_anchor_args.extend(
+                [
+                    "--constitution-anchor-metadata-file",
+                    str(constitution_anchor_metadata_file),
+                ]
+            )
+        elif constitution_anchor_metadata_hash:
+            constitution_anchor_args.extend(
+                [
+                    "--constitution-anchor-metadata-hash",
+                    str(constitution_anchor_metadata_hash),
+                ]
+            )
+        else:
+            raise AssertionError("Either constitution anchor metadata or its hash must be set.")
+
+        self._clusterlib_obj.cli(
+            [
+                *self._group_args,
+                "create-constitution",
+                *self.magic_args,
+                "--governance-action-deposit",
+                str(deposit_amt),
+                *key_args,
+                *prev_action_args,
+                *proposal_anchor_args,
+                *constitution_anchor_args,
+                "--out-file",
+                str(out_file),
+            ]
+        )
+
+        helpers._check_outfiles(out_file)
+        return out_file
+
+    def create_info(
+        self,
+        action_name: str,
+        deposit_amt: int,
+        proposal_anchor_url: str,
+        stake_vkey: str = "",
+        stake_vkey_file: tp.Optional[itp.FileType] = None,
+        stake_key_hash: str = "",
+        proposal_anchor_metadata: str = "",
+        proposal_anchor_metadata_file: tp.Optional[itp.FileType] = None,
+        proposal_anchor_metadata_hash: str = "",
+        destination_dir: itp.FileType = ".",
+    ) -> pl.Path:
+        """Create an info action."""
+        # pylint: disable=too-many-arguments
+        destination_dir = pl.Path(destination_dir).expanduser()
+        out_file = destination_dir / f"{action_name}_info.action"
+        clusterlib_helpers._check_files_exist(out_file, clusterlib_obj=self._clusterlib_obj)
+
+        key_args = self._get_key_args(
+            stake_vkey=stake_vkey,
+            stake_vkey_file=stake_vkey_file,
+            stake_key_hash=stake_key_hash,
+        )
+
+        proposal_anchor_args = self._get_proposal_anchor_args(
+            proposal_anchor_url=proposal_anchor_url,
+            proposal_anchor_metadata=proposal_anchor_metadata,
+            proposal_anchor_metadata_file=proposal_anchor_metadata_file,
+            proposal_anchor_metadata_hash=proposal_anchor_metadata_hash,
+        )
+
+        self._clusterlib_obj.cli(
+            [
+                *self._group_args,
+                "create-info",
+                *self.magic_args,
+                "--governance-action-deposit",
+                str(deposit_amt),
+                *key_args,
+                *proposal_anchor_args,
+                "--out-file",
+                str(out_file),
+            ]
+        )
+
+        helpers._check_outfiles(out_file)
+        return out_file
+
+    def create_no_confidence(
+        self,
+        action_name: str,
+        deposit_amt: int,
+        proposal_anchor_url: str,
+        prev_action_txid: str,
+        prev_action_ix: str,
+        stake_vkey: str = "",
+        stake_vkey_file: tp.Optional[itp.FileType] = None,
+        stake_key_hash: str = "",
+        proposal_anchor_metadata: str = "",
+        proposal_anchor_metadata_file: tp.Optional[itp.FileType] = None,
+        proposal_anchor_metadata_hash: str = "",
+        destination_dir: itp.FileType = ".",
+    ) -> pl.Path:
+        """Create a no confidence proposal."""
+        # pylint: disable=too-many-arguments
+        destination_dir = pl.Path(destination_dir).expanduser()
+        out_file = destination_dir / f"{action_name}_confidence.action"
+        clusterlib_helpers._check_files_exist(out_file, clusterlib_obj=self._clusterlib_obj)
+
+        key_args = self._get_key_args(
+            stake_vkey=stake_vkey,
+            stake_vkey_file=stake_vkey_file,
+            stake_key_hash=stake_key_hash,
+        )
+
+        proposal_anchor_args = self._get_proposal_anchor_args(
+            proposal_anchor_url=proposal_anchor_url,
+            proposal_anchor_metadata=proposal_anchor_metadata,
+            proposal_anchor_metadata_file=proposal_anchor_metadata_file,
+            proposal_anchor_metadata_hash=proposal_anchor_metadata_hash,
+        )
+
+        prev_action_args = [
+            "--governance-action-tx-id",
+            str(prev_action_txid),
+            "--governance-action-index",
+            str(prev_action_ix),
+        ]
+
+        self._clusterlib_obj.cli(
+            [
+                *self._group_args,
+                "create-no-confidence",
+                *self.magic_args,
+                "--governance-action-deposit",
+                str(deposit_amt),
+                *key_args,
+                *prev_action_args,
+                *proposal_anchor_args,
+                "--out-file",
+                str(out_file),
+            ]
+        )
+
+        helpers._check_outfiles(out_file)
+        return out_file