<s> import mock <EOL> from oslo_serialization import jsonutils <EOL> from nailgun import consts <EOL> from nailgun . test import base <EOL> from nailgun . utils import reverse <EOL> from . import base as tests_base <EOL> class TestClusterUpgradeCloneHandler ( tests_base . BaseCloneClusterTest ) : <EOL> def test_clone ( self ) : <EOL> resp = self . app . post ( <EOL> reverse ( "<STR_LIT>" , <EOL> kwargs = { "<STR_LIT>" : self . src_cluster . id } ) , <EOL> jsonutils . dumps ( self . data ) , <EOL> headers = self . default_headers ) <EOL> body = resp . json_body <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( body [ "<STR_LIT:name>" ] , <EOL> "<STR_LIT>" . format ( self . src_cluster . id ) ) <EOL> self . assertEqual ( body [ "<STR_LIT>" ] , self . dst_release . id ) <EOL> def test_clone_cluster_not_found_error ( self ) : <EOL> resp = self . app . post ( <EOL> reverse ( "<STR_LIT>" , <EOL> kwargs = { "<STR_LIT>" : <NUM_LIT> } ) , <EOL> jsonutils . dumps ( self . data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( resp . json_body [ "<STR_LIT:message>" ] , "<STR_LIT>" ) <EOL> def test_clone_cluster_already_in_upgrade_error ( self ) : <EOL> self . app . post ( <EOL> reverse ( "<STR_LIT>" , <EOL> kwargs = { "<STR_LIT>" : self . src_cluster . id } ) , <EOL> jsonutils . dumps ( self . data ) , <EOL> headers = self . default_headers ) <EOL> resp = self . app . post ( <EOL> reverse ( "<STR_LIT>" , <EOL> kwargs = { "<STR_LIT>" : self . src_cluster . id } ) , <EOL> jsonutils . dumps ( self . data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> def test_clone_cluster_name_already_exists_error ( self ) : <EOL> data = dict ( self . data , name = self . src_cluster . name ) <EOL> resp = self . app . post ( <EOL> reverse ( "<STR_LIT>" , <EOL> kwargs = { "<STR_LIT>" : self . src_cluster . id } ) , <EOL> jsonutils . dumps ( data ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> class TestNodeReassignHandler ( base . BaseIntegrationTest ) : <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_node_reassign_handler ( self , mcast ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT:status>' : consts . NODE_STATUSES . ready } ] ) <EOL> seed_cluster = self . env . create_cluster ( ) <EOL> node_id = cluster . nodes [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : seed_cluster [ '<STR_LIT:id>' ] } ) , <EOL> jsonutils . dumps ( { '<STR_LIT>' : node_id } ) , <EOL> headers = self . default_headers ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> args , kwargs = mcast . call_args <EOL> nodes = args [ <NUM_LIT:1> ] [ '<STR_LIT:args>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> provisioned_uids = [ int ( n [ '<STR_LIT>' ] ) for n in nodes ] <EOL> self . assertEqual ( [ node_id , ] , provisioned_uids ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_node_reassign_handler_with_roles ( self , mcast ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT:status>' : consts . NODE_STATUSES . ready , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ] ) <EOL> node = cluster . nodes [ <NUM_LIT:0> ] <EOL> seed_cluster = self . env . create_cluster ( api = False ) <EOL> data = { '<STR_LIT>' : node . id , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : seed_cluster . id } ) , <EOL> jsonutils . dumps ( data ) , <EOL> headers = self . default_headers ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertEqual ( node . roles , [ ] ) <EOL> self . assertEqual ( node . pending_roles , [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( mcast . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_node_reassign_handler_without_reprovisioning ( self , mcast ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT:status>' : consts . NODE_STATUSES . ready , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } ] ) <EOL> node = cluster . nodes [ <NUM_LIT:0> ] <EOL> seed_cluster = self . env . create_cluster ( api = False ) <EOL> data = { '<STR_LIT>' : node . id , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] } <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : seed_cluster . id } ) , <EOL> jsonutils . dumps ( data ) , <EOL> headers = self . default_headers ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code ) <EOL> self . assertFalse ( mcast . called ) <EOL> self . assertEqual ( node . roles , [ '<STR_LIT>' ] ) <EOL> def test_node_reassign_handler_no_node ( self ) : <EOL> cluster = self . env . create_cluster ( ) <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : cluster [ '<STR_LIT:id>' ] } ) , <EOL> jsonutils . dumps ( { '<STR_LIT>' : <NUM_LIT> } ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> resp . json_body [ '<STR_LIT:message>' ] ) <EOL> def test_node_reassing_handler_wrong_status ( self ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT:status>' : '<STR_LIT>' } ] ) <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : cluster [ '<STR_LIT:id>' ] } ) , <EOL> jsonutils . dumps ( { '<STR_LIT>' : cluster . nodes [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertRegexpMatches ( resp . json_body [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_node_reassing_handler_wrong_error_type ( self ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT:status>' : '<STR_LIT:error>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] ) <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : cluster [ '<STR_LIT:id>' ] } ) , <EOL> jsonutils . dumps ( { '<STR_LIT>' : cluster . nodes [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertRegexpMatches ( resp . json_body [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" ) <EOL> def test_node_reassign_handler_to_the_same_cluster ( self ) : <EOL> cluster = self . env . create ( <EOL> cluster_kwargs = { '<STR_LIT>' : False } , <EOL> nodes_kwargs = [ { '<STR_LIT:status>' : '<STR_LIT>' } ] ) <EOL> cluster_id = cluster [ '<STR_LIT:id>' ] <EOL> node_id = cluster . nodes [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : cluster_id } ) , <EOL> jsonutils . dumps ( { '<STR_LIT>' : node_id } ) , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> self . assertEqual ( "<STR_LIT>" . <EOL> format ( node_id , cluster_id ) , <EOL> resp . json_body [ '<STR_LIT:message>' ] ) <EOL> def test_node_reassign_handler_with_empty_data ( self ) : <EOL> cluster = self . env . create_cluster ( api = False ) <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : cluster . id } ) , <EOL> "<STR_LIT:{}>" , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> def test_node_reassign_handler_with_empty_body ( self ) : <EOL> cluster = self . env . create_cluster ( api = False ) <EOL> resp = self . app . post ( <EOL> reverse ( '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : cluster . id } ) , <EOL> "<STR_LIT>" , <EOL> headers = self . default_headers , <EOL> expect_errors = True ) <EOL> self . assertEqual ( <NUM_LIT> , resp . status_code ) <EOL> class TestCopyVipsHandler ( base . BaseIntegrationTest ) : <EOL> def test_copy_vips_called ( self ) : <EOL> from . . objects import relations <EOL> orig_cluster = self . env . create_cluster ( api = False ) <EOL> new_cluster = self . env . create_cluster ( api = False ) <EOL> relations . UpgradeRelationObject . create_relation ( <EOL> orig_cluster . id , new_cluster . id ) <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) as copy_vips_mc : <EOL> resp = self . app . post ( <EOL> reverse ( <EOL> '<STR_LIT>' , <EOL> kwargs = { '<STR_LIT>' : new_cluster . id } <EOL> ) , <EOL> headers = self . default_headers , <EOL> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertTrue ( copy_vips_mc . called ) </s>
