solver service, api, rpc

Author: almogdepaz

chia/_tests/conftest.py

@@ -70,6 +70,7 @@
 )
 from chia.simulator.start_simulator import SimulatorFullNodeService
 from chia.simulator.wallet_tools import WalletTool
+from chia.solver.solver_service import SolverService
 from chia.timelord.timelord_service import TimelordService
 from chia.types.peer_info import PeerInfo
 from chia.util.config import create_default_chia_config, lock_and_load_config

chia/_tests/core/custom_types/test_proof_of_space.py

@@ -223,7 +223,7 @@ def test_calculate_plot_difficulty(height: uint32, difficulty: uint8) -> None:
 
 class TestProofOfSpace:
     @pytest.mark.parametrize("prefix_bits", [DEFAULT_CONSTANTS.NUMBER_ZERO_BITS_PLOT_FILTER_V1, 8, 7, 6, 5, 1, 0])
-    def test_can_create_proof(self, prefix_bits: int, seeded_random: random.Random) -> None:
+    def test_can_create_proof(self, prefix_bits: uint8, seeded_random: random.Random) -> None:
         """
         Tests that the change of getting a correct proof is exactly 1/target_filter.
         """

chia/_tests/harvester/test_harvester_api.py

@@ -17,126 +17,100 @@
 from chia.simulator.block_tools import BlockTools
 
 
-def create_signage_point_harvester_from_constants(bt: BlockTools) -> harvester_protocol.NewSignagePointHarvester:
-    """create a NewSignagePointHarvester using real constants from block tools"""
-    # use the pre-generated signage point data from network_protocol_data.py
-    # but with real constants from block_tools
-    from chia._tests.util.network_protocol_data import new_signage_point_harvester
+def signage_point_from_block(bt: BlockTools) -> harvester_protocol.NewSignagePointHarvester:
+    """Create a real NewSignagePointHarvester from actual blockchain blocks."""
+    # generate real blocks using BlockTools
+    blocks = bt.get_consecutive_blocks(
+        num_blocks=3,
+        guarantee_transaction_block=True,
+        farmer_reward_puzzle_hash=bt.farmer_ph,
+    )
+    block = blocks[-1]  # always use the last block
+    # extract real signage point data from the block
+    sp_index = block.reward_chain_block.signage_point_index
+    challenge_hash = block.reward_chain_block.pos_ss_cc_challenge_hash
+    sp_hash = (
+        block.reward_chain_block.reward_chain_sp_vdf.output.get_hash()
+        if block.reward_chain_block.reward_chain_sp_vdf
+        else challenge_hash
+    )
 
-    # create a version with real constants values
     return harvester_protocol.NewSignagePointHarvester(
-        challenge_hash=new_signage_point_harvester.challenge_hash,
+        challenge_hash=challenge_hash,
         difficulty=uint64(bt.constants.DIFFICULTY_STARTING),
         sub_slot_iters=uint64(bt.constants.SUB_SLOT_ITERS_STARTING),
-        signage_point_index=new_signage_point_harvester.signage_point_index,
-        sp_hash=new_signage_point_harvester.sp_hash,
-        pool_difficulties=[],  # empty for simplicity, unless testing pool functionality
-        peak_height=new_signage_point_harvester.peak_height,
-        last_tx_height=new_signage_point_harvester.last_tx_height,
+        signage_point_index=sp_index,
+        sp_hash=sp_hash,
+        pool_difficulties=[],
+        peak_height=block.height,
+        last_tx_height=block.height,
     )
 
 
-@pytest.mark.anyio
-async def test_new_signage_point_harvester_no_keys(
-    harvester_farmer_environment: HarvesterFarmerEnvironment,
-) -> None:
-    """test that new_signage_point_harvester returns early when no keys available"""
-    _farmer_service, _farmer_rpc_client, harvester_service, _harvester_rpc_client, bt = harvester_farmer_environment
-    harvester_api = harvester_service._server.api
-    assert isinstance(harvester_api, HarvesterAPI)
-
-    # create real signage point data from block tools
-    new_challenge = create_signage_point_harvester_from_constants(bt)
-
-    # mock plot manager to return false for public_keys_available
-    with patch.object(harvester_api.harvester.plot_manager, "public_keys_available", return_value=False):
-        mock_peer = MagicMock(spec=WSChiaConnection)
+def create_plot_info() -> PlotInfo:
+    """Create a realistic PlotInfo mock for testing."""
+    mock_prover = MagicMock()
+    mock_prover.get_id.return_value = bytes32(b"plot_id_123456789012345678901234")  # exactly 32 bytes
+    mock_prover.get_size.return_value = 32  # standard k32 plot
+    mock_prover.get_qualities_for_challenge.return_value = [
+        bytes32(b"quality_123456789012345678901234")
+    ]  # exactly 32 bytes
+    mock_plot_info = MagicMock(spec=PlotInfo)
+    mock_plot_info.prover = mock_prover
+    mock_plot_info.pool_contract_puzzle_hash = None
 
-        result = harvester_api.new_signage_point_harvester(new_challenge, mock_peer)
-        assert result is None
+    return mock_plot_info
 
 
 @pytest.mark.anyio
-async def test_new_signage_point_harvester_happy_path(
-    harvester_farmer_environment: HarvesterFarmerEnvironment,
-) -> None:
-    """test successful signage point processing with valid plots"""
-    _farmer_service, _farmer_rpc_client, harvester_service, _harvester_rpc_client, bt = harvester_farmer_environment
+async def test_new_signage_point_harvester(harvester_farmer_environment: HarvesterFarmerEnvironment) -> None:
+    """Test successful signage point processing with real blockchain data."""
+    _, _, harvester_service, _, bt = harvester_farmer_environment
     harvester_api = harvester_service._server.api
     assert isinstance(harvester_api, HarvesterAPI)
-
-    # create real signage point data from block tools
-    new_challenge = create_signage_point_harvester_from_constants(bt)
-
+    # use real signage point data from actual block
+    new_challenge = signage_point_from_block(bt)
+    # harvester doesn't accept incoming connections, so use mock peer like other tests
     mock_peer = MagicMock(spec=WSChiaConnection)
-
-    # create mock plot info
-    mock_prover = MagicMock()
-    mock_prover.get_id.return_value = bytes32(b"2" * 32)
-    mock_prover.get_size.return_value = 32
-    mock_prover.get_qualities_for_challenge.return_value = [bytes32(b"quality" + b"0" * 25)]
-
-    mock_plot_info = MagicMock(spec=PlotInfo)
-    mock_plot_info.prover = mock_prover
-    mock_plot_info.pool_contract_puzzle_hash = None
-
+    # create realistic plot info for testing
+    mock_plot_info = create_plot_info()
     plot_path = Path("/fake/plot.plot")
 
     with patch.object(harvester_api.harvester.plot_manager, "public_keys_available", return_value=True):
         with patch.object(harvester_api.harvester.plot_manager, "plots", {plot_path: mock_plot_info}):
-            with patch("chia.harvester.harvester_api.passes_plot_filter", return_value=True):
-                with patch("chia.harvester.harvester_api.calculate_pos_challenge") as mock_calc_pos:
-                    mock_calc_pos.return_value = bytes32(b"sp_challenge" + b"0" * 20)
-
-                    with patch("chia.harvester.harvester_api.calculate_iterations_quality") as mock_calc_iter:
-                        # set required_iters low enough to pass the sp_interval_iters check
-                        mock_calc_iter.return_value = uint64(1000)
-
-                        with patch("chia.harvester.harvester_api.calculate_sp_interval_iters") as mock_sp_interval:
-                            mock_sp_interval.return_value = uint64(10000)
-
-                            with patch.object(mock_prover, "get_full_proof") as mock_get_proof:
-                                mock_proof = MagicMock(spec=ProofOfSpace)
-                                mock_get_proof.return_value = mock_proof, None
-
-                                result = harvester_api.new_signage_point_harvester(new_challenge, mock_peer)
-                                # function returns None but should have processed the plot
-                                assert result is None
+            # let passes_plot_filter, calculate_pos_challenge, and calculate_sp_interval_iters use real implementations
+            with patch("chia.harvester.harvester_api.calculate_iterations_quality", return_value=uint64(1000)):
+                with patch.object(mock_plot_info.prover, "get_full_proof") as mock_get_proof:
+                    mock_proof = MagicMock(spec=ProofOfSpace)
+                    mock_get_proof.return_value = mock_proof, None
+                    await harvester_api.new_signage_point_harvester(new_challenge, mock_peer)
 
 
 @pytest.mark.anyio
-async def test_new_signage_point_harvester_pool_difficulty_override(
+async def test_new_signage_point_harvester_pool_difficulty(
     harvester_farmer_environment: HarvesterFarmerEnvironment,
 ) -> None:
-    """test that pool difficulty overrides are applied correctly"""
-    _farmer_service, _farmer_rpc_client, harvester_service, _harvester_rpc_client, bt = harvester_farmer_environment
+    """Test pool difficulty overrides with real blockchain signage points."""
+    _, _, harvester_service, _, bt = harvester_farmer_environment
     harvester_api = harvester_service._server.api
     assert isinstance(harvester_api, HarvesterAPI)
 
+    # harvester doesn't accept incoming connections, so use mock peer like other tests
     mock_peer = MagicMock(spec=WSChiaConnection)
-
     pool_puzzle_hash = bytes32(b"pool" + b"0" * 28)
 
-    mock_prover = MagicMock()
-    mock_prover.get_id.return_value = bytes32(b"2" * 32)
-    mock_prover.get_size.return_value = 32
-    mock_prover.get_qualities_for_challenge.return_value = [bytes32(b"quality" + b"0" * 25)]
-
-    mock_plot_info = MagicMock(spec=PlotInfo)
-    mock_plot_info.prover = mock_prover
+    # create realistic plot info for testing
+    mock_plot_info = create_plot_info()
     mock_plot_info.pool_contract_puzzle_hash = pool_puzzle_hash
-
-    plot_path = Path("/fake/plot.plot")
-
+    plot_path = Path("/fake/pool_plot.plot")
     pool_difficulty = PoolDifficulty(
         pool_contract_puzzle_hash=pool_puzzle_hash,
         difficulty=uint64(500),  # lower than main difficulty
         sub_slot_iters=uint64(67108864),  # different from main
     )
 
-    # create real signage point data from constants with pool difficulty
-    new_challenge = create_signage_point_harvester_from_constants(bt)
-    # override with pool difficulty for this test
+    # create signage point from real block with pool difficulty
+    new_challenge = signage_point_from_block(bt)
     new_challenge = harvester_protocol.NewSignagePointHarvester(
         challenge_hash=new_challenge.challenge_hash,
         difficulty=new_challenge.difficulty,
@@ -150,60 +124,44 @@ async def test_new_signage_point_harvester_pool_difficulty_override(
 
     with patch.object(harvester_api.harvester.plot_manager, "public_keys_available", return_value=True):
         with patch.object(harvester_api.harvester.plot_manager, "plots", {plot_path: mock_plot_info}):
+            # mock passes_plot_filter to return True so we can test pool difficulty logic
             with patch("chia.harvester.harvester_api.passes_plot_filter", return_value=True):
-                with patch("chia.harvester.harvester_api.calculate_pos_challenge") as mock_calc_pos:
-                    mock_calc_pos.return_value = bytes32(b"sp_challenge" + b"0" * 20)
-
-                    with patch("chia.harvester.harvester_api.calculate_iterations_quality") as mock_calc_iter:
-                        mock_calc_iter.return_value = uint64(1000)
-
-                        with patch("chia.harvester.harvester_api.calculate_sp_interval_iters") as mock_sp_interval:
-                            mock_sp_interval.return_value = uint64(10000)
-
-                            with patch.object(mock_prover, "get_full_proof") as mock_get_proof:
-                                mock_proof = MagicMock(spec=ProofOfSpace)
-                                mock_get_proof.return_value = mock_proof, None
-
-                                result = harvester_api.new_signage_point_harvester(new_challenge, mock_peer)
-
-                                # verify that calculate_iterations_quality was called with pool difficulty
-                                mock_calc_iter.assert_called()
-                                call_args = mock_calc_iter.call_args[0]
-                                assert call_args[3] == uint64(500)  # pool difficulty was used
-
-                                assert result is None
+                with patch("chia.harvester.harvester_api.calculate_iterations_quality") as mock_calc_iter:
+                    mock_calc_iter.return_value = uint64(1000)
+                    with patch.object(mock_plot_info.prover, "get_full_proof") as mock_get_proof:
+                        mock_proof = MagicMock(spec=ProofOfSpace)
+                        mock_get_proof.return_value = mock_proof, None
+                        await harvester_api.new_signage_point_harvester(new_challenge, mock_peer)
+                        # verify that calculate_iterations_quality was called with pool difficulty
+                        mock_calc_iter.assert_called()
+                        call_args = mock_calc_iter.call_args[0]
+                        assert call_args[3] == uint64(500)  # pool difficulty was used
 
 
 @pytest.mark.anyio
 async def test_new_signage_point_harvester_prover_error(
     harvester_farmer_environment: HarvesterFarmerEnvironment,
 ) -> None:
-    """test error handling when prover fails"""
-    _farmer_service, _farmer_rpc_client, harvester_service, _harvester_rpc_client, bt = harvester_farmer_environment
+    """Test error handling when prover fails using real blockchain data."""
+    _, _, harvester_service, _, bt = harvester_farmer_environment
     harvester_api = harvester_service._server.api
     assert isinstance(harvester_api, HarvesterAPI)
 
-    # create real signage point data from block tools
-    new_challenge = create_signage_point_harvester_from_constants(bt)
+    # create signage point from real block
+    new_challenge = signage_point_from_block(bt)
 
     mock_peer = MagicMock(spec=WSChiaConnection)
 
-    mock_prover = MagicMock()
-    mock_prover.get_id.return_value = bytes32(b"2" * 32)
-    mock_prover.get_qualities_for_challenge.side_effect = RuntimeError("test error")
-
-    mock_plot_info = MagicMock(spec=PlotInfo)
-    mock_plot_info.prover = mock_prover
-    mock_plot_info.pool_contract_puzzle_hash = None
-
+    # create realistic plot info for testing
+    mock_plot_info = create_plot_info()
     plot_path = Path("/fake/plot.plot")
 
     with patch.object(harvester_api.harvester.plot_manager, "public_keys_available", return_value=True):
         with patch.object(harvester_api.harvester.plot_manager, "plots", {plot_path: mock_plot_info}):
-            with patch("chia.harvester.harvester_api.passes_plot_filter", return_value=True):
-                with patch("chia.harvester.harvester_api.calculate_pos_challenge") as mock_calc_pos:
-                    mock_calc_pos.return_value = bytes32(b"sp_challenge" + b"0" * 20)
-
-                    # should not raise exception, should handle error gracefully
-                    result = harvester_api.new_signage_point_harvester(new_challenge, mock_peer)
-                    assert result is None
+            # let passes_plot_filter and calculate_pos_challenge use real implementations
+            # make the prover fail during quality check
+            with patch.object(
+                mock_plot_info.prover, "get_qualities_for_challenge", side_effect=RuntimeError("test error")
+            ):
+                # should not raise exception, should handle error gracefully
+                await harvester_api.new_signage_point_harvester(new_challenge, mock_peer)

chia/_tests/plotting/test_prover.py

@@ -42,7 +42,7 @@ def test_v2_prover_get_qualities_for_challenge_raises_error(self) -> None:
 
     def test_v2_prover_get_full_proof_raises_error(self) -> None:
         prover = V2Prover("/nonexistent/path/test.plot2")
-        with pytest.raises(NotImplementedError, match="V2 plot format is not yet implemented"):
+        with pytest.raises(NotImplementedError, match="V2 plot format require solver to get full proof"):
             prover.get_full_proof(b"challenge", 0)
 
     def test_v2_prover_bytes_raises_error(self) -> None: