<s> """<STR_LIT>""" <EOL> import yaml <EOL> from pyrsistent import freeze , thaw <EOL> from textwrap import dedent <EOL> from . . _install import ( <EOL> task_configure_flocker_agent , <EOL> task_enable_flocker_agent , <EOL> run , put , run_from_args , <EOL> get_repository_url , UnsupportedDistribution , get_installable_version , <EOL> get_repo_options , <EOL> _remove_dataset_fields , _remove_private_key , <EOL> UnknownAction , DistributionNotSupported ) <EOL> from . . _ssh import Put <EOL> from . . _effect import sequence <EOL> from ... node . backends import backend_loader <EOL> from ... testtools import TestCase <EOL> from ... import __version__ as flocker_version <EOL> THE_AGENT_YML_PATH = b"<STR_LIT>" <EOL> BASIC_AGENT_YML = freeze ( { <EOL> "<STR_LIT:version>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:port>" : <NUM_LIT> , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:version>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } , <EOL> } , <EOL> } , <EOL> } ) <EOL> class ConfigureFlockerAgentTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_agent_yml ( self ) : <EOL> """<STR_LIT>""" <EOL> control_address = BASIC_AGENT_YML [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> expected_pool = u"<STR_LIT>" <EOL> expected_backend_configuration = dict ( pool = expected_pool ) <EOL> commands = task_configure_flocker_agent ( <EOL> control_node = control_address , <EOL> dataset_backend = backend_loader . get ( <EOL> BASIC_AGENT_YML [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> ) , <EOL> dataset_backend_configuration = expected_backend_configuration , <EOL> logging_config = thaw ( BASIC_AGENT_YML [ "<STR_LIT>" ] ) , <EOL> ) <EOL> [ put_agent_yml ] = list ( <EOL> effect . intent <EOL> for effect in <EOL> commands . intent . effects <EOL> if isinstance ( effect . intent , Put ) <EOL> ) <EOL> expected_agent_config = BASIC_AGENT_YML . set ( <EOL> "<STR_LIT>" , <EOL> BASIC_AGENT_YML [ "<STR_LIT>" ] . update ( expected_backend_configuration ) <EOL> ) <EOL> self . assertEqual ( <EOL> put ( <EOL> content = yaml . safe_dump ( thaw ( expected_agent_config ) ) , <EOL> path = THE_AGENT_YML_PATH , <EOL> log_content_filter = _remove_dataset_fields , <EOL> ) . intent , <EOL> put_agent_yml , <EOL> ) <EOL> class EnableFlockerAgentTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_centos_sequence ( self ) : <EOL> """<STR_LIT>""" <EOL> distribution = u"<STR_LIT>" <EOL> commands = task_enable_flocker_agent ( <EOL> distribution = distribution , <EOL> ) <EOL> expected_sequence = sequence ( [ <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> ] ) <EOL> self . assertEqual ( commands , expected_sequence ) <EOL> def test_centos_sequence_managed ( self ) : <EOL> """<STR_LIT>""" <EOL> distribution = u"<STR_LIT>" <EOL> commands = task_enable_flocker_agent ( <EOL> distribution = distribution , <EOL> action = "<STR_LIT>" <EOL> ) <EOL> expected_sequence = sequence ( [ <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> ] ) <EOL> self . assertEqual ( commands , expected_sequence ) <EOL> def test_ubuntu_sequence ( self ) : <EOL> """<STR_LIT>""" <EOL> distribution = u"<STR_LIT>" <EOL> commands = task_enable_flocker_agent ( <EOL> distribution = distribution , <EOL> ) <EOL> expected_sequence = sequence ( [ <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> ] ) <EOL> self . assertEqual ( commands , expected_sequence ) <EOL> def test_ubuntu_sequence_managed ( self ) : <EOL> """<STR_LIT>""" <EOL> distribution = u"<STR_LIT>" <EOL> commands = task_enable_flocker_agent ( <EOL> distribution = distribution , <EOL> action = "<STR_LIT>" <EOL> ) <EOL> expected_sequence = sequence ( [ <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" ) , <EOL> ] ) <EOL> self . assertEqual ( commands , expected_sequence ) <EOL> def test_sequence_invalid_action ( self ) : <EOL> """<STR_LIT>""" <EOL> distribution = u"<STR_LIT>" <EOL> self . assertRaises ( UnknownAction , <EOL> task_enable_flocker_agent , <EOL> distribution = distribution , <EOL> action = "<STR_LIT>" ) <EOL> def test_sequence_invalid_distro ( self ) : <EOL> """<STR_LIT>""" <EOL> distribution = u"<STR_LIT>" <EOL> self . assertRaises ( DistributionNotSupported , <EOL> task_enable_flocker_agent , <EOL> distribution = distribution , <EOL> action = "<STR_LIT>" ) <EOL> def _centos7_install_commands ( version ) : <EOL> """<STR_LIT>""" <EOL> installable_version = get_installable_version ( flocker_version ) <EOL> return sequence ( [ <EOL> run ( command = "<STR_LIT>" ) , <EOL> run ( command = "<STR_LIT>" . format ( get_repository_url ( <EOL> distribution = '<STR_LIT>' , <EOL> flocker_version = installable_version , <EOL> ) ) ) , <EOL> run_from_args ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] + get_repo_options ( installable_version ) + <EOL> [ '<STR_LIT>' , '<STR_LIT>' + version ] ) <EOL> ] ) <EOL> class GetRepoOptionsTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_marketing_release ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( get_repo_options ( flocker_version = '<STR_LIT>' ) , [ ] ) <EOL> def test_development_release ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> get_repo_options ( flocker_version = '<STR_LIT>' ) , <EOL> [ '<STR_LIT>' ] ) <EOL> class GetRepositoryURLTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_centos_7 ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> get_repository_url ( <EOL> distribution = '<STR_LIT>' , <EOL> flocker_version = '<STR_LIT>' ) , <EOL> expected <EOL> ) <EOL> def test_ubuntu_14_04 ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> get_repository_url ( <EOL> distribution = '<STR_LIT>' , <EOL> flocker_version = '<STR_LIT>' ) , <EOL> expected <EOL> ) <EOL> def test_ubuntu_15_10 ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> get_repository_url ( <EOL> distribution = '<STR_LIT>' , <EOL> flocker_version = '<STR_LIT>' ) , <EOL> expected <EOL> ) <EOL> def test_unsupported_distribution ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( <EOL> UnsupportedDistribution , <EOL> get_repository_url , '<STR_LIT>' , '<STR_LIT>' , <EOL> ) <EOL> def test_non_release_ubuntu ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> get_repository_url ( <EOL> distribution = '<STR_LIT>' , <EOL> flocker_version = '<STR_LIT>' ) , <EOL> expected <EOL> ) <EOL> def test_non_release_centos ( self ) : <EOL> """<STR_LIT>""" <EOL> expected = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> get_repository_url ( <EOL> distribution = '<STR_LIT>' , <EOL> flocker_version = '<STR_LIT>' ) , <EOL> expected <EOL> ) <EOL> class PrivateKeyLoggingTest ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_private_key_removed ( self ) : <EOL> """<STR_LIT>""" <EOL> key = dedent ( '''<STR_LIT>''' ) <EOL> self . assertEqual ( <EOL> dedent ( '''<STR_LIT>''' ) , <EOL> _remove_private_key ( key ) ) <EOL> def test_non_key_kept ( self ) : <EOL> """<STR_LIT>""" <EOL> key = '<STR_LIT>' <EOL> self . assertEqual ( key , _remove_private_key ( key ) ) <EOL> def test_short_key_kept ( self ) : <EOL> """<STR_LIT>""" <EOL> key = dedent ( '''<STR_LIT>''' ) <EOL> self . assertEqual ( key , _remove_private_key ( key ) ) <EOL> def test_no_end_key_removed ( self ) : <EOL> """<STR_LIT>""" <EOL> key = dedent ( '''<STR_LIT>''' ) <EOL> self . assertEqual ( <EOL> '<STR_LIT>' , <EOL> _remove_private_key ( key ) ) <EOL> class DatasetLoggingTest ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_dataset_logged_safely ( self ) : <EOL> """<STR_LIT>""" <EOL> config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> content = yaml . safe_dump ( config ) <EOL> logged = _remove_dataset_fields ( content ) <EOL> self . assertEqual ( <EOL> yaml . safe_load ( logged ) , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } ) </s>
