<s> import re <EOL> from nailgun import consts <EOL> from nailgun . db . sqlalchemy . models import Notification <EOL> from nailgun . test . base import BaseIntegrationTest <EOL> from nailgun . test . base import fake_tasks <EOL> class TestErrors ( BaseIntegrationTest ) : <EOL> @ fake_tasks ( error = "<STR_LIT>" ) <EOL> def test_deployment_error_during_provisioning ( self ) : <EOL> self . env . create ( <EOL> cluster_kwargs = { } , <EOL> nodes_kwargs = [ <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True } , <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : True } <EOL> ] <EOL> ) <EOL> supertask = self . env . launch_deployment ( ) <EOL> message = re . compile ( <EOL> "<STR_LIT>" <EOL> ) <EOL> self . assertEqual ( supertask . status , consts . TASK_STATUSES . error ) <EOL> self . assertRegexpMatches ( supertask . message , message ) <EOL> self . env . refresh_nodes ( ) <EOL> self . env . refresh_clusters ( ) <EOL> n_error = lambda n : ( n . status , n . error_type ) == ( '<STR_LIT:error>' , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> sum ( map ( n_error , self . env . nodes ) ) , <EOL> <NUM_LIT:1> <EOL> ) <EOL> self . assertEqual ( supertask . cluster . status , '<STR_LIT:error>' ) <EOL> self . assertFalse ( supertask . cluster . is_locked ) <EOL> @ fake_tasks ( error = "<STR_LIT>" , error_msg = "<STR_LIT>" ) <EOL> def test_deployment_error_from_orchestrator ( self ) : <EOL> self . env . create ( <EOL> cluster_kwargs = { } , <EOL> nodes_kwargs = [ <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True } , <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : True } , <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : True } <EOL> ] <EOL> ) <EOL> supertask = self . env . launch_deployment ( ) <EOL> err_msg = "<STR_LIT>" <EOL> self . assertEqual ( supertask . status , consts . TASK_STATUSES . error ) <EOL> self . assertEqual ( supertask . message , err_msg ) <EOL> self . assertIsNotNone ( <EOL> self . db . query ( Notification ) . filter_by ( message = err_msg ) . first ( ) <EOL> ) <EOL> self . env . refresh_nodes ( ) <EOL> self . env . refresh_clusters ( ) <EOL> failed_nodes = [ <EOL> n . id for n in self . env . nodes <EOL> if n . status == '<STR_LIT:error>' and n . error_type == '<STR_LIT>' <EOL> ] <EOL> self . assertGreater ( len ( failed_nodes ) , <NUM_LIT:0> ) <EOL> err_notification = self . db . query ( Notification ) . filter ( <EOL> Notification . node_id . in_ ( failed_nodes ) <EOL> ) . first ( ) <EOL> self . assertIsNotNone ( err_notification ) <EOL> self . assertEqual ( supertask . cluster . status , '<STR_LIT:error>' ) <EOL> self . assertFalse ( supertask . cluster . is_locked ) <EOL> @ fake_tasks ( error = "<STR_LIT>" ) <EOL> def test_deployment_error_during_deployment ( self ) : <EOL> self . env . create ( <EOL> cluster_kwargs = { } , <EOL> nodes_kwargs = [ <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True } , <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : True } ] ) <EOL> supertask = self . env . launch_deployment ( ) <EOL> message = re . compile ( <EOL> "<STR_LIT>" <EOL> ) <EOL> self . assertEqual ( supertask . status , consts . TASK_STATUSES . error ) <EOL> self . assertRegexpMatches ( supertask . message , message ) <EOL> self . env . refresh_nodes ( ) <EOL> self . env . refresh_clusters ( ) <EOL> n_error = lambda n : ( n . status , n . error_type ) == ( '<STR_LIT:error>' , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( map ( n_error , self . env . nodes ) ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( supertask . cluster . status , '<STR_LIT:error>' ) <EOL> self . assertFalse ( supertask . cluster . is_locked ) <EOL> @ fake_tasks ( error = "<STR_LIT>" , task_ready = True ) <EOL> def test_task_ready_node_error ( self ) : <EOL> self . env . create ( <EOL> cluster_kwargs = { } , <EOL> nodes_kwargs = [ <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True } , <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : True } <EOL> ] <EOL> ) <EOL> supertask = self . env . launch_deployment ( ) <EOL> message = re . compile ( <EOL> "<STR_LIT>" <EOL> ) <EOL> self . assertEqual ( supertask . status , consts . TASK_STATUSES . error ) <EOL> self . assertRegexpMatches ( supertask . message , message ) <EOL> self . assertFalse ( supertask . cluster . is_locked ) </s>
