<commit_before>from os.path import join

import pytest

from cosmo_tester.framework.examples import get_example_deployment
from cosmo_tester.framework.util import (get_resource_path,
                                         validate_cluster_status_and_agents)


@pytest.mark.nine_vms
def test_replace_certificates_on_cluster(full_cluster_ips, logger, ssh_key,
                                         test_config, replace_ca_key=False):
    broker1, broker2, broker3, db1, db2, db3, mgr1, mgr2, mgr3 = \
        full_cluster_ips

    example = get_example_deployment(mgr1, ssh_key, logger,
                                     'cluster_replace_certs', test_config)
    example.inputs['server_ip'] = mgr1.ip_address
    example.upload_and_verify_install()
    validate_cluster_status_and_agents(mgr1, example.tenant, logger)

    for host in broker1, broker2, broker3, db1, db2, db3, mgr1, mgr2, mgr3:
        key_path = join('~', '.cloudify-test-ca',
                        host.private_ip_address + '.key')
        mgr1.run_command('cfy_manager generate-test-cert'
                         ' -s {0},{1}'.format(host.private_ip_address,
                                              host.ip_address))
        mgr1.run_command('chmod 444 {0}'.format(key_path), use_sudo=True)
    replace_certs_config_path = '~/certificates_replacement_config.yaml'
    _create_replace_certs_config_file(mgr1, replace_certs_config_path,
                                      ssh_key.private_key_path,
                                      replace_ca_key=replace_ca_key)

    mgr1.run_command('cfy certificates replace -i {0} -v'.format(
        replace_certs_config_path))
    mgr1.download_rest_ca(force=True)

    validate_cluster_status_and_agents(mgr1, example.tenant, logger)
    example.uninstall()


@pytest.mark.three_vms
def test_replace_certificates_on_compact_cluster(
        three_nodes_cluster, logger, ssh_key, test_config,
        replace_ca_key=False):
    node1, node2, node3 = three_nodes_cluster

    example = get_example_deployment(node1, ssh_key, logger,
                                     'cluster_replace_certs', test_config)
    example.inputs['server_ip'] = node1.ip_address
    example.upload_and_verify_install()
    validate_cluster_status_and_agents(node1, example.tenant, logger)

    for host in node1, node2, node3:
        key_path = join('~', '.cloudify-test-ca',
                        host.private_ip_address + '.key')
        node1.run_command('cfy_manager generate-test-cert'
                          ' -s {0},{1}'.format(host.private_ip_address,
                                               host.ip_address))
        node1.run_command('chmod 444 {0}'.format(key_path), use_sudo=True)
    replace_certs_config_path = '~/certificates_replacement_config.yaml'
    _create_replace_certs_config_file(node1, replace_certs_config_path,
                                      ssh_key.private_key_path,
                                      replace_ca_key=replace_ca_key)

    node1.run_command('cfy certificates replace -i {0} -v'.format(
        replace_certs_config_path))
    node1.download_rest_ca(force=True)

    validate_cluster_status_and_agents(node1, example.tenant, logger)
    example.uninstall()


@pytest.mark.nine_vms
def test_replace_certificates_on_cluster_incl_ca_key(
        full_cluster_ips, logger, ssh_key, test_config):
    test_replace_certificates_on_cluster(full_cluster_ips, logger, ssh_key,
                                         test_config, replace_ca_key=True)


@pytest.mark.three_vms
def test_replace_certificates_compact_cluster_incl_ca_key(
        three_nodes_cluster, logger, ssh_key, test_config):
    test_replace_certificates_on_compact_cluster(
        three_nodes_cluster, logger, ssh_key, test_config, replace_ca_key=True)


def _create_replace_certs_config_file(manager,
                                      replace_certs_config_path,
                                      local_ssh_key_path,
                                      replace_ca_key=False):<commit_msg>Mark cluster cert replacement tests (#1450)<commit_after>from os.path import join

import pytest

from cosmo_tester.framework.examples import get_example_deployment
from cosmo_tester.framework.util import (get_resource_path,
                                         validate_cluster_status_and_agents)


@pytest.mark.nine_vms
@pytest.mark.cert_replace
def test_replace_certificates_on_cluster(full_cluster_ips, logger, ssh_key,
                                         test_config, replace_ca_key=False):
    broker1, broker2, broker3, db1, db2, db3, mgr1, mgr2, mgr3 = \
        full_cluster_ips

    example = get_example_deployment(mgr1, ssh_key, logger,
                                     'cluster_replace_certs', test_config)
    example.inputs['server_ip'] = mgr1.ip_address
    example.upload_and_verify_install()
    validate_cluster_status_and_agents(mgr1, example.tenant, logger)

    for host in broker1, broker2, broker3, db1, db2, db3, mgr1, mgr2, mgr3:
        key_path = join('~', '.cloudify-test-ca',
                        host.private_ip_address + '.key')
        mgr1.run_command('cfy_manager generate-test-cert'
                         ' -s {0},{1}'.format(host.private_ip_address,
                                              host.ip_address))
        mgr1.run_command('chmod 444 {0}'.format(key_path), use_sudo=True)
    replace_certs_config_path = '~/certificates_replacement_config.yaml'
    _create_replace_certs_config_file(mgr1, replace_certs_config_path,
                                      ssh_key.private_key_path,
                                      replace_ca_key=replace_ca_key)

    mgr1.run_command('cfy certificates replace -i {0} -v'.format(
        replace_certs_config_path))
    mgr1.download_rest_ca(force=True)

    validate_cluster_status_and_agents(mgr1, example.tenant, logger)
    example.uninstall()


@pytest.mark.three_vms
@pytest.mark.cert_replace
def test_replace_certificates_on_compact_cluster(
        three_nodes_cluster, logger, ssh_key, test_config,
        replace_ca_key=False):
    node1, node2, node3 = three_nodes_cluster

    example = get_example_deployment(node1, ssh_key, logger,
                                     'cluster_replace_certs', test_config)
    example.inputs['server_ip'] = node1.ip_address
    example.upload_and_verify_install()
    validate_cluster_status_and_agents(node1, example.tenant, logger)

    for host in node1, node2, node3:
        key_path = join('~', '.cloudify-test-ca',
                        host.private_ip_address + '.key')
        node1.run_command('cfy_manager generate-test-cert'
                          ' -s {0},{1}'.format(host.private_ip_address,
                                               host.ip_address))
        node1.run_command('chmod 444 {0}'.format(key_path), use_sudo=True)
    replace_certs_config_path = '~/certificates_replacement_config.yaml'
    _create_replace_certs_config_file(node1, replace_certs_config_path,
                                      ssh_key.private_key_path,
                                      replace_ca_key=replace_ca_key)

    node1.run_command('cfy certificates replace -i {0} -v'.format(
        replace_certs_config_path))
    node1.download_rest_ca(force=True)

    validate_cluster_status_and_agents(node1, example.tenant, logger)
    example.uninstall()


@pytest.mark.nine_vms
@pytest.mark.cert_replace
def test_replace_certificates_on_cluster_incl_ca_key(
        full_cluster_ips, logger, ssh_key, test_config):
    test_replace_certificates_on_cluster(full_cluster_ips, logger, ssh_key,
                                         test_config, replace_ca_key=True)


@pytest.mark.three_vms
@pytest.mark.cert_replace
def test_replace_certificates_compact_cluster_incl_ca_key(
        three_nodes_cluster, logger, ssh_key, test_config):
    test_replace_certificates_on_compact_cluster(
        three_nodes_cluster, logger, ssh_key, test_config, replace_ca_key=True)


def _create_replace_certs_config_file(manager,
                                      replace_certs_config_path,
                                      local_ssh_key_path,
                                      replace_ca_key=False):
{'fraction_non_alphanumeric': 0.08010241404535479, 'fraction_numerical': 0.014142891977566447, 'mean_word_length': 2.5236254295532645, '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': 42, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3437997', 'n_tokens_mistral': 2574, 'n_tokens_neox': 2487, 'n_words': 368}