Merge pull request #2854 from IntersectMBO/typecheck_cache_fixture

feat: add type annotations to FixtureCache

cardano_node_tests/cluster_management/cluster_management.py

@@ -3,4 +3,5 @@
 # flake8: noqa
 from cardano_node_tests.cluster_management.common import CLUSTER_LOCK
 from cardano_node_tests.cluster_management.manager import ClusterManager
+from cardano_node_tests.cluster_management.manager import FixtureCache
 from cardano_node_tests.cluster_management.resources import Resources

cardano_node_tests/cluster_management/manager.py

@@ -29,6 +29,8 @@
 
 LOGGER = logging.getLogger(__name__)
 
+T = tp.TypeVar("T")
+
 if configuration.CLUSTERS_COUNT > 1 and configuration.DEV_CLUSTER_RUNNING:
     msg = "Cannot run multiple cluster instances when 'DEV_CLUSTER_RUNNING' is set."
     raise RuntimeError(msg)
@@ -43,10 +45,10 @@ def _get_manager_fixture_line_str() -> str:
 
 
 @dataclasses.dataclass
-class FixtureCache:
+class FixtureCache(tp.Generic[T]):
     """Cache for a fixture."""
 
-    value: tp.Any
+    value: T | None
 
 
 class ClusterManager:
@@ -219,12 +221,12 @@ def respin_on_failure(self) -> tp.Iterator[None]:
             raise
 
     @contextlib.contextmanager
-    def cache_fixture(self, key: str = "") -> tp.Iterator[FixtureCache]:
+    def cache_fixture(self, key: str = "") -> tp.Iterator[FixtureCache[tp.Any]]:
         """Cache fixture value - context manager."""
         key_str = key or _get_manager_fixture_line_str()
         key_hash = int(hashlib.sha1(key_str.encode("utf-8")).hexdigest(), 16)
         cached_value = self.cache.test_data.get(key_hash)
-        container = FixtureCache(value=cached_value)
+        container: FixtureCache[tp.Any] = FixtureCache(value=cached_value)
 
         yield container
 

cardano_node_tests/tests/common.py

@@ -347,9 +347,10 @@ def _create_addrs() -> list[clusterlib.AddressRecord]:
         return addrs
 
     if caching_key:
+        fixture_cache: cluster_management.FixtureCache[list[clusterlib.AddressRecord] | None]
         with cluster_manager.cache_fixture(key=caching_key) as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             addrs = _create_addrs()
             fixture_cache.value = addrs
@@ -410,9 +411,10 @@ def _create_pool_users() -> list[clusterlib.PoolUser]:
         return users
 
     if caching_key:
+        fixture_cache: cluster_management.FixtureCache[list[clusterlib.PoolUser] | None]
         with cluster_manager.cache_fixture(key=caching_key) as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             users = _create_pool_users()
             fixture_cache.value = users

cardano_node_tests/tests/test_mir_certs.py

@@ -82,9 +82,10 @@ def registered_users(
     """Register pool user's stake address."""
     registered = pool_users[1:3]
 
+    fixture_cache: cluster_management.FixtureCache[list[clusterlib.PoolUser] | None]
     with cluster_manager.cache_fixture() as fixture_cache:
-        if fixture_cache.value:
-            return fixture_cache.value  # type: ignore
+        if fixture_cache.value is not None:
+            return fixture_cache.value
         fixture_cache.value = registered
 
     for i, pool_user in enumerate(registered):

cardano_node_tests/tests/test_native_tokens.py

@@ -1665,9 +1665,10 @@ def new_token(
         cluster: clusterlib.ClusterLib,
         payment_addrs: list[clusterlib.AddressRecord],
     ) -> clusterlib_utils.TokenRecord:
+        fixture_cache: cluster_management.FixtureCache[clusterlib_utils.TokenRecord | None]
         with cluster_manager.cache_fixture() as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             rand = clusterlib.get_rand_str(4)
             temp_template = f"test_tx_new_token_ci{cluster_manager.cluster_instance_num}_{rand}"

cardano_node_tests/tests/test_tx_basic.py

@@ -53,9 +53,10 @@ def byron_addrs(
         cluster: clusterlib.ClusterLib,
     ) -> list[clusterlib.AddressRecord]:
         """Create 2 new Byron payment addresses."""
+        fixture_cache: cluster_management.FixtureCache[list[clusterlib.AddressRecord] | None]
         with cluster_manager.cache_fixture() as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             new_byron_addrs = [
                 clusterlib_utils.gen_byron_addr(

cardano_node_tests/tests/tests_conway/conway_common.py

@@ -129,9 +129,10 @@ def _create_user() -> clusterlib.PoolUser:
         return pool_user
 
     if caching_key:
+        fixture_cache: cluster_management.FixtureCache[clusterlib.PoolUser | None]
         with cluster_manager.cache_fixture(key=caching_key) as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             pool_user = _create_user()
             fixture_cache.value = pool_user

cardano_node_tests/tests/tests_conway/test_drep.py

@@ -98,9 +98,10 @@ def get_custom_drep(
     if cluster_nodes.get_cluster_type().type != cluster_nodes.ClusterType.LOCAL:
         pytest.skip("runs only on local cluster")
 
+    fixture_cache: cluster_management.FixtureCache[governance_utils.DRepRegistration | None]
     with cluster_manager.cache_fixture(key=caching_key) as fixture_cache:
-        if fixture_cache.value:
-            return fixture_cache.value  # type: ignore
+        if fixture_cache.value is not None:
+            return fixture_cache.value
 
         reg_drep = create_drep(
             name_template=name_template,

cardano_node_tests/tests/tests_plutus/test_mint_build.py

@@ -32,6 +32,8 @@
     pytest.mark.plutus,
 ]
 
+PHorizonFundsT = tuple[list[clusterlib.UTXOData], list[clusterlib.UTXOData], clusterlib.TxRawOutput]
+
 
 @pytest.fixture
 def payment_addrs(
@@ -59,11 +61,12 @@ def past_horizon_funds(
         cluster_manager: cluster_management.ClusterManager,
         cluster: clusterlib.ClusterLib,
         payment_addrs: list[clusterlib.AddressRecord],
-    ) -> tuple[list[clusterlib.UTXOData], list[clusterlib.UTXOData], clusterlib.TxRawOutput]:
+    ) -> PHorizonFundsT:
         """Create UTxOs for `test_ttl_horizon`."""
+        fixture_cache: cluster_management.FixtureCache[PHorizonFundsT | None]
         with cluster_manager.cache_fixture() as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             temp_template = common.get_test_id(cluster)
             payment_addr = payment_addrs[0]
@@ -1140,9 +1143,7 @@ def test_ttl_horizon(
         self,
         cluster: clusterlib.ClusterLib,
         payment_addrs: list[clusterlib.AddressRecord],
-        past_horizon_funds: tuple[
-            list[clusterlib.UTXOData], list[clusterlib.UTXOData], clusterlib.TxRawOutput
-        ],
+        past_horizon_funds: PHorizonFundsT,
         plutus_version: str,
         ttl_offset: int,
     ):

cardano_node_tests/tests/tests_plutus/test_spend_negative_raw.py

@@ -734,9 +734,10 @@ def fund_script_guessing_game_v1(
         cluster_manager: cluster_management.ClusterManager,
         cluster: clusterlib.ClusterLib,
     ) -> FundTupleT:
+        fixture_cache: cluster_management.FixtureCache[FundTupleT | None]
         with cluster_manager.cache_fixture() as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             temp_template = common.get_test_id(cluster)
 
@@ -756,9 +757,10 @@ def fund_script_guessing_game_v2(
         cluster_manager: cluster_management.ClusterManager,
         cluster: clusterlib.ClusterLib,
     ) -> FundTupleT:
+        fixture_cache: cluster_management.FixtureCache[FundTupleT | None]
         with cluster_manager.cache_fixture() as fixture_cache:
-            if fixture_cache.value:
-                return fixture_cache.value  # type: ignore
+            if fixture_cache.value is not None:
+                return fixture_cache.value
 
             temp_template = common.get_test_id(cluster)
 

cardano_node_tests/utils/governance_setup.py

@@ -258,7 +258,7 @@ def get_default_governance(
     cluster_env = cluster_nodes.get_cluster_env()
     gov_data_dir = cluster_env.state_dir / GOV_DATA_DIR
     gov_data_store = gov_data_dir / GOV_DATA_STORE
-    governance_data = None
+    governance_data: governance_utils.GovernanceRecords | None = None
 
     def _setup_gov() -> governance_utils.GovernanceRecords | None:
         if gov_data_store.exists():
@@ -280,16 +280,17 @@ def _setup_gov() -> governance_utils.GovernanceRecords | None:
 
     gov_data_checksum = helpers.checksum(gov_data_store)
 
+    fixture_cache: cluster_management.FixtureCache[governance_utils.GovernanceRecords | None]
     with cluster_manager.cache_fixture(key=gov_data_checksum) as fixture_cache:
-        if fixture_cache.value:
-            return fixture_cache.value  # type: ignore
+        if fixture_cache.value is not None:
+            return fixture_cache.value
 
         if governance_data is None:
             with open(gov_data_store, "rb") as in_data:
                 governance_data = pickle.load(in_data)
 
         fixture_cache.value = governance_data
-        return fixture_cache.value  # type: ignore
+        return fixture_cache.value
 
 
 def save_default_governance(