REFACTOR. pull out genesis and protocol param fetching logic to be independently testable

REFACTOR. renaming chain tip status checking helper method

Author: daehan-koreapool

pycardano/backend/ogmios.py

@@ -112,7 +112,7 @@ def _query_utxos_by_tx_id(self, tx_id: str, index: int) -> List[List[JSON]]:
         args = {"query": {"utxo": [{"txId": tx_id, "index": index}]}}
         return self._request(OgmiosQueryType.Query, args)
 
-    def _check_chain_tip_and_update(self):
+    def _is_chain_tip_updated(self):
         slot = self.last_block_slot
         if self._last_known_block_slot != slot:
             self._last_known_block_slot = slot
@@ -128,81 +128,84 @@ def _fraction_parser(fraction: str) -> float:
     @property
     def protocol_param(self) -> ProtocolParameters:
         """Get current protocol parameters"""
-        if not self._protocol_param or self._check_chain_tip_and_update():
-            result = self._query_current_protocol_params()
-            param = ProtocolParameters(
-                min_fee_constant=result["minFeeConstant"],
-                min_fee_coefficient=result["minFeeCoefficient"],
-                max_block_size=result["maxBlockBodySize"],
-                max_tx_size=result["maxTxSize"],
-                max_block_header_size=result["maxBlockHeaderSize"],
-                key_deposit=result["stakeKeyDeposit"],
-                pool_deposit=result["poolDeposit"],
-                pool_influence=self._fraction_parser(result["poolInfluence"]),
-                monetary_expansion=self._fraction_parser(result["monetaryExpansion"]),
-                treasury_expansion=self._fraction_parser(result["treasuryExpansion"]),
-                decentralization_param=self._fraction_parser(
-                    result.get("decentralizationParameter", "0/1")
-                ),
-                extra_entropy=result.get("extraEntropy", ""),
-                protocol_major_version=result["protocolVersion"]["major"],
-                protocol_minor_version=result["protocolVersion"]["minor"],
-                min_pool_cost=result["minPoolCost"],
-                price_mem=self._fraction_parser(result["prices"]["memory"]),
-                price_step=self._fraction_parser(result["prices"]["steps"]),
-                max_tx_ex_mem=result["maxExecutionUnitsPerTransaction"]["memory"],
-                max_tx_ex_steps=result["maxExecutionUnitsPerTransaction"]["steps"],
-                max_block_ex_mem=result["maxExecutionUnitsPerBlock"]["memory"],
-                max_block_ex_steps=result["maxExecutionUnitsPerBlock"]["steps"],
-                max_val_size=result["maxValueSize"],
-                collateral_percent=result["collateralPercentage"],
-                max_collateral_inputs=result["maxCollateralInputs"],
-                coins_per_utxo_word=result.get(
-                    "coinsPerUtxoWord", ALONZO_COINS_PER_UTXO_WORD
-                ),
-                coins_per_utxo_byte=result.get("coinsPerUtxoByte", 0),
-                cost_models=result.get("costModels", {}),
-            )
+        if not self._protocol_param or self._is_chain_tip_updated():
+            self._protocol_param = self._fetch_protocol_param()
+        return self._protocol_param
 
-            if "plutus:v1" in param.cost_models:
-                param.cost_models["PlutusV1"] = param.cost_models.pop("plutus:v1")
-            if "plutus:v2" in param.cost_models:
-                param.cost_models["PlutusV2"] = param.cost_models.pop("plutus:v2")
+    def _fetch_protocol_param(self) -> ProtocolParameters:
+        result = self._query_current_protocol_params()
+        param = ProtocolParameters(
+            min_fee_constant=result["minFeeConstant"],
+            min_fee_coefficient=result["minFeeCoefficient"],
+            max_block_size=result["maxBlockBodySize"],
+            max_tx_size=result["maxTxSize"],
+            max_block_header_size=result["maxBlockHeaderSize"],
+            key_deposit=result["stakeKeyDeposit"],
+            pool_deposit=result["poolDeposit"],
+            pool_influence=self._fraction_parser(result["poolInfluence"]),
+            monetary_expansion=self._fraction_parser(result["monetaryExpansion"]),
+            treasury_expansion=self._fraction_parser(result["treasuryExpansion"]),
+            decentralization_param=self._fraction_parser(
+                result.get("decentralizationParameter", "0/1")
+            ),
+            extra_entropy=result.get("extraEntropy", ""),
+            protocol_major_version=result["protocolVersion"]["major"],
+            protocol_minor_version=result["protocolVersion"]["minor"],
+            min_pool_cost=result["minPoolCost"],
+            price_mem=self._fraction_parser(result["prices"]["memory"]),
+            price_step=self._fraction_parser(result["prices"]["steps"]),
+            max_tx_ex_mem=result["maxExecutionUnitsPerTransaction"]["memory"],
+            max_tx_ex_steps=result["maxExecutionUnitsPerTransaction"]["steps"],
+            max_block_ex_mem=result["maxExecutionUnitsPerBlock"]["memory"],
+            max_block_ex_steps=result["maxExecutionUnitsPerBlock"]["steps"],
+            max_val_size=result["maxValueSize"],
+            collateral_percent=result["collateralPercentage"],
+            max_collateral_inputs=result["maxCollateralInputs"],
+            coins_per_utxo_word=result.get(
+                "coinsPerUtxoWord", ALONZO_COINS_PER_UTXO_WORD
+            ),
+            coins_per_utxo_byte=result.get("coinsPerUtxoByte", 0),
+            cost_models=result.get("costModels", {}),
+        )
 
-            result = self._query_genesis_config()
-            param.min_utxo = result["protocolParameters"]["minUtxoValue"]
+        if "plutus:v1" in param.cost_models:
+            param.cost_models["PlutusV1"] = param.cost_models.pop("plutus:v1")
+        if "plutus:v2" in param.cost_models:
+            param.cost_models["PlutusV2"] = param.cost_models.pop("plutus:v2")
 
-            self._protocol_param = param
-        return self._protocol_param
+        result = self._query_genesis_config()
+        param.min_utxo = result["protocolParameters"]["minUtxoValue"]
+        return param
 
     @property
     def genesis_param(self) -> GenesisParameters:
         """Get chain genesis parameters"""
-        if not self._genesis_param or self._check_chain_tip_and_update():
-            result = self._query_genesis_config()
-            system_start_unix = int(
-                calendar.timegm(
-                    time.strptime(
-                        result["systemStart"].split(".")[0], "%Y-%m-%dT%H:%M:%S"
-                    ),
-                )
-            )
-            self._genesis_param = GenesisParameters(
-                active_slots_coefficient=self._fraction_parser(
-                    result["activeSlotsCoefficient"]
-                ),
-                update_quorum=result["updateQuorum"],
-                max_lovelace_supply=result["maxLovelaceSupply"],
-                network_magic=result["networkMagic"],
-                epoch_length=result["epochLength"],
-                system_start=system_start_unix,
-                slots_per_kes_period=result["slotsPerKesPeriod"],
-                slot_length=result["slotLength"],
-                max_kes_evolutions=result["maxKesEvolutions"],
-                security_param=result["securityParameter"],
-            )
+        if not self._genesis_param or self._is_chain_tip_updated():
+            self._genesis_param = self._fetch_genesis_param()
         return self._genesis_param
 
+    def _fetch_genesis_param(self) -> GenesisParameters:
+        result = self._query_genesis_config()
+        system_start_unix = int(
+            calendar.timegm(
+                time.strptime(result["systemStart"].split(".")[0], "%Y-%m-%dT%H:%M:%S"),
+            )
+        )
+        return GenesisParameters(
+            active_slots_coefficient=self._fraction_parser(
+                result["activeSlotsCoefficient"]
+            ),
+            update_quorum=result["updateQuorum"],
+            max_lovelace_supply=result["maxLovelaceSupply"],
+            network_magic=result["networkMagic"],
+            epoch_length=result["epochLength"],
+            system_start=system_start_unix,
+            slots_per_kes_period=result["slotsPerKesPeriod"],
+            slot_length=result["slotLength"],
+            max_kes_evolutions=result["maxKesEvolutions"],
+            security_param=result["securityParameter"],
+        )
+
     @property
     def network(self) -> Network:
         """Get current network"""