P2P Interface scenario

src/warnet/server.py

@@ -354,10 +354,11 @@ def _start_scenario(
             t = threading.Thread(target=lambda: self.scenario_log(proc))
             t.daemon = True
             t.start()
+            cmd = f"{scenario_name} {' '.join(additional_args)}".strip()
             self.running_scenarios.append(
                 {
                     "pid": proc.pid,
-                    "cmd": f"{scenario_name} {' '.join(additional_args)}",
+                    "cmd": cmd,
                     "proc": proc,
                     "network": network,
                 }

test/data/scenario_p2p_interface.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+from collections import defaultdict
+from time import sleep
+
+from test_framework.messages import CInv, msg_getdata
+from test_framework.p2p import P2PInterface
+from warnet.test_framework_bridge import WarnetTestFramework
+
+
+def cli_help():
+    return "Run P2P GETDATA test"
+
+
+class P2PStoreBlock(P2PInterface):
+    def __init__(self):
+        super().__init__()
+        self.blocks = defaultdict(int)
+
+    def on_block(self, message):
+        message.block.calc_sha256()
+        self.blocks[message.block.sha256] += 1
+
+
+class GetdataTest(WarnetTestFramework):
+    def set_test_params(self):
+        self.num_nodes = 1
+
+    def run_test(self):
+        while not self.warnet.network_connected():
+            self.log.info("Waiting for complete network connection...")
+            sleep(5)
+        self.log.info("Network connected")
+
+        self.log.info("Adding the p2p connection")
+
+        p2p_block_store = self.nodes[0].add_p2p_connection(
+            P2PStoreBlock(), dstaddr=self.warnet.tanks[0].ipv4, dstport=18444
+        )
+
+        self.log.info("test that an invalid GETDATA doesn't prevent processing of future messages")
+
+        # Send invalid message and verify that node responds to later ping
+        invalid_getdata = msg_getdata()
+        invalid_getdata.inv.append(CInv(t=0, h=0))  # INV type 0 is invalid.
+        p2p_block_store.send_and_ping(invalid_getdata)
+
+        # Check getdata still works by fetching tip block
+        best_block = int(self.nodes[0].getbestblockhash(), 16)
+        good_getdata = msg_getdata()
+        good_getdata.inv.append(CInv(t=2, h=best_block))
+        p2p_block_store.send_and_ping(good_getdata)
+        p2p_block_store.wait_until(lambda: p2p_block_store.blocks[best_block] == 1)
+
+
+if __name__ == "__main__":
+    GetdataTest().main()

test/scenarios_test.py

@@ -26,8 +26,9 @@ def setup_network(self):
 
     def test_scenarios(self):
         self.check_available_scenarios()
-        self.run_and_check_scenario("miner_std")
-        self.run_and_check_scenario_from_file("src/warnet/scenarios/miner_std.py")
+        self.run_and_check_miner_scenario("miner_std")
+        self.run_and_check_miner_scenario_from_file("src/warnet/scenarios/miner_std.py")
+        self.run_and_check_scenario_from_file("test/data/scenario_p2p_interface.py")
 
     def check_available_scenarios(self):
         self.log.info("Checking available scenarios")
@@ -42,22 +43,46 @@ def scenario_running(self, scenario_name: str):
         running = scenario_name in active[0]["cmd"]
         return running and len(active) == 1
 
-    def run_and_check_scenario(self, scenario_name):
+    def run_and_check_scenario_from_file(self, scenario_file):
+        scenario_name = self.get_scenario_name_from_path(scenario_file)
+
+        def check_scenario_clean_exit():
+            running = self.rpc("scenarios_list_running")
+            scenarios = [s for s in running if s["cmd"].strip() == scenario_name]
+            if not scenarios:
+                return False
+            scenario = scenarios[0]
+            if scenario["active"]:
+                return False
+            if scenario["return_code"] != 0:
+                raise Exception(
+                    f"Scenario {scenario_name} failed with return code {scenario['return_code']}"
+                )
+            return True
+
+        self.log.info(f"Running scenario: {scenario_name}")
+        self.warcli(f"scenarios run-file {scenario_file}")
+        self.wait_for_predicate(lambda: check_scenario_clean_exit())
+
+    def run_and_check_miner_scenario(self, scenario_name):
         self.log.info(f"Running scenario: {scenario_name}")
         self.warcli(f"scenarios run {scenario_name} --allnodes --interval=1")
         self.wait_for_predicate(lambda: self.scenario_running(scenario_name))
         self.wait_for_predicate(lambda: self.check_blocks(30))
         self.stop_scenario()
 
-    def run_and_check_scenario_from_file(self, scenario_file):
+    def run_and_check_miner_scenario_from_file(self, scenario_file):
         self.log.info(f"Running scenario from file: {scenario_file}")
         self.warcli(f"scenarios run-file {scenario_file} --allnodes --interval=1")
         start = int(self.warcli("bitcoin rpc 0 getblockcount"))
-        scenario_name = os.path.splitext(os.path.basename(scenario_file))[0]
+        scenario_name = self.get_scenario_name_from_path(scenario_file)
         self.wait_for_predicate(lambda: self.scenario_running(scenario_name))
         self.wait_for_predicate(lambda: self.check_blocks(2, start=start))
         self.stop_scenario()
 
+    def get_scenario_name_from_path(self, scenario_file):
+        return os.path.splitext(os.path.basename(scenario_file))[0]
+
     def check_blocks(self, target_blocks, start: int = 0):
         running = self.rpc("scenarios_list_running")
         assert len(running) == 1, f"Expected one running scenario, got {len(running)}"