<s> """<STR_LIT>""" <EOL> import unittest <EOL> from mock import Mock , patch <EOL> import random <EOL> import falcon <EOL> from . common import * <EOL> from freezer_api . common . exceptions import * <EOL> from freezer_api . api . v1 import sessions as v1_sessions <EOL> class TestSessionsCollectionResource ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mock_db = Mock ( ) <EOL> self . mock_req = Mock ( ) <EOL> self . mock_req . get_header . return_value = fake_session_0 [ '<STR_LIT>' ] <EOL> self . mock_req . status = falcon . HTTP_200 <EOL> self . resource = v1_sessions . SessionsCollectionResource ( self . mock_db ) <EOL> self . mock_json_body = Mock ( ) <EOL> self . mock_json_body . return_value = { } <EOL> self . resource . json_body = self . mock_json_body <EOL> def test_on_get_return_empty_list ( self ) : <EOL> self . mock_db . search_session . return_value = [ ] <EOL> expected_result = { '<STR_LIT>' : [ ] } <EOL> self . resource . on_get ( self . mock_req , self . mock_req ) <EOL> result = self . mock_req . body <EOL> self . assertEqual ( result , expected_result ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_200 ) <EOL> def test_on_get_return_correct_list ( self ) : <EOL> self . mock_db . search_session . return_value = [ get_fake_session_0 ( ) , get_fake_session_1 ( ) ] <EOL> expected_result = { '<STR_LIT>' : [ get_fake_session_0 ( ) , get_fake_session_1 ( ) ] } <EOL> self . resource . on_get ( self . mock_req , self . mock_req ) <EOL> result = self . mock_req . body <EOL> self . assertEqual ( result , expected_result ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_200 ) <EOL> def test_on_post_raises_when_missing_body ( self ) : <EOL> self . mock_db . add_session . return_value = fake_session_0 [ '<STR_LIT>' ] <EOL> self . assertRaises ( BadDataFormat , self . resource . on_post , self . mock_req , self . mock_req ) <EOL> def test_on_post_inserts_correct_data ( self ) : <EOL> session = get_fake_session_0 ( ) <EOL> self . mock_json_body . return_value = session <EOL> self . mock_db . add_session . return_value = '<STR_LIT>' <EOL> expected_result = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . resource . on_post ( self . mock_req , self . mock_req ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_201 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> class TestSessionsResource ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mock_db = Mock ( ) <EOL> self . mock_req = Mock ( ) <EOL> self . mock_req . get_header . return_value = fake_session_0 [ '<STR_LIT>' ] <EOL> self . mock_req . status = falcon . HTTP_200 <EOL> self . resource = v1_sessions . SessionsResource ( self . mock_db ) <EOL> self . mock_json_body = Mock ( ) <EOL> self . mock_json_body . return_value = { } <EOL> self . resource . json_body = self . mock_json_body <EOL> def test_create_resource ( self ) : <EOL> self . assertIsInstance ( self . resource , v1_sessions . SessionsResource ) <EOL> def test_on_get_return_no_result_and_404_when_not_found ( self ) : <EOL> self . mock_db . get_session . return_value = None <EOL> self . mock_req . body = None <EOL> self . resource . on_get ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertIsNone ( self . mock_req . body ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_404 ) <EOL> def test_on_get_return_correct_data ( self ) : <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . resource . on_get ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> result = self . mock_req . body <EOL> self . assertEqual ( result , get_fake_session_0 ( ) ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_200 ) <EOL> def test_on_delete_removes_proper_data ( self ) : <EOL> self . resource . on_delete ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> result = self . mock_req . body <EOL> expected_result = { '<STR_LIT>' : fake_session_0 [ '<STR_LIT>' ] } <EOL> self . assertEquals ( self . mock_req . status , falcon . HTTP_204 ) <EOL> self . assertEqual ( result , expected_result ) <EOL> def test_on_patch_ok_with_some_fields ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> patch_doc = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . mock_json_body . return_value = patch_doc <EOL> expected_patch = patch_doc . copy ( ) <EOL> expected_result = { '<STR_LIT>' : fake_session_0 [ '<STR_LIT>' ] , <EOL> '<STR_LIT:version>' : new_version } <EOL> self . resource . on_patch ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . mock_db . update_session . assert_called_with ( <EOL> user_id = fake_session_0 [ '<STR_LIT>' ] , <EOL> session_id = fake_session_0 [ '<STR_LIT>' ] , <EOL> patch_doc = patch_doc ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_200 ) <EOL> result = self . mock_req . body <EOL> self . assertEqual ( result , expected_result ) <EOL> def test_on_post_ok ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . replace_session . return_value = new_version <EOL> session = get_fake_session_0 ( ) <EOL> self . mock_json_body . return_value = session <EOL> expected_result = { '<STR_LIT>' : fake_session_0 [ '<STR_LIT>' ] , <EOL> '<STR_LIT:version>' : new_version } <EOL> self . resource . on_post ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_201 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> def test_on_post_raises_when_db_replace_session_raises ( self ) : <EOL> self . mock_db . replace_session . side_effect = AccessForbidden ( '<STR_LIT>' ) <EOL> session = get_fake_session_0 ( ) <EOL> self . mock_json_body . return_value = session <EOL> self . assertRaises ( AccessForbidden , self . resource . on_post , <EOL> self . mock_req , <EOL> self . mock_req , <EOL> fake_session_0 [ '<STR_LIT>' ] ) <EOL> class TestSessionsAction ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mock_db = Mock ( ) <EOL> self . mock_req = Mock ( ) <EOL> self . mock_req . get_header . return_value = fake_session_0 [ '<STR_LIT>' ] <EOL> self . mock_req . status = falcon . HTTP_200 <EOL> self . resource = v1_sessions . SessionsAction ( self . mock_db ) <EOL> self . mock_json_body = Mock ( ) <EOL> self . mock_json_body . return_value = { } <EOL> self . resource . json_body = self . mock_json_body <EOL> def test_create_resource ( self ) : <EOL> self . assertIsInstance ( self . resource , v1_sessions . SessionsAction ) <EOL> def test_on_post_raises_when_unable_to_read_action_from_body ( self ) : <EOL> self . mock_json_body . return_value = { } <EOL> self . assertRaises ( BadDataFormat , self . resource . on_post , <EOL> self . mock_req , <EOL> self . mock_req , <EOL> fake_session_0 [ '<STR_LIT>' ] ) <EOL> def test_on_post_start_action_ok ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:start>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:5> <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> expected_result = { '<STR_LIT:result>' : '<STR_LIT:success>' , <EOL> '<STR_LIT>' : <NUM_LIT:6> } <EOL> self . resource . on_post ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_202 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> def test_on_post_start_action_raises_BadDataFormat_when_job_not_in_session ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:start>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:5> <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> self . assertRaises ( BadDataFormat , self . resource . on_post , self . mock_req , <EOL> self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> def test_on_post_start_action_raises_BadDataFormat_when_current_tag_too_high ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:start>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:6> <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> self . assertRaises ( BadDataFormat , self . resource . on_post , self . mock_req , <EOL> self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> def test_on_post_end_action_ok ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:end>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT:result>" : "<STR_LIT:success>" <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> expected_result = { '<STR_LIT:result>' : '<STR_LIT:success>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> } <EOL> self . resource . on_post ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_202 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> def test_on_post_end_action_raises_BadDataFormat_when_job_not_in_session ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:end>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT:result>" : "<STR_LIT:success>" <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> self . assertRaises ( BadDataFormat , self . resource . on_post , self . mock_req , <EOL> self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> def test_on_post_raises_MethodNotImplemented_when_methon_not_implemented ( self ) : <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . mock_db . get_session . return_value = get_fake_session_0 ( ) <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:5> , <EOL> "<STR_LIT:result>" : "<STR_LIT:success>" <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> self . assertRaises ( MethodNotImplemented , self . resource . on_post , self . mock_req , <EOL> self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_on_post_start_succeeds_in_holdoff_if_tag_needs_not_increment ( self , mock_time ) : <EOL> mock_time . time . return_value = <NUM_LIT:1000> <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> session_doc = get_fake_session_0 ( ) <EOL> session_doc [ '<STR_LIT>' ] = <NUM_LIT> <EOL> self . mock_db . get_session . return_value = session_doc <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:start>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:4> <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> expected_result = { '<STR_LIT:result>' : '<STR_LIT:success>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> } <EOL> self . resource . on_post ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_202 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_on_post_start_replies_holdoff_if_tag_would_increment ( self , mock_time ) : <EOL> mock_time . time . return_value = <NUM_LIT:1000> <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> session_doc = get_fake_session_0 ( ) <EOL> session_doc [ '<STR_LIT>' ] = <NUM_LIT> <EOL> self . mock_db . get_session . return_value = session_doc <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:start>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:5> <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> expected_result = { '<STR_LIT:result>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> } <EOL> self . resource . on_post ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_202 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_on_post_start_outofholdoff_replies_outofsync_when_tag_too_low ( self , mock_time ) : <EOL> mock_time . time . return_value = <NUM_LIT> <EOL> new_version = random . randint ( <NUM_LIT:0> , <NUM_LIT> ) <EOL> session_doc = get_fake_session_0 ( ) <EOL> session_doc [ '<STR_LIT>' ] = <NUM_LIT> <EOL> self . mock_db . get_session . return_value = session_doc <EOL> self . mock_db . update_session . return_value = new_version <EOL> action = { "<STR_LIT:start>" : { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> "<STR_LIT>" : <NUM_LIT:2> <EOL> } } <EOL> self . mock_json_body . return_value = action <EOL> expected_result = { '<STR_LIT:result>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:5> } <EOL> self . resource . on_post ( self . mock_req , self . mock_req , fake_session_0 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( self . mock_req . status , falcon . HTTP_202 ) <EOL> self . assertEqual ( self . mock_req . body , expected_result ) <EOL> class TestSessions ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . session_doc = { } <EOL> self . session = v1_sessions . Session ( self . session_doc ) <EOL> def test_create_resource ( self ) : <EOL> self . assertIsInstance ( self . session , v1_sessions . Session ) <EOL> def test_overall_result_running ( self ) : <EOL> self . session_doc [ '<STR_LIT>' ] = { '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : '<STR_LIT:success>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : '<STR_LIT>' } } <EOL> res = self . session . get_job_overall_result ( ) <EOL> self . assertEquals ( res , '<STR_LIT>' ) <EOL> def test_overall_result_fail ( self ) : <EOL> self . session_doc [ '<STR_LIT>' ] = { '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : '<STR_LIT:success>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : '<STR_LIT>' } } <EOL> res = self . session . get_job_overall_result ( ) <EOL> self . assertEquals ( res , '<STR_LIT>' ) <EOL> def test_overall_result_success ( self ) : <EOL> self . session_doc [ '<STR_LIT>' ] = { '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : '<STR_LIT:success>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : '<STR_LIT:success>' } } <EOL> res = self . session . get_job_overall_result ( ) <EOL> self . assertEquals ( res , '<STR_LIT:success>' ) <EOL> class TestSessionsJobs ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mock_db = Mock ( ) <EOL> self . mock_req = Mock ( ) <EOL> self . mock_req . get_header . return_value = fake_session_0 [ '<STR_LIT>' ] <EOL> self . mock_req . status = falcon . HTTP_200 <EOL> self . resource = v1_sessions . SessionsJob ( self . mock_db ) <EOL> def test_create_resource ( self ) : <EOL> self . assertIsInstance ( self . resource , v1_sessions . SessionsJob ) <EOL> def test_on_put_adds_job_to_session_jobs ( self ) : <EOL> session = get_fake_session_0 ( ) <EOL> job = get_fake_job_0 ( ) <EOL> job_info = { job [ '<STR_LIT>' ] : { '<STR_LIT>' : job [ '<STR_LIT>' ] , <EOL> '<STR_LIT:status>' : job [ '<STR_LIT>' ] [ '<STR_LIT:status>' ] , <EOL> '<STR_LIT:result>' : job [ '<STR_LIT>' ] [ '<STR_LIT:result>' ] , <EOL> '<STR_LIT>' : job [ '<STR_LIT>' ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : job [ '<STR_LIT>' ] [ '<STR_LIT>' ] } } <EOL> session_update_doc = { '<STR_LIT>' : job_info } <EOL> self . mock_db . get_session . return_value = session <EOL> self . mock_db . get_job . return_value = job <EOL> self . resource . on_put ( self . mock_req , self . mock_req , <EOL> session [ '<STR_LIT>' ] , <EOL> job [ '<STR_LIT>' ] ) <EOL> self . mock_db . update_session . assert_called_with ( user_id = session [ '<STR_LIT>' ] , <EOL> session_id = session [ '<STR_LIT>' ] , <EOL> patch_doc = session_update_doc ) <EOL> def test_on_put_updates_job_with_session_info ( self ) : <EOL> session = get_fake_session_0 ( ) <EOL> job = get_fake_job_0 ( ) <EOL> self . mock_db . get_session . return_value = session <EOL> self . mock_db . get_job . return_value = job <EOL> job_update_doc = { <EOL> '<STR_LIT>' : session [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : session [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : session [ '<STR_LIT>' ] <EOL> } <EOL> self . resource . on_put ( self . mock_req , self . mock_req , <EOL> session [ '<STR_LIT>' ] , <EOL> job [ '<STR_LIT>' ] ) <EOL> self . mock_db . update_job . assert_called_with ( user_id = session [ '<STR_LIT>' ] , <EOL> job_id = job [ '<STR_LIT>' ] , <EOL> patch_doc = job_update_doc ) <EOL> def test_on_delete_removes_job_from_session_jobs ( self ) : <EOL> session = get_fake_session_0 ( ) <EOL> updated_session = get_fake_session_1 ( ) <EOL> job = get_fake_job_0 ( ) <EOL> self . mock_db . get_session . return_value = session <EOL> self . mock_db . get_job . return_value = job <EOL> self . resource . on_delete ( self . mock_req , self . mock_req , <EOL> session [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> self . mock_db . replace_session . assert_called_with ( user_id = session [ '<STR_LIT>' ] , <EOL> session_id = session [ '<STR_LIT>' ] , <EOL> doc = updated_session ) <EOL> def test_on_delete_removes_session_info_from_job_and_stops_job ( self ) : <EOL> session = get_fake_session_0 ( ) <EOL> job = get_fake_job_0 ( ) <EOL> self . mock_db . get_session . return_value = session <EOL> self . mock_db . get_job . return_value = job <EOL> job_update_doc = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> self . resource . on_delete ( self . mock_req , self . mock_req , <EOL> session [ '<STR_LIT>' ] , <EOL> job [ '<STR_LIT>' ] ) <EOL> self . mock_db . update_job . assert_called_with ( user_id = session [ '<STR_LIT>' ] , <EOL> job_id = job [ '<STR_LIT>' ] , <EOL> patch_doc = job_update_doc ) </s>
