Shutdown test

.github/workflows/integrations_test.yml

@@ -48,6 +48,9 @@ jobs:
           - name: Testnet4 Bootstrap And Catchup
             test-path: integrations/testnet4_test.py 
 
+          - name: Testnet4 Start and Shutdown
+            test-path: integrations/shutdown_test.py
+
           # Other tests
 
           # run the docker-compose tests

counterparty-core/counterpartycore/lib/backend/bitcoind.py

@@ -13,6 +13,7 @@
 from requests.exceptions import ChunkedEncodingError, ConnectionError, ReadTimeout, Timeout
 
 from counterpartycore.lib import config, exceptions
+from counterpartycore.lib.ledger.currentstate import CurrentState
 from counterpartycore.lib.parser import deserialize, utxosinfo
 
 logger = logging.getLogger(config.LOGGER_NAME)
@@ -36,6 +37,8 @@ def clean_url_for_log(url):
 
 # for testing
 def should_retry():
+    if CurrentState().block_parser_status() == "Stopping":
+        return False
     return True
 
 

counterparty-core/counterpartycore/lib/backend/rsfetcher.py

@@ -63,7 +63,6 @@ def __init__(self, indexer_config=None):
                 self.config["log_level"] = config.LOG_LEVEL_STRING
         else:
             logger.warning("Using custom indexer config.")
-            print(indexer_config)
             self.config = indexer_config
         self.config["network"] = config.NETWORK_NAME
         self.fetcher = None

counterparty-core/counterpartycore/lib/cli/server.py

@@ -756,7 +756,7 @@ def run(self):
         blocks.reset_rust_fetcher_database()
 
         # catch up
-        blocks.catch_up(self.db)
+        blocks.catch_up(self.db, self.api_stop_event)
 
         # Blockchain watcher
         logger.info("Watching for new blocks...")
@@ -765,6 +765,7 @@ def run(self):
 
     def stop(self):
         logger.info("Shutting down...")
+        CurrentState().set_block_parser_status("Stopping")
 
         # Ensure all threads are stopped
         if self.follower_daemon:

counterparty-core/counterpartycore/lib/parser/blocks.py

@@ -890,6 +890,8 @@ def catch_up(db, check_asset_conservation=True):
         parsed_blocks = 0
 
         while CurrentState().current_block_index() < block_count:
+            if CurrentState().block_parser_status() == "Stopping":
+                return
             # Get block information and transactions
             fetch_time_start = time.time()
             if fetcher is None:

counterparty-core/counterpartycore/lib/parser/follow.py

@@ -296,9 +296,6 @@ async def handle(self):
                 break  # Exit the loop if the task is cancelled
             except Exception as e:
                 logger.error("Error in handle loop: %s", e)
-                import traceback
-
-                print(traceback.format_exc())  # for debugging
                 capture_exception(e)
                 self.stop()
                 break  # Optionally break the loop on other exceptions

counterparty-core/counterpartycore/test/integrations/http2https.py

@@ -38,7 +38,7 @@ def handle(self):
                 data=body,
                 headers={"Content-Type": "application/json"},
                 verify=False,
-                timeout=5,
+                timeout=15,
             )
 
             # Send response back to client

counterparty-core/counterpartycore/test/integrations/reparsetest.py

@@ -9,7 +9,7 @@
 from http2https import PROXY_PORT, start_http_proxy, stop_http_proxy
 
 # DATA_DIR = os.path.join(tempfile.gettempdir(), "counterparty-data")
-DATA_DIR = os.path.join(os.path.expanduser("~/.cache"), "counterparty-data")
+DATA_DIR = os.path.join(os.path.expanduser("~/.cache"), "counterparty-test-data")
 
 
 def prepare(network):

counterparty-core/counterpartycore/test/integrations/shutdown_test.py

@@ -1,49 +1,76 @@
-import os
 import random
-import signal
+import socket
 import time
 from io import StringIO
 
-import sh
+from counterpartycore.lib.cli import server
+from counterpartycore.lib.cli.main import arg_parser
 from counterpartycore.test.integrations import reparsetest
 from http2https import PROXY_PORT, start_http_proxy, stop_http_proxy
 
 
+def is_port_in_used(port):
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        s.bind(("127.0.0.1", port))
+        return False
+    except socket.error:
+        return True
+    finally:
+        s.close()
+
+
 def test_shutdown():
-    sh_counterparty_server, backend_url, db_file, api_url = reparsetest.prepare("mainnet")
+    sh_counterparty_server, backend_url, db_file, api_url = reparsetest.prepare("testnet4")
 
-    out = StringIO()
     try:
         start_http_proxy(backend_url)
-        server_process = sh_counterparty_server(
-            "start",
-            "--backend-connect",
-            "127.0.0.1",
-            "--backend-port",
-            PROXY_PORT,
-            "--wsgi-server",
-            "gunicorn",
-            _bg=True,
-            _out=out,
-            _err_to_out=True,
-            _bg_exc=False,
+
+        parser = arg_parser(no_config_file=True)
+        args = parser.parse_args(
+            [
+                "--testnet4",
+                "--data-dir",
+                reparsetest.DATA_DIR,
+                "--cache-dir",
+                reparsetest.DATA_DIR,
+                "start",
+                "--backend-connect",
+                "127.0.0.1",
+                "--backend-port",
+                f"{PROXY_PORT}",
+                "--wsgi-server",
+                "gunicorn",
+            ]
         )
 
-        duration = random.randint(1, 60)  # noqa S311
+        log_stream = StringIO()
+        server.initialise_log_and_config(args, log_stream=log_stream)
+
+        test_duration = random.randint(1, 60)  # noqa S311
         start_time = time.time()
 
-        print(f"Waiting random time: {duration}s")
-        while time.time() - start_time < duration:
-            time.sleep(1)
+        print("Test duration: ", test_duration)
 
-        print("Shutting down server...")
-        # server_process.terminate()
-        os.kill(server_process.pid, signal.SIGTERM)
+        counterparty_server = server.CounterpartyServer(args, log_stream)
+        counterparty_server.start()
+        while time.time() - start_time < test_duration:
+            counterparty_server.join(1)
+
+        assert is_port_in_used(44000)
 
-    except sh.SignalException_SIGTERM:
-        pass
     finally:
-        print("Waiting for server to shutdown...")
-        time.sleep(30)
+        print("Shutting down server...")
+        counterparty_server.stop()
         stop_http_proxy()
-        print(out.getvalue())
+
+    logs = log_stream.getvalue()
+
+    assert "Ledger.Main - Shutting down..." in logs
+    assert "Ledger.Main - Asset Conservation Checker thread stopped." in logs
+    assert "Ledger.BackendHeight - BackendHeight Thread stopped." in logs
+    assert "Ledger.Main - API Server v1 thread stopped." in logs
+    assert "Ledger.Main - API Server process stopped." in logs
+    assert "Ledger.Main - Shutdown complete." in logs
+
+    assert not is_port_in_used(44000)