<commit_before>from copy import copy

import base58
import pytest

from crypto.bls.bls_bft_replica import BlsBftReplica
from crypto.bls.bls_multi_signature import MultiSignature, MultiSignatureValue
from plenum.bls.bls_bft_factory import create_default_bls_bft_factory
from plenum.common.constants import DOMAIN_LEDGER_ID, POOL_LEDGER_ID, CONFIG_LEDGER_ID, AUDIT_LEDGER_ID, TXN_PAYLOAD, \
    TXN_PAYLOAD_DATA, AUDIT_TXN_STATE_ROOT, AUDIT_TXN_LEDGER_ROOT
from plenum.common.messages.node_messages import PrePrepare
from plenum.common.types import f
from plenum.common.util import get_utc_epoch
from plenum.server.quorums import Quorums
from plenum.test.bls.helper import process_commits_for_key, calculate_multi_sig, process_ordered
from plenum.test.helper import create_pre_prepare_params, create_pre_prepare_no_bls, create_commit_params, \
    create_commit_no_bls_sig, create_commit_with_bls_sig, create_commit_bls_sig, create_prepare_params, create_prepare, \
    generate_state_root, create_commit_with_bls_sigs

whitelist = ['Indy Crypto error']


@pytest.fixture()
def bls_bft_replicas(txnPoolNodeSet):
    bls_bft_replicas = []
    for node in txnPoolNodeSet:
        bls_bft_replica = create_default_bls_bft_factory(node).create_bls_bft_replica(is_master=True)
        bls_bft_replicas.append(bls_bft_replica)
    return bls_bft_replicas


@pytest.fixture()
def quorums(txnPoolNodeSet):
    return Quorums(len(txnPoolNodeSet))


@pytest.fixture(params=[POOL_LEDGER_ID, DOMAIN_LEDGER_ID, CONFIG_LEDGER_ID])
def ledger_id(request):
    return request.param


@pytest.fixture()
def state_root():
    return generate_state_root()


@pytest.fixture()
def pool_state_root(bls_bft_replicas):
    bls_bft_replica = bls_bft_replicas[0]
    return bls_bft_replica.state_root_serializer.serialize(
        bytes(bls_bft_replica._bls_bft.bls_key_register.get_pool_root_hash_committed()))


@pytest.fixture()
def fake_pre_prepare_with_bls(fake_multi_sig, ledger_id):
    params = create_pre_prepare_params(state_root=fake_multi_sig.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=fake_multi_sig.value.pool_state_root_hash,
                                       bls_multi_sig=fake_multi_sig)
    return PrePrepare(*params)


@pytest.fixture()
def multi_sig_value(state_root, pool_state_root, ledger_id):
    return MultiSignatureValue(ledger_id=ledger_id,
                               state_root_hash=state_root,
                               pool_state_root_hash=pool_state_root,
                               txn_root_hash=generate_state_root(),
                               timestamp=get_utc_epoch())


@pytest.fixture()
def multi_signature(bls_bft_replicas, multi_sig_value):
    sigs = []
    participants = []
    message = multi_sig_value.as_single_value()
    for bls_bft_replica in bls_bft_replicas:
        sigs.append(bls_bft_replica._bls_bft.bls_crypto_signer.sign(message))
        participants.append(bls_bft_replica.node_id)

    multi_sig = bls_bft_replicas[0]._bls_bft.bls_crypto_verifier.create_multi_sig(sigs)

    return MultiSignature(signature=multi_sig,
                          participants=participants,
                          value=multi_sig_value)


@pytest.fixture()
def pre_prepare_with_bls(multi_signature, ledger_id):
    params = create_pre_prepare_params(state_root=multi_signature.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=multi_signature.value.pool_state_root_hash,
                                       bls_multi_sig=multi_signature)
    return PrePrepare(*params)

@pytest.fixture()
def pre_prepare_with_bls_multi(multi_signature, ledger_id):
    params = create_pre_prepare_params(state_root=multi_signature.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=multi_signature.value.pool_state_root_hash,
                                       bls_multi_sig=multi_signature,
                                       bls_multi_sigs=[multi_signature])
    return PrePrepare(*params)


@pytest.fixture()
def pre_prepare_with_incorrect_bls(multi_signature, ledger_id):
    multi_signature.signature = base58.b58encode(b"somefakesignaturesomefakesignaturesomefakesignature").decode("utf-8")
    params = create_pre_prepare_params(state_root=multi_signature.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=multi_signature.value.pool_state_root_hash,
                                       bls_multi_sig=multi_signature)
    return PrePrepare(*params)


@pytest.fixture()
def pre_prepare_no_bls(state_root, pool_state_root, ledger_id):
    params = create_pre_prepare_params(state_root=state_root,
                                       ledger_id=ledger_id,
                                       pool_state_root=pool_state_root)
    return PrePrepare(*params)


@pytest.fixture()
def create_audit_txn_with_multiple_ledgers(txnPoolNodeSet):
    for node in txnPoolNodeSet:
        audit_ledger = node.db_manager.get_ledger(AUDIT_LEDGER_ID)
        audit_ledger.get_last_txn = lambda: {
            TXN_PAYLOAD: {
                TXN_PAYLOAD_DATA: {
                    AUDIT_TXN_STATE_ROOT: {
                        POOL_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm",
                        DOMAIN_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm"
                    },
                    AUDIT_TXN_LEDGER_ROOT: {
                        POOL_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm",
                        DOMAIN_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm"
                    }
                }
            }
        }


@pytest.yield_fixture(scope="function", params=['state_root', 'timestamp', 'txn_root'])
def pre_prepare_incorrect(state_root, request, ledger_id):<commit_msg>ST-623 -- Make fixture yield and fix<commit_after>from copy import copy

import base58
import pytest

from crypto.bls.bls_bft_replica import BlsBftReplica
from crypto.bls.bls_multi_signature import MultiSignature, MultiSignatureValue
from plenum.bls.bls_bft_factory import create_default_bls_bft_factory
from plenum.common.constants import DOMAIN_LEDGER_ID, POOL_LEDGER_ID, CONFIG_LEDGER_ID, AUDIT_LEDGER_ID, TXN_PAYLOAD, \
    TXN_PAYLOAD_DATA, AUDIT_TXN_STATE_ROOT, AUDIT_TXN_LEDGER_ROOT
from plenum.common.messages.node_messages import PrePrepare
from plenum.common.types import f
from plenum.common.util import get_utc_epoch
from plenum.server.quorums import Quorums
from plenum.test.bls.helper import process_commits_for_key, calculate_multi_sig, process_ordered
from plenum.test.helper import create_pre_prepare_params, create_pre_prepare_no_bls, create_commit_params, \
    create_commit_no_bls_sig, create_commit_with_bls_sig, create_commit_bls_sig, create_prepare_params, create_prepare, \
    generate_state_root, create_commit_with_bls_sigs

whitelist = ['Indy Crypto error']


@pytest.fixture()
def bls_bft_replicas(txnPoolNodeSet):
    bls_bft_replicas = []
    for node in txnPoolNodeSet:
        bls_bft_replica = create_default_bls_bft_factory(node).create_bls_bft_replica(is_master=True)
        bls_bft_replicas.append(bls_bft_replica)
    return bls_bft_replicas


@pytest.fixture()
def quorums(txnPoolNodeSet):
    return Quorums(len(txnPoolNodeSet))


@pytest.fixture(params=[POOL_LEDGER_ID, DOMAIN_LEDGER_ID, CONFIG_LEDGER_ID])
def ledger_id(request):
    return request.param


@pytest.fixture()
def state_root():
    return generate_state_root()


@pytest.fixture()
def pool_state_root(bls_bft_replicas):
    bls_bft_replica = bls_bft_replicas[0]
    return bls_bft_replica.state_root_serializer.serialize(
        bytes(bls_bft_replica._bls_bft.bls_key_register.get_pool_root_hash_committed()))


@pytest.fixture()
def fake_pre_prepare_with_bls(fake_multi_sig, ledger_id):
    params = create_pre_prepare_params(state_root=fake_multi_sig.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=fake_multi_sig.value.pool_state_root_hash,
                                       bls_multi_sig=fake_multi_sig)
    return PrePrepare(*params)


@pytest.fixture()
def multi_sig_value(state_root, pool_state_root, ledger_id):
    return MultiSignatureValue(ledger_id=ledger_id,
                               state_root_hash=state_root,
                               pool_state_root_hash=pool_state_root,
                               txn_root_hash=generate_state_root(),
                               timestamp=get_utc_epoch())


@pytest.fixture()
def multi_signature(bls_bft_replicas, multi_sig_value):
    sigs = []
    participants = []
    message = multi_sig_value.as_single_value()
    for bls_bft_replica in bls_bft_replicas:
        sigs.append(bls_bft_replica._bls_bft.bls_crypto_signer.sign(message))
        participants.append(bls_bft_replica.node_id)

    multi_sig = bls_bft_replicas[0]._bls_bft.bls_crypto_verifier.create_multi_sig(sigs)

    return MultiSignature(signature=multi_sig,
                          participants=participants,
                          value=multi_sig_value)


@pytest.fixture()
def pre_prepare_with_bls(multi_signature, ledger_id):
    params = create_pre_prepare_params(state_root=multi_signature.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=multi_signature.value.pool_state_root_hash,
                                       bls_multi_sig=multi_signature)
    return PrePrepare(*params)

@pytest.fixture()
def pre_prepare_with_bls_multi(multi_signature, ledger_id):
    params = create_pre_prepare_params(state_root=multi_signature.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=multi_signature.value.pool_state_root_hash,
                                       bls_multi_sig=multi_signature,
                                       bls_multi_sigs=[multi_signature])
    return PrePrepare(*params)


@pytest.fixture()
def pre_prepare_with_incorrect_bls(multi_signature, ledger_id):
    multi_signature.signature = base58.b58encode(b"somefakesignaturesomefakesignaturesomefakesignature").decode("utf-8")
    params = create_pre_prepare_params(state_root=multi_signature.value.state_root_hash,
                                       ledger_id=ledger_id,
                                       pool_state_root=multi_signature.value.pool_state_root_hash,
                                       bls_multi_sig=multi_signature)
    return PrePrepare(*params)


@pytest.fixture()
def pre_prepare_no_bls(state_root, pool_state_root, ledger_id):
    params = create_pre_prepare_params(state_root=state_root,
                                       ledger_id=ledger_id,
                                       pool_state_root=pool_state_root)
    return PrePrepare(*params)


@pytest.yield_fixture()
def create_audit_txn_with_multiple_ledgers(txnPoolNodeSet):
    for node in txnPoolNodeSet:
        audit_ledger = node.db_manager.get_ledger(AUDIT_LEDGER_ID)
        old_last = audit_ledger.get_last_txn
        audit_ledger.get_last_txn = lambda: {
            TXN_PAYLOAD: {
                TXN_PAYLOAD_DATA: {
                    AUDIT_TXN_STATE_ROOT: {
                        POOL_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm",
                        DOMAIN_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm"
                    },
                    AUDIT_TXN_LEDGER_ROOT: {
                        POOL_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm",
                        DOMAIN_LEDGER_ID: "2UQ3Da54cQ6SamunzXVAtBozFnkACELBH7HzbRPgfKzm"
                    }
                }
            }
        }
        yield audit_ledger
        audit_ledger.get_last_txn = old_last


@pytest.yield_fixture(scope="function", params=['state_root', 'timestamp', 'txn_root'])
def pre_prepare_incorrect(state_root, request, ledger_id):
{'fraction_non_alphanumeric': 0.06654116876739237, 'fraction_numerical': 0.005647405467343264, 'mean_word_length': 2.888924252068746, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3505650', 'n_tokens_mistral': 4213, 'n_tokens_neox': 4151, 'n_words': 468}