diff --git a/.github/workflows/db_sync_full_sync.yaml b/.github/workflows/db_sync_full_sync.yaml index b614645b..c928b816 100644 --- a/.github/workflows/db_sync_full_sync.yaml +++ b/.github/workflows/db_sync_full_sync.yaml @@ -23,10 +23,6 @@ on: - preprod - preview default: preprod - run_only_sync_test: - type: boolean - default: true - description: "If checked only full sync test will be run otherwise local snapshot creation and restoration tests will be started after sync test is completed" configs_base_url: description: "Base URL for Cardano environment configs" required: false @@ -93,25 +89,15 @@ jobs: HB_PID=$! trap 'pkill -P "$HB_PID" 2>/dev/null || true; kill "$HB_PID" 2>/dev/null || true' EXIT - if [ "${{ inputs.run_only_sync_test }}" = "true" ]; then - test_marker="node_sync or db_sync" - else - test_marker="node_sync or db_sync or snapshot_creation or local_snapshot" - fi - pytest_args=( sync_tests/tests/ - -m "${test_marker}" + -m "node_sync or db_sync" --environment "${{ inputs.environment }}" --node-revision "${{ inputs.node_version }}" --db-sync-revision "${{ inputs.db_sync_version }}" --db-sync-start-options "${DB_SYNC_START_OPTIONS}" ) - if [ "${{ inputs.run_only_sync_test }}" = "true" ]; then - pytest_args+=(--run-only-sync-test) - fi - pytest_exit=0 { echo "=== ci step started $(date -u +%FT%TZ) ===" diff --git a/pyproject.toml b/pyproject.toml index cc2c6fc3..feb78ecc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,9 +63,6 @@ log_level = "INFO" markers = [ "node_sync: marks tests that require a fully synced cardano-node", "db_sync: marks tests that require a fully synced cardano-db-sync", - "snapshot_creation: marks snapshot creation tests (run after full sync)", - "local_snapshot: marks local snapshot restoration tests", - "iohk_snapshot: marks IOHK official snapshot restoration tests", "mainnet_tx_count: marks mainnet transaction count tests", ] diff --git a/sync_tests/tests/conftest.py b/sync_tests/tests/conftest.py index 86fdf98b..0280b07e 100644 --- a/sync_tests/tests/conftest.py +++ b/sync_tests/tests/conftest.py @@ -123,18 +123,6 @@ def pytest_addoption(parser: tp.Any) -> None: default=None, help="Working directory for logs and artifacts", ) - parser.addoption( - "--snapshot-url", - action="store", - default=None, - help="Snapshot download URL for IOHK snapshot restoration", - ) - parser.addoption( - "--run-only-sync-test", - action="store_true", - default=False, - help="Skip snapshot creation/restoration steps", - ) parser.addoption( "--pg-port", action="store", diff --git a/sync_tests/tests/test_iohk_snapshot_restoration.py b/sync_tests/tests/test_iohk_snapshot_restoration.py deleted file mode 100644 index 59b5236b..00000000 --- a/sync_tests/tests/test_iohk_snapshot_restoration.py +++ /dev/null @@ -1,319 +0,0 @@ -"""Tests for db-sync restoration from an IOHK official snapshot.""" - -from __future__ import annotations - -import datetime -import logging -import os -import pathlib as pl -import time -import typing as tp -from collections import OrderedDict - -import pytest -from _pytest.fixtures import FixtureRequest - -from sync_tests.tests.conftest import SyncContext -from sync_tests.utils import db_sync -from sync_tests.utils import helpers -from sync_tests.utils import sync_entries -from sync_tests.utils.external import gitpython -from sync_tests.utils.logs import log_analyzer - -LOGGER = logging.getLogger(__name__) - -pytestmark = pytest.mark.iohk_snapshot - -POST_SYNC_WAIT_MINUTES = 30 - - -@pytest.fixture(scope="session") -def iohk_restoration_result( - request: FixtureRequest, - sync_context: SyncContext, -) -> tp.Generator[dict[str, tp.Any], None, None]: - """Download IOHK snapshot, restore, sync node + db-sync, yield results. - - Args: - request: Pytest fixture request for CLI options. - sync_context: Shared session context. - - Yields: - Dict with restoration timing, tip data, versions, and sync results. - """ - env = sync_context.env - node_revision: str | None = request.config.getoption("--node-revision") - db_sync_revision: str | None = request.config.getoption( - "--db-sync-revision", - ) - snapshot_url_opt: str | None = request.config.getoption("--snapshot-url") - - if not node_revision: - pytest.skip( - "--node-revision required for IOHK snapshot restoration", - ) - if not db_sync_revision: - pytest.skip( - "--db-sync-revision required for IOHK snapshot restoration", - ) - - config = db_sync.create_db_sync_config( - env=env, - workdir=sync_context.workdir, - ) - - platform_system, platform_release, platform_version = helpers.get_os_type() - start_test_time = datetime.datetime.now( - tz=datetime.timezone.utc, - ).strftime("%d/%m/%Y %H:%M:%S") - - snapshot_url = snapshot_url_opt or db_sync.get_latest_snapshot_url(env, None) - LOGGER.info("Snapshot URL: %s", snapshot_url) - - # node setup (reuses conftest's run_node_sync instead of hand-rolling the same - # build+start+wait-for-sync sequence a third time; clean_start=False preserves - # this test's original behavior of reusing an already-synced node db if one is - # already present from an earlier step in the same session) - base_dir = pl.Path.cwd() - node_run_result = sync_entries.run_node_sync( - env=env, - node_revision=node_revision, - node_logfile_path=config.node_log_file, - base_dir=base_dir, - conf_dir=base_dir, - start_era="shelley", - clean_start=False, - full_sync=True, - ) - cli_version = node_run_result.cli_version - cli_git_rev = node_run_result.cli_git_rev - node_sync_time_seconds = node_run_result.sync_time_sec - - # db-sync setup - LOGGER.info("Building db-sync from revision: %s", db_sync_revision) - db_sync_dir = gitpython.clone_repo( - "cardano-db-sync", - db_sync_revision, - ) - db_sync.setup_postgres(config) - db_sync.create_pgpass_file(config) - db_sync.create_database() - db_sync.list_databases(config) - helpers.execute_command( - "nix build -v --accept-flake-config --print-build-logs .#cardano-db-sync -o db-sync-node", - cwd=db_sync_dir, - ) - helpers.execute_command( - "nix build -v --accept-flake-config --print-build-logs .#cardano-db-tool -o db-sync-tool", - cwd=db_sync_dir, - ) - db_sync.copy_db_sync_executables(config, build_method="nix") - - # download and verify snapshot - LOGGER.info("Downloading snapshot") - snapshot_name = db_sync.download_db_sync_snapshot(snapshot_url) - expected_sha = db_sync.get_snapshot_sha_256_sum(snapshot_url) - actual_sha = helpers.get_file_sha256_sum(snapshot_name) - assert expected_sha == actual_sha, "Snapshot SHA-256 mismatch" - - # restore snapshot - LOGGER.info("Restoring from IOHK snapshot") - restoration_time = db_sync.restore_db_sync_from_snapshot( - config, - snapshot_name, - remove_ledger_dir="no", - ) - LOGGER.info("Restoration time: %d seconds", restoration_time) - - db_sync_tip = db_sync.get_db_sync_tip(config) - if db_sync_tip is None: - msg = "db-sync tip unavailable after snapshot restoration" - raise RuntimeError(msg) - snapshot_epoch_no = db_sync_tip.epoch_no - snapshot_block_no = db_sync_tip.block_no - snapshot_slot_no = db_sync_tip.slot_no - - # start db-sync - LOGGER.info("Starting db-sync after IOHK snapshot restoration") - db_sync.start_db_sync(config, start_args="", first_start="True") - helpers.print_last_n_lines(config.db_sync_log_file, 30) - db_sync_version, db_sync_git_rev = db_sync.get_db_sync_version(config) - db_full_sync_time_in_secs, perf_stats = db_sync.wait_for_db_to_sync( - config, - ) - - end_test_time = datetime.datetime.now( - tz=datetime.timezone.utc, - ).strftime("%d/%m/%Y %H:%M:%S") - - LOGGER.info( - "Waiting %d minutes for additional syncing", - POST_SYNC_WAIT_MINUTES, - ) - time.sleep(POST_SYNC_WAIT_MINUTES * 60) - helpers.print_last_n_lines(config.db_sync_log_file, 60) - - db_sync_tip = db_sync.get_db_sync_tip(config) - if db_sync_tip is None: - msg = "db-sync tip unavailable after post-restore sync wait" - raise RuntimeError(msg) - - last_perf = db_sync.get_last_perf_stats_point(perf_stats) - - result: dict[str, tp.Any] = { - "platform_system": platform_system, - "platform_release": platform_release, - "platform_version": platform_version, - "env": env, - "cli_version": cli_version, - "cli_git_rev": cli_git_rev, - "db_sync_version": db_sync_version, - "db_sync_git_rev": db_sync_git_rev, - "start_test_time": start_test_time, - "end_test_time": end_test_time, - "node_sync_time_seconds": node_sync_time_seconds, - "db_full_sync_time_in_secs": db_full_sync_time_in_secs, - "snapshot_url": snapshot_url, - "snapshot_name": snapshot_name, - "restoration_time": restoration_time, - "snapshot_epoch_no": snapshot_epoch_no, - "snapshot_block_no": snapshot_block_no, - "snapshot_slot_no": snapshot_slot_no, - "last_synced_epoch_no": db_sync_tip.epoch_no, - "last_synced_block_no": db_sync_tip.block_no, - "last_synced_slot_no": db_sync_tip.slot_no, - "cpu_percent_usage": last_perf.cpu_percent_usage, - "rss_mem_usage": last_perf.rss_mem_usage, - "perf_stats": perf_stats, - "config": config, - } - - yield result - - LOGGER.info("Teardown: terminating cardano services") - sync_entries.teardown_node_and_db_sync(base_dir=base_dir, config=config) - - -class TestIohkSnapshotRestoration: - """Validate db-sync restoration from an IOHK official snapshot.""" - - def test_restoration_completed( - self, - iohk_restoration_result: dict[str, tp.Any], - ): - """Check that the IOHK snapshot was restored and db-sync synced.""" - assert iohk_restoration_result["restoration_time"] > 0 - assert iohk_restoration_result["db_full_sync_time_in_secs"] >= 0 - assert ( - iohk_restoration_result["last_synced_epoch_no"] - >= (iohk_restoration_result["snapshot_epoch_no"]) - ) - - def test_restoration_results_json( - self, - sync_context: SyncContext, # noqa: ARG002 - iohk_restoration_result: dict[str, tp.Any], - ): - """Generate and upload IOHK snapshot restoration test results.""" - config = iohk_restoration_result["config"] - test_results_file = config.workdir / "db_sync_iohk_snapshot_restoration_test_results.json" - - test_data: OrderedDict[str, tp.Any] = OrderedDict() - test_data["platform_system"] = iohk_restoration_result["platform_system"] - test_data["platform_release"] = iohk_restoration_result["platform_release"] - test_data["platform_version"] = iohk_restoration_result["platform_version"] - test_data["no_of_cpu_cores"] = os.cpu_count() - test_data["total_ram_in_GB"] = helpers.get_total_ram_in_gb() - test_data["env"] = iohk_restoration_result["env"] - test_data["node_cli_version"] = iohk_restoration_result["cli_version"] - test_data["node_git_revision"] = iohk_restoration_result["cli_git_rev"] - test_data["db_sync_version"] = iohk_restoration_result["db_sync_version"] - test_data["db_sync_git_rev"] = iohk_restoration_result["db_sync_git_rev"] - test_data["start_test_time"] = iohk_restoration_result["start_test_time"] - test_data["end_test_time"] = iohk_restoration_result["end_test_time"] - test_data["node_total_sync_time_in_sec"] = iohk_restoration_result["node_sync_time_seconds"] - test_data["node_total_sync_time_in_h_m_s"] = str( - datetime.timedelta( - seconds=int(iohk_restoration_result["node_sync_time_seconds"]), - ) - ) - test_data["db_total_sync_time_in_sec"] = iohk_restoration_result[ - "db_full_sync_time_in_secs" - ] - test_data["db_total_sync_time_in_h_m_s"] = str( - datetime.timedelta( - seconds=iohk_restoration_result["db_full_sync_time_in_secs"], - ) - ) - test_data["snapshot_url"] = iohk_restoration_result["snapshot_url"] - test_data["snapshot_name"] = iohk_restoration_result["snapshot_name"] - test_data["snapshot_epoch_no"] = iohk_restoration_result["snapshot_epoch_no"] - test_data["snapshot_block_no"] = iohk_restoration_result["snapshot_block_no"] - test_data["snapshot_slot_no"] = iohk_restoration_result["snapshot_slot_no"] - test_data["last_synced_epoch_no"] = iohk_restoration_result["last_synced_epoch_no"] - test_data["last_synced_block_no"] = iohk_restoration_result["last_synced_block_no"] - test_data["last_synced_slot_no"] = iohk_restoration_result["last_synced_slot_no"] - test_data["cpu_percent_usage"] = iohk_restoration_result["cpu_percent_usage"] - test_data["total_rss_memory_usage_in_B"] = iohk_restoration_result["rss_mem_usage"] - test_data["total_database_size"] = db_sync.get_total_db_size( - config, - ) - test_data["rollbacks"] = log_analyzer.are_rollbacks_present_in_logs( - log_file=config.db_sync_log_file, - ) - test_data["errors"] = log_analyzer.is_string_present_in_file( - file_to_check=config.db_sync_log_file, - search_string="db-sync-node:Error", - ) - - helpers.write_json_to_file(test_results_file, test_data) - - # write enriched perf stats - era_activation = db_sync.get_era_activation_data(config) - enriched_perf_stats = db_sync.enrich_perf_stats_with_era( - iohk_restoration_result["perf_stats"], - era_activation, - ) - helpers.write_json_to_file( - config.perf_stats_file, - enriched_perf_stats, - ) - db_sync.export_epoch_sync_times_from_db( - config, - config.epoch_sync_times_file, - iohk_restoration_result["snapshot_epoch_no"], - ) - - # compress and upload artifacts - helpers.zip_file( - config.node_archive_name, - config.node_log_file, - ) - helpers.zip_file( - config.db_sync_archive_name, - config.db_sync_log_file, - ) - helpers.zip_file( - config.sync_data_archive_name, - config.epoch_sync_times_file, - ) - helpers.zip_file( - config.perf_stats_archive_name, - config.perf_stats_file, - ) - - db_sync.upload_artifact(config.node_archive_name) - db_sync.upload_artifact(config.db_sync_archive_name) - db_sync.upload_artifact(config.sync_data_archive_name) - db_sync.upload_artifact(config.perf_stats_archive_name) - db_sync.upload_artifact(str(test_results_file)) - - assert test_results_file.exists() - - def test_restoration_log_analysis( - self, - iohk_restoration_result: dict[str, tp.Any], - ): - """Check db-sync logs for errors after IOHK snapshot restoration.""" - config = iohk_restoration_result["config"] - log_analyzer.check_db_sync_logs(log_file=config.db_sync_log_file) diff --git a/sync_tests/tests/test_local_snapshot_restoration.py b/sync_tests/tests/test_local_snapshot_restoration.py deleted file mode 100644 index ad03f086..00000000 --- a/sync_tests/tests/test_local_snapshot_restoration.py +++ /dev/null @@ -1,266 +0,0 @@ -"""Tests for db-sync restoration from a locally created snapshot.""" - -from __future__ import annotations - -import datetime -import json -import logging -import os -import pathlib as pl -import time -import typing as tp -from collections import OrderedDict - -import pytest -from _pytest.fixtures import FixtureRequest - -from sync_tests.tests.conftest import SyncContext -from sync_tests.utils import db_sync -from sync_tests.utils import helpers -from sync_tests.utils import sync_entries -from sync_tests.utils.logs import log_analyzer - -LOGGER = logging.getLogger(__name__) - -pytestmark = pytest.mark.local_snapshot - -POST_SYNC_WAIT_MINUTES = 20 - - -@pytest.fixture(scope="session") -def local_restoration_result( - request: FixtureRequest, - sync_context: SyncContext, -) -> tp.Generator[dict[str, tp.Any], None, None]: - """Restore db-sync from a local snapshot, sync, and yield result data. - - Reads snapshot metadata written by a prior snapshot creation run. Run the - snapshot_creation marked tests first to produce that metadata. - - Args: - request: Pytest fixture request for CLI options. - sync_context: Shared session context. - - Yields: - Dict with restoration timing, tip data, and sync results. - """ - if request.config.getoption("--run-only-sync-test"): - pytest.skip("--run-only-sync-test: skipping snapshot restoration") - - env = sync_context.env - node_revision: str | None = request.config.getoption("--node-revision") - db_sync_revision: str | None = request.config.getoption( - "--db-sync-revision", - ) - pg_port: str = request.config.getoption("--pg-port") or "5433" - - if not node_revision: - pytest.skip("--node-revision required for local snapshot restoration") - if not db_sync_revision: - pytest.skip( - "--db-sync-revision required for local snapshot restoration", - ) - - config = db_sync.create_db_sync_config( - env=env, - workdir=sync_context.workdir, - pg_port=pg_port, - ) - - platform_system, platform_release, platform_version = helpers.get_os_type() - start_test_time = datetime.datetime.now( - tz=datetime.timezone.utc, - ).strftime("%d/%m/%Y %H:%M:%S") - - # database setup - LOGGER.info("Local snapshot restoration: postgres and database setup") - db_sync.setup_postgres(config, pg_port=pg_port) - db_sync.create_pgpass_file(config) - db_sync.create_database() - - # restore snapshot - snapshot_state_file = sync_context.workdir / "sync_session_state.json" - if not snapshot_state_file.exists(): - msg = f"Snapshot metadata file not found: {snapshot_state_file}" - raise FileNotFoundError(msg) - try: - with open(snapshot_state_file, encoding="utf-8") as state_fh: - snapshot_data = json.load(state_fh) - snapshot_file = snapshot_data["snapshot_file"] - except (json.JSONDecodeError, KeyError) as exc: - msg = f"Invalid snapshot metadata in {snapshot_state_file}: {exc}" - raise ValueError(msg) from exc - LOGGER.info("Restoring from snapshot: %s", snapshot_file) - restoration_time = db_sync.restore_db_sync_from_snapshot( - config, - snapshot_file, - ) - LOGGER.info("Restoration time: %d seconds", restoration_time) - - db_sync_tip = db_sync.get_db_sync_tip(config) - if db_sync_tip is None: - msg = "db-sync tip unavailable after snapshot restoration" - raise RuntimeError(msg) - snapshot_epoch_no = db_sync_tip.epoch_no - snapshot_block_no = db_sync_tip.block_no - snapshot_slot_no = db_sync_tip.slot_no - LOGGER.info( - "Tip after restoration: epoch=%s block=%s slot=%s", - snapshot_epoch_no, - snapshot_block_no, - snapshot_slot_no, - ) - - # start node (reuses conftest's run_node_sync instead of hand-rolling the same - # build+start+wait-for-sync sequence a third time; clean_start=False preserves - # this test's original behavior of reusing an already-synced node db if one is - # already present from an earlier step in the same session) - LOGGER.info("Starting node for post-restoration sync") - base_dir = pl.Path.cwd() - sync_entries.run_node_sync( - env=env, - node_revision=node_revision, - node_logfile_path=config.node_log_file, - base_dir=base_dir, - conf_dir=base_dir, - start_era="shelley", - clean_start=False, - full_sync=True, - ) - - # start db-sync - LOGGER.info("Starting db-sync after snapshot restoration") - helpers.export_env_var("PGPORT", pg_port) - db_sync.start_db_sync(config, start_args="", first_start="False") - helpers.print_last_n_lines(config.db_sync_log_file, 20) - time.sleep(60) - db_sync_version, db_sync_git_rev = db_sync.get_db_sync_version(config) - db_full_sync_time_in_secs, _perf_stats = db_sync.wait_for_db_to_sync( - config, - ) - - end_test_time = datetime.datetime.now( - tz=datetime.timezone.utc, - ).strftime("%d/%m/%Y %H:%M:%S") - - LOGGER.info( - "Waiting %d minutes for additional syncing", - POST_SYNC_WAIT_MINUTES, - ) - time.sleep(POST_SYNC_WAIT_MINUTES * 60) - - db_sync_tip = db_sync.get_db_sync_tip(config) - if db_sync_tip is None: - msg = "db-sync tip unavailable after post-restore sync wait" - raise RuntimeError(msg) - - result: dict[str, tp.Any] = { - "platform_system": platform_system, - "platform_release": platform_release, - "platform_version": platform_version, - "env": env, - "db_sync_version": db_sync_version, - "db_sync_git_rev": db_sync_git_rev, - "start_test_time": start_test_time, - "end_test_time": end_test_time, - "db_full_sync_time_in_secs": db_full_sync_time_in_secs, - "snapshot_file": snapshot_file, - "restoration_time": restoration_time, - "snapshot_epoch_no": snapshot_epoch_no, - "snapshot_block_no": snapshot_block_no, - "snapshot_slot_no": snapshot_slot_no, - "last_synced_epoch_no": db_sync_tip.epoch_no, - "last_synced_block_no": db_sync_tip.block_no, - "last_synced_slot_no": db_sync_tip.slot_no, - "config": config, - } - - yield result - - LOGGER.info("Teardown: terminating cardano services") - sync_entries.teardown_node_and_db_sync(base_dir=base_dir, config=config) - - -class TestLocalSnapshotRestoration: - """Validate db-sync restoration from a locally created snapshot.""" - - def test_restoration_completed( - self, - local_restoration_result: dict[str, tp.Any], - ): - """Check that the snapshot was restored and db-sync re-synced.""" - assert local_restoration_result["restoration_time"] > 0 - assert local_restoration_result["db_full_sync_time_in_secs"] >= 0 - assert ( - local_restoration_result["last_synced_epoch_no"] - >= (local_restoration_result["snapshot_epoch_no"]) - ) - - def test_restoration_results_json( - self, - sync_context: SyncContext, - local_restoration_result: dict[str, tp.Any], - ): - """Generate and upload restoration test results JSON.""" - config = local_restoration_result["config"] - test_results_file = ( - config.workdir / f"db_sync_{sync_context.env}" - f"_local_snapshot_restoration_test_results.json" - ) - - test_data: OrderedDict[str, tp.Any] = OrderedDict() - test_data["platform_system"] = local_restoration_result["platform_system"] - test_data["platform_release"] = local_restoration_result["platform_release"] - test_data["platform_version"] = local_restoration_result["platform_version"] - test_data["no_of_cpu_cores"] = os.cpu_count() - test_data["total_ram_in_GB"] = helpers.get_total_ram_in_gb() - test_data["env"] = local_restoration_result["env"] - test_data["db_sync_version"] = local_restoration_result["db_sync_version"] - test_data["db_sync_git_rev"] = local_restoration_result["db_sync_git_rev"] - test_data["start_test_time"] = local_restoration_result["start_test_time"] - test_data["end_test_time"] = local_restoration_result["end_test_time"] - test_data["db_total_sync_time_in_sec"] = local_restoration_result[ - "db_full_sync_time_in_secs" - ] - test_data["db_total_sync_time_in_h_m_s"] = str( - datetime.timedelta( - seconds=int(local_restoration_result["db_full_sync_time_in_secs"]), - ) - ) - test_data["snapshot_name"] = local_restoration_result["snapshot_file"] - test_data["snapshot_size_in_mb"] = db_sync.get_file_size( - local_restoration_result["snapshot_file"], - ) - test_data["restoration_time"] = local_restoration_result["restoration_time"] - test_data["snapshot_epoch_no"] = local_restoration_result["snapshot_epoch_no"] - test_data["snapshot_block_no"] = local_restoration_result["snapshot_block_no"] - test_data["snapshot_slot_no"] = local_restoration_result["snapshot_slot_no"] - test_data["last_synced_epoch_no"] = local_restoration_result["last_synced_epoch_no"] - test_data["last_synced_block_no"] = local_restoration_result["last_synced_block_no"] - test_data["last_synced_slot_no"] = local_restoration_result["last_synced_slot_no"] - test_data["total_database_size"] = db_sync.get_total_db_size(config) - test_data["rollbacks"] = log_analyzer.is_string_present_in_file( - file_to_check=config.db_sync_log_file, - search_string="rolling back to", - ) - test_data["errors"] = log_analyzer.is_string_present_in_file( - file_to_check=config.db_sync_log_file, - search_string="db-sync-node:Error", - ) - - helpers.write_json_to_file(test_results_file, test_data) - - archive_name = f"cardano_db_sync_{sync_context.env}_restoration.zip" - helpers.zip_file(archive_name, config.db_sync_log_file) - db_sync.upload_artifact(archive_name) - db_sync.upload_artifact(str(test_results_file)) - - assert test_results_file.exists() - - def test_restoration_log_analysis( - self, - local_restoration_result: dict[str, tp.Any], - ): - """Check db-sync logs for errors after snapshot restoration.""" - config = local_restoration_result["config"] - log_analyzer.check_db_sync_logs(log_file=config.db_sync_log_file) diff --git a/sync_tests/tests/test_snapshot_creation.py b/sync_tests/tests/test_snapshot_creation.py deleted file mode 100644 index d6424a0e..00000000 --- a/sync_tests/tests/test_snapshot_creation.py +++ /dev/null @@ -1,124 +0,0 @@ -"""Tests for db-sync snapshot creation.""" - -from __future__ import annotations - -import datetime -import logging -import os -import typing as tp -from collections import OrderedDict - -import pytest -from _pytest.fixtures import FixtureRequest - -from sync_tests.tests.conftest import DbSyncResult -from sync_tests.tests.conftest import SyncContext -from sync_tests.utils import db_sync -from sync_tests.utils import helpers - -LOGGER = logging.getLogger(__name__) - -pytestmark = pytest.mark.snapshot_creation - - -@pytest.fixture(scope="session") -def snapshot_created( - request: FixtureRequest, - sync_context: SyncContext, - db_sync_synced: DbSyncResult, # noqa: ARG001 -) -> dict[str, tp.Any]: - """Create a db-sync snapshot and return creation metadata.""" - if request.config.getoption("--run-only-sync-test"): - pytest.skip("--run-only-sync-test: skipping snapshot creation") - - env = sync_context.env - config = db_sync.create_db_sync_config( - env=env, - workdir=sync_context.workdir, - ) - - LOGGER.info("Starting snapshot creation") - start_time = datetime.datetime.now(tz=datetime.timezone.utc) - stage_2_cmd = db_sync.create_db_sync_snapshot_stage_1(config) - LOGGER.info("Stage 2 command: %s", stage_2_cmd) - stage_2_result = db_sync.create_db_sync_snapshot_stage_2( - config, - stage_2_cmd, - ) - LOGGER.info("Stage 2 result: %s", stage_2_result) - end_time = datetime.datetime.now(tz=datetime.timezone.utc) - - snapshot_file = stage_2_result - - creation_secs = int((end_time - start_time).total_seconds()) - LOGGER.info("Snapshot creation time: %d seconds", creation_secs) - - snapshot_data = { - "snapshot_file": snapshot_file, - "stage_2_cmd": stage_2_cmd, - "stage_2_result": stage_2_result, - "creation_time_secs": creation_secs, - "start_time": start_time.strftime("%d/%m/%Y %H:%M:%S"), - "end_time": end_time.strftime("%d/%m/%Y %H:%M:%S"), - } - helpers.write_json_to_file(sync_context.workdir / "sync_session_state.json", snapshot_data) - return snapshot_data - - -class TestSnapshotCreation: - """Validate db-sync snapshot creation.""" - - def test_snapshot_created( - self, - snapshot_created: dict[str, tp.Any], - ): - """Check that the snapshot file was created successfully.""" - assert snapshot_created["snapshot_file"], "Snapshot file name is empty" - assert snapshot_created["creation_time_secs"] > 0 - - def test_snapshot_results_json( - self, - sync_context: SyncContext, - db_sync_synced: DbSyncResult, - snapshot_created: dict[str, tp.Any], - ): - """Generate and upload snapshot creation test results.""" - env = sync_context.env - config = db_sync.create_db_sync_config( - env=env, - workdir=sync_context.workdir, - ) - test_results_file = config.workdir / f"snapshot_creation_{env}_test_results.json" - - platform_system, platform_release, platform_version = helpers.get_os_type() - - test_data: OrderedDict[str, tp.Any] = OrderedDict() - test_data["platform_system"] = platform_system - test_data["platform_release"] = platform_release - test_data["platform_version"] = platform_version - test_data["no_of_cpu_cores"] = os.cpu_count() - test_data["total_ram_in_GB"] = helpers.get_total_ram_in_gb() - test_data["env"] = env - test_data["db_sync_version"] = db_sync_synced.db_sync_version - test_data["db_sync_git_rev"] = db_sync_synced.db_sync_git_rev - test_data["start_test_time"] = snapshot_created["start_time"] - test_data["end_test_time"] = snapshot_created["end_time"] - test_data["snapshot_creation_time_in_sec"] = snapshot_created["creation_time_secs"] - test_data["snapshot_creation_time_in_h_m_s"] = str( - datetime.timedelta( - seconds=snapshot_created["creation_time_secs"], - ) - ) - test_data["snapshot_size_in_mb"] = db_sync.get_file_size( - snapshot_created["snapshot_file"], - ) - test_data["stage_2_cmd"] = snapshot_created["stage_2_cmd"] - test_data["stage_2_result"] = snapshot_created["stage_2_result"] - - helpers.write_json_to_file(test_results_file, test_data) - db_sync.upload_artifact(str(test_results_file)) - - if env != "mainnet": - db_sync.upload_artifact(snapshot_created["snapshot_file"]) - - assert test_results_file.exists() diff --git a/sync_tests/utils/db_sync/__init__.py b/sync_tests/utils/db_sync/__init__.py index 72bf0775..ed524c37 100755 --- a/sync_tests/utils/db_sync/__init__.py +++ b/sync_tests/utils/db_sync/__init__.py @@ -11,7 +11,6 @@ import subprocess import sys import time -import typing as tp from datetime import timedelta import psutil @@ -22,7 +21,6 @@ from sync_tests.utils.db_sync import config as db_sync_config from sync_tests.utils.db_sync import data as db_sync_data from sync_tests.utils.db_sync import postgres -from sync_tests.utils.db_sync import snapshots from sync_tests.utils.db_sync.config import DbSyncConfig from sync_tests.utils.db_sync.config import DbSyncTip from sync_tests.utils.db_sync.config import PerfStats @@ -195,7 +193,7 @@ def get_db_sync_version(_config: DbSyncConfig) -> tuple[str, str]: db_sync_dir = get_db_sync_dir() # Use the nix-built binary directly; avoids depending on the _cardano-db-sync # convenience copy which may not exist when copy_db_sync_executables was skipped - # (e.g. snapshot restoration tests) or failed due to a PermissionError. + # or failed due to a PermissionError. db_sync_binary = db_sync_dir / "db-sync-node" / "bin" / "cardano-db-sync" try: cmd = [str(db_sync_binary), "--version"] @@ -674,38 +672,6 @@ def create_node_database_archive(config: DbSyncConfig) -> pl.Path: return artifacts.create_node_database_archive(config) -def get_latest_snapshot_url(env: str, args: tp.Any) -> str: - """Retrieve the latest snapshot URL for the specified environment.""" - return snapshots.get_latest_snapshot_url(env, args) - - -def download_db_sync_snapshot(snapshot_url: str) -> str: - """Download a db-sync snapshot from the specified URL.""" - return snapshots.download_db_sync_snapshot(snapshot_url) - - -def get_snapshot_sha_256_sum(snapshot_url: str) -> str | None: - """Retrieve the expected sha256 checksum for the specified snapshot.""" - return snapshots.get_snapshot_sha_256_sum(snapshot_url) - - -def restore_db_sync_from_snapshot( - config: DbSyncConfig, snapshot_file: str | pl.Path, remove_ledger_dir: str = "yes" -) -> float: - """Restore db-sync from a snapshot file.""" - return snapshots.restore_db_sync_from_snapshot(config, snapshot_file, remove_ledger_dir) - - -def create_db_sync_snapshot_stage_1(config: DbSyncConfig) -> str: - """Create a db-sync snapshot (stage 1) and return the snapshot file path.""" - return snapshots.create_db_sync_snapshot_stage_1(config) - - -def create_db_sync_snapshot_stage_2(config: DbSyncConfig, stage_2_cmd: str) -> str: - """Create a db-sync snapshot (stage 2) and return the snapshot file path.""" - return snapshots.create_db_sync_snapshot_stage_2(config, stage_2_cmd) - - def start_monitor(workdir: pl.Path, env: str) -> None: """Start the background resource monitor for this workspace. diff --git a/sync_tests/utils/db_sync/snapshots.py b/sync_tests/utils/db_sync/snapshots.py deleted file mode 100644 index f6c3414d..00000000 --- a/sync_tests/utils/db_sync/snapshots.py +++ /dev/null @@ -1,282 +0,0 @@ -"""Download, upload, and restore Cardano DB Sync snapshots.""" - -from __future__ import annotations - -import logging -import os -import pathlib as pl -import shutil -import subprocess -import tarfile -import time -import typing as tp - -import requests -import xmltodict - -from sync_tests.utils import helpers -from sync_tests.utils.db_sync.config import DbSyncConfig -from sync_tests.utils.path_utils import get_db_sync_dir - -LOGGER = logging.getLogger(__name__) - - -def download_and_extract_node_snapshot(env: str) -> None: - """Download and extracts the Cardano node snapshot for the specified environment.""" - current_directory = os.getcwd() - headers = {"User-Agent": "Mozilla/5.0"} - if env == "mainnet": - snapshot_url = "https://update-cardano-mainnet.iohk.io/cardano-node-state/db-mainnet.tar.gz" - else: - snapshot_url = "" # no other environments are supported for now - - archive_name = f"db-{env}.tar.gz" - - LOGGER.info("Download node snapshot file:") - LOGGER.info(" - current_directory: %s", current_directory) - LOGGER.info(" - download_url: %s", snapshot_url) - LOGGER.info(" - archive name: %s", archive_name) - - with requests.get(snapshot_url, headers=headers, stream=True, timeout=2800) as r: - r.raise_for_status() - with open(archive_name, "wb") as f: - f.writelines(r.iter_content(chunk_size=8192)) - - LOGGER.info(" ------ listdir (before archive extraction): %s", os.listdir(current_directory)) - tf = tarfile.open(pl.Path(current_directory) / archive_name) - tf.extractall(pl.Path(current_directory)) - os.rename(f"db-{env}", "db") - helpers.delete_file(pl.Path(current_directory) / archive_name) - LOGGER.info(" ------ listdir (after archive extraction): %s", os.listdir(current_directory)) - - -def get_latest_snapshot_url(env: str, args: tp.Any) -> str: - """Fetch the latest snapshot URL for the specified environment.""" - github_snapshot_url: str = helpers.get_arg_value(args=args, key="snapshot_url") - if github_snapshot_url != "latest": - return github_snapshot_url - - if env == "mainnet": - general_snapshot_url = "https://update-cardano-mainnet.iohk.io/?list-type=2&delimiter=/&prefix=cardano-db-sync/&max-keys=50&cachestamp=459588" - else: - msg = "Snapshot are currently available only for mainnet environment" - raise ValueError(msg) - - headers = {"Content-type": "application/json"} - res_with_latest_db_sync_version = helpers.request_with_retry( - "get", general_snapshot_url, headers=headers - ) - dict_with_latest_db_sync_version = xmltodict.parse(res_with_latest_db_sync_version.content) - db_sync_latest_version_prefix = dict_with_latest_db_sync_version["ListBucketResult"][ - "CommonPrefixes" - ]["Prefix"] - - if env == "mainnet": - latest_snapshots_list_url = f"https://update-cardano-mainnet.iohk.io/?list-type=2&delimiter=/&prefix={db_sync_latest_version_prefix}&max-keys=50&cachestamp=462903" - else: - msg = "Snapshot are currently available only for mainnet environment" - raise ValueError(msg) - - res_snapshots_list = helpers.request_with_retry( - "get", latest_snapshots_list_url, headers=headers - ) - dict_snapshots_list = xmltodict.parse(res_snapshots_list.content) - latest_snapshot = dict_snapshots_list["ListBucketResult"]["Contents"][-2]["Key"] - - if env == "mainnet": - latest_snapshot_url = f"https://update-cardano-mainnet.iohk.io/{latest_snapshot}" - else: - msg = "Snapshot are currently available only for mainnet environment" - raise ValueError(msg) - - return latest_snapshot_url - - -def download_db_sync_snapshot(snapshot_url: str) -> str: - """Download the database synchronization snapshot from a given URL.""" - current_directory = os.getcwd() - headers = {"User-Agent": "Mozilla/5.0"} - archive_name = snapshot_url.rsplit("/", maxsplit=1)[-1].strip() - - LOGGER.info("Download db-sync snapshot file:") - LOGGER.info(" - current_directory: %s", current_directory) - LOGGER.info(" - download_url: %s", snapshot_url) - LOGGER.info(" - archive name: %s", archive_name) - - with requests.get(snapshot_url, headers=headers, stream=True, timeout=60 * 60) as r: - r.raise_for_status() - with open(archive_name, "wb") as f: - f.writelines(r.iter_content(chunk_size=8192)) - return archive_name - - -def get_snapshot_sha_256_sum(snapshot_url: str) -> str | None: - """Calculate the SHA-256 checksum of the downloaded snapshot.""" - snapshot_sha_256_sum_url = snapshot_url + ".sha256sum" - response = helpers.request_with_retry("get", snapshot_sha_256_sum_url) - for line in response: - return line.decode("utf-8").split(" ")[0] - return None - - -def restore_db_sync_from_snapshot( - config: DbSyncConfig, snapshot_file: str | pl.Path, remove_ledger_dir: str = "yes" -) -> int: - """Restore the Cardano DB Sync database from a snapshot. - - Args: - config: A DbSyncConfig instance with paths and settings. - snapshot_file: Path to the snapshot file to restore. - remove_ledger_dir: Whether to remove the existing ledger directory (defaults to "yes"). - - Returns: - int: Restoration time in seconds. - """ - db_sync_dir = get_db_sync_dir() - snapshot_path = ( - pl.Path(snapshot_file).resolve() - if not isinstance(snapshot_file, pl.Path) - else snapshot_file.resolve() - ) - - if remove_ledger_dir == "yes": - ledger_state_dir = db_sync_dir / "ledger-state" / config.env - if ledger_state_dir.exists(): - shutil.rmtree(ledger_state_dir) - - ledger_dir = db_sync_dir / "ledger-state" / config.env - ledger_dir.mkdir(parents=True, exist_ok=True) - LOGGER.info("ledger_dir: %s", ledger_dir) - - tmp_dir = db_sync_dir / "tmp" - tmp_dir.mkdir(parents=True, exist_ok=True) - helpers.export_env_var("TMPDIR", str(tmp_dir)) - - pgpass_file = db_sync_dir / "config" / f"pgpass-{config.env}" - helpers.export_env_var("PGPASSFILE", str(pgpass_file)) - helpers.export_env_var("ENVIRONMENT", config.env) - helpers.export_env_var("RESTORE_RECREATE_DB", "N") - start_restoration = time.perf_counter() - - script_path = db_sync_dir / "scripts" / "postgresql-setup.sh" - p = subprocess.Popen( - [ - str(script_path), - "--restore-snapshot", - str(snapshot_path), - str(ledger_dir), - ], - cwd=str(db_sync_dir), - stdout=subprocess.PIPE, - ) - try: - outs, errs = p.communicate(timeout=36000) - output = outs.decode("utf-8") - print(f"Restore database: {output}") - if errs: - errors = errs.decode("utf-8") - LOGGER.error("Error during restoration: %s", errors) - - except subprocess.CalledProcessError as e: - msg = "command '{}' return with error (code {}): {}".format( - e.cmd, e.returncode, " ".join(str(e.output).split()) - ) - raise RuntimeError(msg) from e - except subprocess.TimeoutExpired: - p.kill() - LOGGER.exception("Process timeout expired") - - finally: - helpers.export_env_var("TMPDIR", "/tmp") - - if "All good!" not in outs.decode("utf-8"): - msg = "Restoration has not ended successfully" - raise RuntimeError(msg) - - end_restoration = time.perf_counter() - return int(end_restoration - start_restoration) - - -def create_db_sync_snapshot_stage_1(config: DbSyncConfig) -> str: - """Perform the first stage of creating a DB Sync snapshot. - - Args: - config: A DbSyncConfig instance with paths and settings. - - Returns: - str: The command to run for stage 2 of snapshot creation. - """ - db_sync_dir = get_db_sync_dir() - db_tool_binary = db_sync_dir / "_cardano-db-tool" - - pgpass_file = db_sync_dir / "config" / f"pgpass-{config.env}" - helpers.export_env_var("PGPASSFILE", str(pgpass_file)) - - cmd = [str(db_tool_binary), "prepare-snapshot", "--state-dir", f"ledger-state/{config.env}"] - p = subprocess.Popen( - cmd, - cwd=str(db_sync_dir), - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - encoding="utf-8", - ) - - try: - outs, errs = p.communicate(timeout=300) - if errs: - LOGGER.error("Warnings or Errors: %s", errs) - final_line_with_script_cmd = outs.split("\n")[2].lstrip() - LOGGER.info("Snapshot Creation - Stage 1 result: %s", final_line_with_script_cmd) - except subprocess.CalledProcessError as e: - msg = "command '{}' return with error (code {}): {}".format( - e.cmd, e.returncode, " ".join(str(e.output).split()) - ) - raise RuntimeError(msg) from e - else: - return final_line_with_script_cmd - - -def create_db_sync_snapshot_stage_2(config: DbSyncConfig, stage_2_cmd: str) -> str: - """Perform the second stage of creating a DB Sync snapshot. - - Args: - config: A DbSyncConfig instance with paths and settings. - stage_2_cmd: The command to run for stage 2 (generated by stage 1). - - Returns: - str: Path to the created snapshot file. - """ - db_sync_dir = get_db_sync_dir() - - pgpass_file = db_sync_dir / "config" / f"pgpass-{config.env}" - helpers.export_env_var("PGPASSFILE", str(pgpass_file)) - - try: - result = subprocess.run( - stage_2_cmd, - shell=True, - cwd=str(db_sync_dir), - capture_output=True, - text=True, - timeout=43200, - check=False, - ) - - LOGGER.info("Snapshot Creation - Stage 2 Output:\n%s", result.stdout) - if result.stderr: - LOGGER.error("Warnings or Errors:\n%s", result.stderr) - snapshot_line = next( - (line for line in result.stdout.splitlines() if line.startswith("Created")), - "Snapshot creation output not found.", - ) - snapshot_path = ( - snapshot_line.split()[1] if "Created" in snapshot_line else "Snapshot path unknown" - ) - except subprocess.TimeoutExpired as e: - msg = "Snapshot creation timed out." - raise RuntimeError(msg) from e - except subprocess.CalledProcessError as e: - msg = f"Command '{e.cmd}' failed with error: {e.stderr}" - raise RuntimeError(msg) from e - else: - return snapshot_path diff --git a/sync_tests/utils/sync_entries.py b/sync_tests/utils/sync_entries.py index afbbd5ec..c9877ad8 100644 --- a/sync_tests/utils/sync_entries.py +++ b/sync_tests/utils/sync_entries.py @@ -277,20 +277,3 @@ def run_db_sync( perf_stats=perf_stats, db_sync_tip=db_sync_tip, ) - - -def teardown_node_and_db_sync(base_dir: pl.Path, config: db_sync.DbSyncConfig) -> None: - """Terminate node/db-sync processes and clean up after a session. - - Shared by the snapshot-restoration test fixtures so the same teardown - sequence isn't hand-rolled in each one. - - Args: - base_dir: Root directory the node was started in (for DB dir removal). - config: Db-sync configuration used to stop postgres and finalize cleanup. - """ - helpers.manage_process(proc_name="cardano-db-sync", action="terminate") - helpers.manage_process(proc_name="cardano-node", action="terminate") - node.rm_node_db_dir(base_dir=base_dir) - db_sync.stop_postgres(config) - db_sync.finalize_session_disk_cleanup(config)