refactor(mypy): add stricter rules to some test modules [part II/VI]

hathor/simulator/tx_generator.py

@@ -20,12 +20,12 @@
 from hathor.conf.get_settings import get_global_settings
 from hathor.simulator.utils import NoCandidatesError, gen_new_double_spending, gen_new_tx
 from hathor.transaction.exceptions import RewardLocked
+from hathor.types import VertexId
 from hathor.util import Random
 from hathor.wallet.exceptions import InsufficientFunds
 
 if TYPE_CHECKING:
     from hathor.manager import HathorManager
-    from hathor.transaction import Transaction
 
 logger = get_logger()
 
@@ -62,7 +62,7 @@ def __init__(self, manager: 'HathorManager', rng: Random, *,
         # Most recent transactions generated here.
         # The lowest index has the most recent transaction.
         self.transactions_found: int = 0
-        self.latest_transactions: deque[Transaction] = deque()
+        self.latest_transactions: deque[VertexId] = deque()
 
         self.double_spending_only = False
 

pyproject.toml

@@ -154,6 +154,20 @@ warn_return_any = true
 # disallow_subclassing_any = true
 # disallow_untyped_calls = true
 
+# We also override to add disallow_untyped_defs for some specific test modules
+[[tool.mypy.overrides]]
+module = [
+    "tests.consensus.*",
+    "tests.crypto.*",
+    "tests.event.*",
+    "tests.execution_manager.*",
+    "tests.feature_activation.*",
+#    "tests.p2p.*",
+    "tests.pubsub.*",
+#    "tests.simulation.*",
+]
+disallow_untyped_defs = true
+
 [tool.pydantic-mypy]
 init_typed = true
 init_forbid_extra = true

tests/consensus/test_consensus.py

@@ -3,21 +3,22 @@
 from hathor.execution_manager import ExecutionManager
 from hathor.simulator.utils import add_new_block, add_new_blocks, gen_new_tx
 from hathor.transaction.storage import TransactionMemoryStorage
+from hathor.util import not_none
 from tests import unittest
 from tests.utils import add_blocks_unlock_reward, add_new_double_spending, add_new_transactions
 
 
 class BaseConsensusTestCase(unittest.TestCase):
     __test__ = False
 
-    def setUp(self):
+    def setUp(self) -> None:
         super().setUp()
         self.tx_storage = TransactionMemoryStorage()
         self.genesis = self.tx_storage.get_all_genesis()
         self.genesis_blocks = [tx for tx in self.genesis if tx.is_block]
         self.genesis_txs = [tx for tx in self.genesis if not tx.is_block]
 
-    def test_unhandled_exception(self):
+    def test_unhandled_exception(self) -> None:
         manager = self.create_peer('testnet', tx_storage=self.tx_storage)
 
         # Mine a few blocks in a row with no transaction but the genesis
@@ -45,7 +46,7 @@ class MyError(Exception):
         meta2 = tx2.get_metadata()
         self.assertEqual({self._settings.CONSENSUS_FAIL_ID}, meta2.voided_by)
 
-    def test_revert_block_high_weight(self):
+    def test_revert_block_high_weight(self) -> None:
         """ A conflict transaction will be propagated. At first, it will be voided.
         But, a new block with high weight will verify it, which will flip it to executed.
         """
@@ -108,7 +109,7 @@ def test_revert_block_high_weight(self):
 
         self.assertConsensusValid(manager)
 
-    def test_dont_revert_block_low_weight(self):
+    def test_dont_revert_block_low_weight(self) -> None:
         """ A conflict transaction will be propagated and voided.
         A new block with low weight will verify it, which won't be enough to flip to executed.
         So, it will remain voided.
@@ -162,7 +163,7 @@ def test_dont_revert_block_low_weight(self):
 
         self.assertConsensusValid(manager)
 
-    def test_dont_revert_block_high_weight_transaction_verify_other(self):
+    def test_dont_revert_block_high_weight_transaction_verify_other(self) -> None:
         """ A conflict transaction will be propagated and voided. But this transaction
         verifies its conflicting transaction. So, its accumulated weight will always be smaller
         than the others and it will never be executed.
@@ -180,8 +181,8 @@ def test_dont_revert_block_high_weight_transaction_verify_other(self):
         # Create a double spending transaction.
         conflicting_tx = add_new_double_spending(manager, tx=txs[-1])
         meta = conflicting_tx.get_metadata()
-        self.assertEqual(len(meta.conflict_with), 1)
-        self.assertIn(list(meta.conflict_with)[0], conflicting_tx.parents)
+        self.assertEqual(len(not_none(meta.conflict_with)), 1)
+        self.assertIn(not_none(meta.conflict_with)[0], conflicting_tx.parents)
 
         # Add a few transactions.
         add_new_transactions(manager, 10, advance_clock=15)
@@ -219,7 +220,7 @@ def test_dont_revert_block_high_weight_transaction_verify_other(self):
 
         self.assertConsensusValid(manager)
 
-    def test_dont_revert_block_high_weight_verify_both(self):
+    def test_dont_revert_block_high_weight_verify_both(self) -> None:
         """ A conflicting transaction will be propagated and voided. But the block with high weight
         verifies both the conflicting transactions, so this block will always be voided.
         """

tests/consensus/test_consensus2.py

@@ -1,12 +1,15 @@
 from hathor.graphviz import GraphvizVisualizer
+from hathor.manager import HathorManager
 from hathor.simulator.utils import gen_new_tx
+from hathor.transaction import Transaction
+from hathor.util import not_none
 from tests import unittest
 from tests.simulation.base import SimulatorTestCase
 from tests.utils import add_custom_tx
 
 
 class BaseConsensusSimulatorTestCase(SimulatorTestCase):
-    def checkConflict(self, tx1, tx2):
+    def checkConflict(self, tx1: Transaction, tx2: Transaction) -> None:
         meta1 = tx1.get_metadata()
         meta2 = tx2.get_metadata()
         self.assertIn(tx1.hash, meta2.conflict_with)
@@ -19,7 +22,7 @@ def checkConflict(self, tx1, tx2):
             cnt += 1
         self.assertLessEqual(cnt, 1)
 
-    def do_step(self, i, manager1, tx_base):
+    def do_step(self, i: int, manager1: HathorManager, tx_base: Transaction) -> Transaction:
         txA = add_custom_tx(manager1, [(tx_base, 0)], n_outputs=2)
         self.graphviz.labels[txA.hash] = f'txA-{i}'
 
@@ -52,7 +55,7 @@ def do_step(self, i, manager1, tx_base):
 
         return txH
 
-    def test_two_conflicts_intertwined_once(self):
+    def test_two_conflicts_intertwined_once(self) -> None:
         manager1 = self.create_peer()
         manager1.allow_mining_without_peers()
 
@@ -87,7 +90,7 @@ def test_two_conflicts_intertwined_once(self):
         # dot = self.graphviz.dot()
         # dot.render('dot0')
 
-    def test_two_conflicts_intertwined_multiple_times(self):
+    def test_two_conflicts_intertwined_multiple_times(self) -> None:
         manager1 = self.create_peer()
         manager1.allow_mining_without_peers()
 
@@ -113,7 +116,7 @@ def test_two_conflicts_intertwined_multiple_times(self):
         initial.weight = 25
         initial.update_hash()
         manager1.propagate_tx(initial, fails_silently=False)
-        self.graphviz.labels[initial.hash] = 'initial'
+        self.graphviz.labels[not_none(initial.hash)] = 'initial'
 
         x = initial
         x = self.do_step(0, manager1, x)

tests/consensus/test_consensus3.py

@@ -6,14 +6,14 @@
 
 
 class DoubleSpendingTestCase(unittest.TestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         super().setUp()
 
         self.network = 'testnet'
         self.manager1 = self.create_peer(self.network, unlock_wallet=True, enable_sync_v1=True, enable_sync_v2=False)
 
     @pytest.mark.xfail(strict=True)
-    def test_double_spending_attempt_1(self):
+    def test_double_spending_attempt_1(self) -> None:
         manager = self.manager1
 
         add_new_blocks(manager, 5, advance_clock=15)
@@ -38,7 +38,7 @@ def test_double_spending_attempt_1(self):
         manager.cpu_mining_service.resolve(tx_fund0)
         self.assertTrue(manager.propagate_tx(tx_fund0))
 
-        def do_step(tx_fund):
+        def do_step(tx_fund: Transaction) -> Transaction:
             inputs = [WalletInputInfo(tx_fund.hash, 0, manager.wallet.get_private_key(addr))]
             outputs = [WalletOutputInfo(decode_address(addr), 1, None)]
             tx1 = manager.wallet.prepare_transaction(Transaction, inputs, outputs, tx_fund.timestamp+1)
@@ -105,7 +105,7 @@ def do_step(tx_fund):
         self.assertConsensusValid(manager)
 
     @pytest.mark.xfail(strict=True)
-    def test_double_spending_attempt_2(self):
+    def test_double_spending_attempt_2(self) -> None:
         manager = self.manager1
 
         add_new_blocks(manager, 5, advance_clock=15)
@@ -128,7 +128,7 @@ def test_double_spending_attempt_2(self):
         manager.cpu_mining_service.resolve(tx_fund0)
         self.assertTrue(manager.propagate_tx(tx_fund0))
 
-        def do_step(tx_fund):
+        def do_step(tx_fund: Transaction) -> Transaction:
             inputs = [WalletInputInfo(tx_fund.hash, 0, manager.wallet.get_private_key(addr))]
             outputs = [WalletOutputInfo(decode_address(addr), 1, None)]
             tx1 = manager.wallet.prepare_transaction(Transaction, inputs, outputs, tx_fund.timestamp+1)

tests/consensus/test_consensus4.py

@@ -1,12 +1,22 @@
 from hathor.graphviz import GraphvizVisualizer
+from hathor.manager import HathorManager
+from hathor.transaction import Block
+from hathor.types import VertexId
 from tests import unittest
 from tests.simulation.base import SimulatorTestCase
 from tests.utils import gen_custom_tx
 
 
 class BaseConsensusSimulatorTestCase(SimulatorTestCase):
 
-    def create_chain(self, manager, first_parent_block_hash, length, prefix, tx_parents=None):
+    def create_chain(
+        self,
+        manager: HathorManager,
+        first_parent_block_hash: VertexId,
+        length: int,
+        prefix: str,
+        tx_parents: list[VertexId] | None = None
+    ) -> list[Block]:
         current = first_parent_block_hash
         v = []
         for i in range(length):
@@ -23,7 +33,7 @@ def create_chain(self, manager, first_parent_block_hash, length, prefix, tx_pare
             current = blk.hash
         return v
 
-    def test_conflict_with_parent_tx(self):
+    def test_conflict_with_parent_tx(self) -> None:
         manager1 = self.create_peer()
         manager1.allow_mining_without_peers()
 

tests/consensus/test_consensus5.py

@@ -1,12 +1,22 @@
 from hathor.graphviz import GraphvizVisualizer
+from hathor.manager import HathorManager
+from hathor.transaction import Block
+from hathor.types import VertexId
 from tests import unittest
 from tests.simulation.base import SimulatorTestCase
 from tests.utils import gen_custom_tx
 
 
 class BaseConsensusSimulatorTestCase(SimulatorTestCase):
 
-    def create_chain(self, manager, first_parent_block_hash, length, prefix, tx_parents=None):
+    def create_chain(
+        self,
+        manager: HathorManager,
+        first_parent_block_hash: VertexId,
+        length: int,
+        prefix: str,
+        tx_parents: list[VertexId] | None = None
+    ) -> list[Block]:
         current = first_parent_block_hash
         v = []
         for i in range(length):
@@ -23,7 +33,7 @@ def create_chain(self, manager, first_parent_block_hash, length, prefix, tx_pare
             current = blk.hash
         return v
 
-    def test_conflict_with_parent_tx(self):
+    def test_conflict_with_parent_tx(self) -> None:
         manager1 = self.create_peer()
         manager1.allow_mining_without_peers()
 

tests/consensus/test_soft_voided.py

@@ -1,7 +1,11 @@
+from typing import Iterator
+
 from hathor.graphviz import GraphvizVisualizer
-from hathor.simulator import FakeConnection, Simulator
+from hathor.simulator import FakeConnection, RandomTransactionGenerator, Simulator
 from hathor.simulator.trigger import StopAfterNTransactions
 from hathor.simulator.utils import gen_new_tx
+from hathor.transaction import Block
+from hathor.types import VertexId
 from tests import unittest
 from tests.simulation.base import SimulatorTestCase
 from tests.utils import add_custom_tx
@@ -10,14 +14,19 @@
 class BaseSoftVoidedTestCase(SimulatorTestCase):
     seed_config = 5988775361793628170
 
-    def assertNoParentsAreSoftVoided(self, tx):
+    def assertNoParentsAreSoftVoided(self, tx: Block) -> None:
+        assert tx.storage is not None
         for h in tx.parents:
             tx2 = tx.storage.get_transaction(h)
             tx2_meta = tx2.get_metadata()
             tx2_voided_by = tx2_meta.voided_by or set()
             self.assertNotIn(self._settings.SOFT_VOIDED_ID, tx2_voided_by)
 
-    def _run_test(self, simulator, soft_voided_tx_ids):
+    def _run_test(
+        self,
+        simulator: Simulator,
+        soft_voided_tx_ids: set[VertexId]
+    ) -> Iterator[RandomTransactionGenerator]:
         manager1 = self.create_peer(soft_voided_tx_ids=soft_voided_tx_ids, simulator=simulator)
         manager1.allow_mining_without_peers()
 
@@ -127,7 +136,7 @@ def _run_test(self, simulator, soft_voided_tx_ids):
         # dot = graphviz.dot()
         # dot.render('dot0')
 
-    def _get_txA_hash(self):
+    def _get_txA_hash(self) -> VertexId:
         simulator = Simulator(seed=self.simulator.seed)
         simulator.start()
 
@@ -140,7 +149,7 @@ def _get_txA_hash(self):
 
         return txA_hash
 
-    def test_soft_voided(self):
+    def test_soft_voided(self) -> None:
         txA_hash = self._get_txA_hash()
         soft_voided_tx_ids = set([
             txA_hash,