refactor: replace Optional[type] with type | None

Author: mkoura

cardano_node_tests/cluster_management/cache.py

@@ -12,7 +12,7 @@ class ClusterManagerCache:
     """
 
     # single `ClusterLib` instance can be used in multiple tests executed on the same worker
-    cluster_obj: tp.Optional[clusterlib.ClusterLib] = None
+    cluster_obj: clusterlib.ClusterLib | None = None
     # data for initialized cluster instance
     test_data: dict = dataclasses.field(default_factory=dict)
     addrs_data: dict = dataclasses.field(default_factory=dict)

cardano_node_tests/cluster_management/cluster_getter.py

@@ -184,7 +184,7 @@ def _respin(self, start_cmd: str = "", stop_cmd: str = "") -> bool:  # noqa: C90
             f"stop_cmd='{startup_files.stop_script}'"
         )
 
-        excp: tp.Optional[Exception] = None
+        excp: Exception | None = None
         for i in range(2):
             if i > 0:
                 self.log(

cardano_node_tests/cluster_management/manager.py

@@ -290,7 +290,7 @@ def on_test_stop(self) -> None:
     def _get_resources_by_glob(
         self,
         glob: str,
-        from_set: tp.Optional[tp.Iterable[str]] = None,
+        from_set: tp.Iterable[str] | None = None,
     ) -> list[str]:
         if from_set is not None and isinstance(from_set, str):
             msg = "`from_set` cannot be a string"
@@ -305,8 +305,8 @@ def _get_resources_by_glob(
 
     def get_locked_resources(
         self,
-        from_set: tp.Optional[tp.Iterable[str]] = None,
-        worker_id: tp.Optional[str] = None,
+        from_set: tp.Iterable[str] | None = None,
+        worker_id: str | None = None,
     ) -> list[str]:
         """Get resources locked by worker.
 
@@ -317,8 +317,8 @@ def get_locked_resources(
 
     def get_used_resources(
         self,
-        from_set: tp.Optional[tp.Iterable[str]] = None,
-        worker_id: tp.Optional[str] = None,
+        from_set: tp.Iterable[str] | None = None,
+        worker_id: str | None = None,
     ) -> list[str]:
         """Get resources used by worker.
 

cardano_node_tests/tests/delegation.py

@@ -107,7 +107,7 @@ def cluster_and_pool(
 
 def db_check_delegation(
     pool_user: clusterlib.PoolUser | PoolUserScript,
-    db_record: tp.Optional[dbsync_types.TxRecord],
+    db_record: dbsync_types.TxRecord | None,
     deleg_epoch: int,
     pool_id: str,
     check_registration: bool = True,
@@ -128,9 +128,9 @@ def delegate_stake_addr(
     cluster_obj: clusterlib.ClusterLib,
     addrs_data: dict,
     temp_template: str,
-    pool_user: tp.Optional[clusterlib.PoolUser] = None,
+    pool_user: clusterlib.PoolUser | None = None,
     pool_id: str = "",
-    cold_vkey: tp.Optional[pl.Path] = None,
+    cold_vkey: pl.Path | None = None,
     amount: int = 100_000_000,
     use_build_cmd: bool = False,
 ) -> DelegationOut:

cardano_node_tests/tests/kes.py

@@ -28,8 +28,8 @@ def check_kes_period_info_result(  # noqa: C901
     kes_output: dict[str, tp.Any],
     expected_scenario: str,
     check_id: str,
-    expected_start_kes: tp.Optional[int] = None,
-    pool_num: tp.Optional[int] = None,
+    expected_start_kes: int | None = None,
+    pool_num: int | None = None,
 ) -> list[str]:
     """Check output `kes-period-info` command.
 

cardano_node_tests/tests/plutus_common.py

@@ -1,7 +1,6 @@
 import dataclasses
 import itertools
 import pathlib as pl
-import typing as tp
 
 import pytest
 from cardano_clusterlib import clusterlib
@@ -469,13 +468,13 @@ class PlutusScriptData:
 @dataclasses.dataclass(frozen=True, order=True)
 class PlutusOp:
     script_file: clusterlib.FileType
-    datum_file: tp.Optional[pl.Path] = None
-    datum_cbor_file: tp.Optional[pl.Path] = None
-    datum_value: tp.Optional[str] = None
-    redeemer_file: tp.Optional[pl.Path] = None
-    redeemer_cbor_file: tp.Optional[pl.Path] = None
-    redeemer_value: tp.Optional[str] = None
-    execution_cost: tp.Optional[ExecutionCost] = None
+    datum_file: pl.Path | None = None
+    datum_cbor_file: pl.Path | None = None
+    datum_value: str | None = None
+    redeemer_file: pl.Path | None = None
+    redeemer_cbor_file: pl.Path | None = None
+    redeemer_value: str | None = None
+    execution_cost: ExecutionCost | None = None
 
 
 @dataclasses.dataclass(frozen=True, order=True)
@@ -734,7 +733,7 @@ def create_script_context_w_blockers(
     cluster_obj: clusterlib.ClusterLib,
     plutus_version: int,
     redeemer_file: pl.Path,
-    tx_file: tp.Optional[pl.Path] = None,
+    tx_file: pl.Path | None = None,
 ) -> None:
     """Run the `create-script-context` command (available in plutus-apps).
 

cardano_node_tests/tests/test_chain_transactions.py

@@ -3,7 +3,6 @@
 import logging
 import pathlib as pl
 import time
-import typing as tp
 
 import allure
 import pytest
@@ -50,7 +49,7 @@ def _gen_signed_tx(
     out_addr: clusterlib.AddressRecord,
     tx_name: str,
     fee: int,
-    invalid_hereafter: tp.Optional[int] = None,
+    invalid_hereafter: int | None = None,
 ) -> tuple[clusterlib.UTXOData, clusterlib.TxRawOutput, pl.Path]:
     """Generate Tx and return Tx output in a format that can be used as input for next Tx."""
     send_amount = txin.amount - fee

cardano_node_tests/tests/test_pools.py

@@ -12,7 +12,6 @@
 import json
 import logging
 import pathlib as pl
-import typing as tp
 
 import allure
 import hypothesis
@@ -126,8 +125,8 @@ def _register_stake_pool_w_build(
     vrf_vkey_file: clusterlib.FileType,
     cold_key_pair: clusterlib.ColdKeyPair,
     tx_name: str,
-    reward_account_vkey_file: tp.Optional[clusterlib.FileType] = None,
-    deposit: tp.Optional[int] = None,
+    reward_account_vkey_file: clusterlib.FileType | None = None,
+    deposit: int | None = None,
     destination_dir: clusterlib.FileType = ".",
 ) -> tuple[pl.Path, clusterlib.TxRawOutput]:
     """Register a stake pool using a `transaction build` command.
@@ -335,7 +334,7 @@ def _create_register_pool(
     temp_dir: pl.Path,
     pool_owners: list[clusterlib.PoolUser],
     pool_data: clusterlib.PoolData,
-    request: tp.Optional[FixtureRequest] = None,
+    request: FixtureRequest | None = None,
     use_build_cmd: bool = False,
 ) -> clusterlib.PoolCreationOutput:
     """Create and register a stake pool.
@@ -402,7 +401,7 @@ def _create_register_pool_delegate_stake_tx(
     temp_template: str,
     temp_dir: pl.Path,
     pool_data: clusterlib.PoolData,
-    request: tp.Optional[FixtureRequest] = None,
+    request: FixtureRequest | None = None,
     use_build_cmd: bool = False,
 ) -> clusterlib.PoolCreationOutput:
     """Create and register a stake pool, delegate stake address - all in single TX.
@@ -532,7 +531,7 @@ def _create_register_pool_tx_delegate_stake_tx(
     temp_template: str,
     temp_dir: pl.Path,
     pool_data: clusterlib.PoolData,
-    request: tp.Optional[FixtureRequest] = None,
+    request: FixtureRequest | None = None,
     use_build_cmd: bool = False,
 ) -> clusterlib.PoolCreationOutput:
     """Create and register a stake pool - first TX; delegate stake address - second TX.

cardano_node_tests/tests/test_reconnect.py

@@ -72,7 +72,7 @@ def node_query_utxo(
         cluster_obj: clusterlib.ClusterLib,
         node: str,
         address: str = "",
-        tx_raw_output: tp.Optional[clusterlib.TxRawOutput] = None,
+        tx_raw_output: clusterlib.TxRawOutput | None = None,
     ) -> list[clusterlib.UTXOData]:
         """Query UTxO on given node."""
         orig_socket = os.environ.get("CARDANO_NODE_SOCKET_PATH")

cardano_node_tests/tests/test_rollback.py

@@ -9,7 +9,6 @@
 import pathlib as pl
 import shutil
 import time
-import typing as tp
 
 import allure
 import pytest
@@ -137,7 +136,7 @@ def node_query_utxo(
         cluster_obj: clusterlib.ClusterLib,
         node: str,
         address: str = "",
-        tx_raw_output: tp.Optional[clusterlib.TxRawOutput] = None,
+        tx_raw_output: clusterlib.TxRawOutput | None = None,
     ) -> list[clusterlib.UTXOData]:
         """Query UTxO on given node."""
         orig_socket = os.environ.get("CARDANO_NODE_SOCKET_PATH")