tests/lts_compatibility.py
def run_ledger_compatibility_since_first(args, local_branch, use_snapshot):
    """
    Tests that a service from the very first LTS can be recovered
    to the next LTS, and so forth, until the version of the local checkout.

    The recovery process uses snapshot is `use_snapshot` is True. Otherwise, the
    entire historical ledger is used.
    """

    LOG.info("Use snapshot: {}", use_snapshot)
    repo = infra.github.Repository()
    lts_releases = repo.get_lts_releases()

    LOG.info(f"LTS releases: {[r[1] for r in lts_releases.items()]}")

    lts_versions = []

    # Add an empty entry to release to indicate local checkout
    # Note: dicts are ordered from Python3.7
    lts_releases[None] = None

    jwt_issuer = infra.jwt_issuer.JwtIssuer("https://localhost")
    with jwt_issuer.start_openid_server():
        txs = app.LoggingTxs(jwt_issuer=jwt_issuer)
        for idx, (_, lts_release) in enumerate(lts_releases.items()):
            if lts_release:
                version, install_path = repo.install_release(lts_release)
                lts_versions.append(version)
                set_js_args(args, install_path)
            else:
                version = args.ccf_version
                install_path = LOCAL_CHECKOUT_DIRECTORY

            binary_dir, library_dir = get_bin_and_lib_dirs_for_install_path(
                install_path
            )

            if not args.dry_run:
                network_args = {
                    "hosts": args.nodes,
                    "binary_dir": binary_dir,
                    "library_dir": library_dir,
                    "txs": txs,
                    "jwt_issuer": jwt_issuer,
                    "version": version,
                }
                if idx == 0:
                    LOG.info(f"Starting new service (version: {version})")
                    network = infra.network.Network(**network_args)
                    network.start_and_join(args)
                else:
                    LOG.info(f"Recovering service (new version: {version})")
                    network = infra.network.Network(
                        **network_args, existing_network=network
                    )
                    network.start_in_recovery(
                        args,
                        ledger_dir,
                        committed_ledger_dir,
                        snapshot_dir=snapshot_dir,
                    )
                    network.recover(args)

                nodes = network.get_joined_nodes()
                primary, _ = network.find_primary()

                # Verify that all nodes run the expected CCF version
                for node in nodes:
                    # Note: /node/version endpoint was added in 2.x
                    if not node.major_version or node.major_version > 1:
                        with node.client() as c:
                            r = c.get("/node/version")
                            expected_version = node.version or args.ccf_version
                            version = r.body.json()["ccf_version"]
                            assert (
                                r.body.json()["ccf_version"] == expected_version
                            ), f"Node version is not {expected_version}"

                # Rollover JWKS so that new primary must read historical CA bundle table
                # and retrieve new keys via auto refresh
                jwt_issuer.refresh_keys()
                # Note: /gov/jwt_keys/all endpoint was added in 2.x
                primary, _ = network.find_nodes()
                if not primary.major_version or primary.major_version > 1:
                    jwt_issuer.wait_for_refresh(network)
                else:
                    time.sleep(3)

                test_new_service(
                    network,
                    args,
                    install_path,
                    binary_dir,
                    library_dir,
                    version,
                )

                snapshot_dir = (
                    network.get_committed_snapshots(primary) if use_snapshot else None
                )
                ledger_dir, committed_ledger_dir = primary.get_ledger(
                    include_read_only_dirs=True
                )
                network.stop_all_nodes(verbose_verification=False)

                # Check that ledger and snapshots can be parsed
                ccf.ledger.Ledger([committed_ledger_dir]).get_latest_public_state()
                if snapshot_dir:
                    for s in os.listdir(snapshot_dir):
                        with ccf.ledger.Snapshot(
                            os.path.join(snapshot_dir, s)
                        ) as snapshot:
                            snapshot.get_public_domain()

    return lts_versions

    Tests that a service from the very first LTS can be recovered
    to the next LTS, and so forth, until the version of the local checkout.

    The recovery process uses snapshot is `use_snapshot` is True. Otherwise, the
    entire historical ledger is used.
    
Tests that a service from the very first LTS can be recovered
to the next LTS, and so forth, until the version of the local checkout.

The recovery process uses snapshot is `use_snapshot` is True. Otherwise, the
entire historical ledger is used.
Tests that a service from the very first LTS can be recovered to the next LTS and so forth until the version of the local checkout . The recovery process uses snapshot is ` use_snapshot ` is True . Otherwise the entire historical ledger is used .
def run_ledger_compatibility_since_first(args, local_branch, use_snapshot):
    LOG.info("Use snapshot: {}", use_snapshot)
    repo = infra.github.Repository()
    lts_releases = repo.get_lts_releases()
    LOG.info(f"LTS releases: {[r[1] for r in lts_releases.items()]}")
    lts_versions = []
    lts_releases[None] = None
    jwt_issuer = infra.jwt_issuer.JwtIssuer("https://localhost")
    with jwt_issuer.start_openid_server():
        txs = app.LoggingTxs(jwt_issuer=jwt_issuer)
        for idx, (_, lts_release) in enumerate(lts_releases.items()):
            if lts_release:
                version, install_path = repo.install_release(lts_release)
                lts_versions.append(version)
                set_js_args(args, install_path)
            else:
                version = args.ccf_version
                install_path = LOCAL_CHECKOUT_DIRECTORY
            binary_dir, library_dir = get_bin_and_lib_dirs_for_install_path(
                install_path
            )
            if not args.dry_run:
                network_args = {
                    "hosts": args.nodes,
                    "binary_dir": binary_dir,
                    "library_dir": library_dir,
                    "txs": txs,
                    "jwt_issuer": jwt_issuer,
                    "version": version,
                }
                if idx == 0:
                    LOG.info(f"Starting new service (version: {version})")
                    network = infra.network.Network(**network_args)
                    network.start_and_join(args)
                else:
                    LOG.info(f"Recovering service (new version: {version})")
                    network = infra.network.Network(
                        **network_args, existing_network=network
                    )
                    network.start_in_recovery(
                        args,
                        ledger_dir,
                        committed_ledger_dir,
                        snapshot_dir=snapshot_dir,
                    )
                    network.recover(args)
                nodes = network.get_joined_nodes()
                primary, _ = network.find_primary()
                for node in nodes:
                    if not node.major_version or node.major_version > 1:
                        with node.client() as c:
                            r = c.get("/node/version")
                            expected_version = node.version or args.ccf_version
                            version = r.body.json()["ccf_version"]
                            assert (
                                r.body.json()["ccf_version"] == expected_version
                            ), f"Node version is not {expected_version}"
                jwt_issuer.refresh_keys()
                primary, _ = network.find_nodes()
                if not primary.major_version or primary.major_version > 1:
                    jwt_issuer.wait_for_refresh(network)
                else:
                    time.sleep(3)
                test_new_service(
                    network,
                    args,
                    install_path,
                    binary_dir,
                    library_dir,
                    version,
                )
                snapshot_dir = (
                    network.get_committed_snapshots(primary) if use_snapshot else None
                )
                ledger_dir, committed_ledger_dir = primary.get_ledger(
                    include_read_only_dirs=True
                )
                network.stop_all_nodes(verbose_verification=False)
                ccf.ledger.Ledger([committed_ledger_dir]).get_latest_public_state()
                if snapshot_dir:
                    for s in os.listdir(snapshot_dir):
                        with ccf.ledger.Snapshot(
                            os.path.join(snapshot_dir, s)
                        ) as snapshot:
                            snapshot.get_public_domain()
    return lts_versions
def run_ledger_compatibility_since_first ( args , local_branch , use_snapshot ) : LOG . info ( "Use snapshot: {}" , use_snapshot ) repo = infra . github . Repository ( ) lts_releases = repo . get_lts_releases ( ) LOG . info ( f"LTS releases: {[r[1] for r in lts_releases.items()]}" ) lts_versions = [ ] lts_releases [ None ] = None jwt_issuer = infra . jwt_issuer . JwtIssuer ( "https://localhost" ) with jwt_issuer . start_openid_server ( ) : txs = app . LoggingTxs ( jwt_issuer = jwt_issuer ) for idx , ( _ , lts_release ) in enumerate ( lts_releases . items ( ) ) : if lts_release : version , install_path = repo . install_release ( lts_release ) lts_versions . append ( version ) set_js_args ( args , install_path ) else : version = args . ccf_version install_path = LOCAL_CHECKOUT_DIRECTORY binary_dir , library_dir = get_bin_and_lib_dirs_for_install_path ( install_path ) if not args . dry_run : network_args = { "hosts" : args . nodes , "binary_dir" : binary_dir , "library_dir" : library_dir , "txs" : txs , "jwt_issuer" : jwt_issuer , "version" : version , } if idx == 0 : LOG . info ( f"Starting new service (version: {version})" ) network = infra . network . Network ( ** network_args ) network . start_and_join ( args ) else : LOG . info ( f"Recovering service (new version: {version})" ) network = infra . network . Network ( ** network_args , existing_network = network ) network . start_in_recovery ( args , ledger_dir , committed_ledger_dir , snapshot_dir = snapshot_dir , ) network . recover ( args ) nodes = network . get_joined_nodes ( ) primary , _ = network . find_primary ( ) for node in nodes : if not node . major_version or node . major_version > 1 : with node . client ( ) as c : r = c . get ( "/node/version" ) expected_version = node . version or args . ccf_version version = r . body . json ( ) [ "ccf_version" ] assert ( r . body . json ( ) [ "ccf_version" ] == expected_version ) , f"Node version is not {expected_version}" jwt_issuer . refresh_keys ( ) primary , _ = network . find_nodes ( ) if not primary . major_version or primary . major_version > 1 : jwt_issuer . wait_for_refresh ( network ) else : time . sleep ( 3 ) test_new_service ( network , args , install_path , binary_dir , library_dir , version , ) snapshot_dir = ( network . get_committed_snapshots ( primary ) if use_snapshot else None ) ledger_dir , committed_ledger_dir = primary . get_ledger ( include_read_only_dirs = True ) network . stop_all_nodes ( verbose_verification = False ) ccf . ledger . Ledger ( [ committed_ledger_dir ] ) . get_latest_public_state ( ) if snapshot_dir : for s in os . listdir ( snapshot_dir ) : with ccf . ledger . Snapshot ( os . path . join ( snapshot_dir , s ) ) as snapshot : snapshot . get_public_domain ( ) return lts_versions
Tests that a service from the very first LTS can be recovered
to the next LTS, and so forth, until the version of the local checkout.
Tests that a service from the very first LTS can be recovered to the next LTS and so forth until the version of the local checkout .
"""
    Tests that a service from the very first LTS can be recovered
    to the next LTS, and so forth, until the version of the local checkout.

    The recovery process uses snapshot is `use_snapshot` is True. Otherwise, the
    entire historical ledger is used.
    """ # Add an empty entry to release to indicate local checkout # Note: dicts are ordered from Python3.7 # Verify that all nodes run the expected CCF version # Note: /node/version endpoint was added in 2.x # Rollover JWKS so that new primary must read historical CA bundle table # and retrieve new keys via auto refresh # Note: /gov/jwt_keys/all endpoint was added in 2.x # Check that ledger and snapshots can be parsed
args
local_branch
use_snapshot