<s> import unittest <EOL> from performanceplatform . collector . webtrends . reports import ( Collector , <EOL> V3Parser , <EOL> V2Parser ) <EOL> import performanceplatform <EOL> import json <EOL> from mock import patch , call <EOL> import requests <EOL> from hamcrest import assert_that , equal_to , has_entries <EOL> from nose . tools import assert_raises <EOL> import datetime <EOL> import pytz <EOL> from tests . performanceplatform . collector . ga import dt <EOL> def build_collector ( start_at = None , <EOL> end_at = None , <EOL> query = None ) : <EOL> if not query : <EOL> query = { '<STR_LIT>' : '<STR_LIT>' } <EOL> credentials = { <EOL> '<STR_LIT:user>' : '<STR_LIT:abc>' , <EOL> '<STR_LIT:password>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> return Collector ( credentials , query , start_at , end_at ) <EOL> def get_fake_response ( file = '<STR_LIT>' ) : <EOL> with open ( "<STR_LIT>" . format ( file ) , "<STR_LIT:r>" ) as f : <EOL> return json . loads ( f . read ( ) ) <EOL> class TestCollector ( unittest . TestCase ) : <EOL> @ patch . object ( <EOL> performanceplatform . client . DataSet , "<STR_LIT>" ) <EOL> @ patch ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_collect_parse_and_push ( self , mock_get , mock_post ) : <EOL> mock_get ( ) . json . return_value = get_fake_response ( ) <EOL> query = { '<STR_LIT>' : '<STR_LIT>' } <EOL> options = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> collector = build_collector ( query = query ) <EOL> collector . collect_parse_and_push ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:url>' : '<STR_LIT:abc>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False } , <EOL> options ) <EOL> posted_data = [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : datetime . datetime ( <EOL> <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , tzinfo = pytz . UTC ) , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT:test>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : datetime . datetime ( <EOL> <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , tzinfo = pytz . UTC ) , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:test>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> mock_post . assert_called_once_with ( posted_data , chunk_size = <NUM_LIT:100> ) <EOL> @ patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_collect_when_specified_start_and_end_and_daily ( self , mock_get ) : <EOL> mock_get ( ) . json . return_value = get_fake_response ( ) <EOL> mock_get . reset_mock ( ) <EOL> collector = build_collector ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> assert_that ( collector . collect ( ) , equal_to ( [ <EOL> get_fake_response ( ) [ "<STR_LIT:data>" ] , <EOL> get_fake_response ( ) [ "<STR_LIT:data>" ] , <EOL> get_fake_response ( ) [ "<STR_LIT:data>" ] ] ) ) <EOL> calls = [ <EOL> call ( <EOL> url = "<STR_LIT>" , <EOL> auth = ( '<STR_LIT:abc>' , '<STR_LIT>' ) , <EOL> params = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ) , <EOL> call ( <EOL> url = "<STR_LIT>" , <EOL> auth = ( '<STR_LIT:abc>' , '<STR_LIT>' ) , <EOL> params = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ) , <EOL> call ( <EOL> url = "<STR_LIT>" , <EOL> auth = ( '<STR_LIT:abc>' , '<STR_LIT>' ) , <EOL> params = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> ] <EOL> assert_that ( mock_get . call_args_list [ <NUM_LIT:0> ] , equal_to ( calls [ <NUM_LIT:0> ] ) ) <EOL> assert_that ( mock_get . call_args_list [ <NUM_LIT:1> ] , equal_to ( calls [ <NUM_LIT:1> ] ) ) <EOL> assert_that ( mock_get . call_args_list [ <NUM_LIT:2> ] , equal_to ( calls [ <NUM_LIT:2> ] ) ) <EOL> @ patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_collect_when_no_specified_start_and_end ( self , mock_get ) : <EOL> mock_get ( ) . json . return_value = get_fake_response ( ) <EOL> mock_get . reset_mock ( ) <EOL> collector = build_collector ( ) <EOL> assert_that ( <EOL> collector . collect ( ) , <EOL> equal_to ( [ get_fake_response ( ) [ "<STR_LIT:data>" ] ] ) ) <EOL> mock_get . assert_called_once_with ( <EOL> url = "<STR_LIT>" , <EOL> auth = ( '<STR_LIT:abc>' , '<STR_LIT>' ) , <EOL> params = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> ) <EOL> @ patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_collect_correctly_raises_for_status ( self , mock_get ) : <EOL> response = requests . Response ( ) <EOL> response . status_code = <NUM_LIT> <EOL> mock_get . return_value = response <EOL> collector = build_collector ( ) <EOL> assert_raises ( requests . exceptions . HTTPError , collector . collect ) <EOL> def test_collect_parses_start_and_end_date_format_correctly ( self ) : <EOL> date = Collector . parse_standard_date_string_to_date ( "<STR_LIT>" ) <EOL> assert_that ( <EOL> Collector . parse_date_for_query ( date ) , equal_to ( "<STR_LIT>" ) ) <EOL> date = Collector . parse_standard_date_string_to_date ( "<STR_LIT>" ) <EOL> assert_that ( <EOL> Collector . parse_date_for_query ( date ) , equal_to ( "<STR_LIT>" ) ) <EOL> def test_parse_standard_date_string_to_date_leaves_datetimes ( self ) : <EOL> date = Collector . parse_standard_date_string_to_date ( "<STR_LIT>" ) <EOL> date_unchanged = Collector . parse_standard_date_string_to_date ( date ) <EOL> assert_that ( date_unchanged , equal_to ( date ) ) <EOL> assert_that ( type ( date ) , datetime . datetime ) <EOL> def test_collect_builds_date_ranges_correctly_when_start_and_end_given ( <EOL> self ) : <EOL> assert_that ( Collector . date_range_for_webtrends ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ) , <EOL> equal_to ( [ <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> ] ) <EOL> ) <EOL> def test_collect_builds_date_ranges_correctly_when_no_start_and_end ( self ) : <EOL> assert_that ( <EOL> Collector . date_range_for_webtrends ( ) , <EOL> equal_to ( [ <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> ] ) <EOL> ) <EOL> class TestParser ( unittest . TestCase ) : <EOL> def test_handles_returned_date_format_correctly_when_v2 ( self ) : <EOL> assert_that ( <EOL> V2Parser . to_datetime ( <EOL> "<STR_LIT>" ) , <EOL> equal_to ( dt ( <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" ) ) ) <EOL> def test_handles_no_data_in_period_when_v2 ( self ) : <EOL> options = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> data_type = "<STR_LIT>" <EOL> parser = V2Parser ( options , data_type ) <EOL> no_data_response = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:0.0> <EOL> } <EOL> } <EOL> } <EOL> results = list ( parser . parse ( [ no_data_response ] ) ) <EOL> assert_that ( results , equal_to ( [ ] ) ) <EOL> def test_parses_data_correctly_when_v2 ( self ) : <EOL> posted_data = [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : datetime . datetime ( <EOL> <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , tzinfo = pytz . UTC ) , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT:test>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : datetime . datetime ( <EOL> <NUM_LIT> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , tzinfo = pytz . UTC ) , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:test>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> options = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> data_type = "<STR_LIT>" <EOL> parser = V2Parser ( options , data_type ) <EOL> results = list ( parser . parse ( [ get_fake_response ( ) [ '<STR_LIT:data>' ] ] ) ) <EOL> assert_that ( results [ <NUM_LIT:0> ] , has_entries ( posted_data [ <NUM_LIT:0> ] ) ) <EOL> assert_that ( results [ <NUM_LIT:1> ] , has_entries ( posted_data [ <NUM_LIT:1> ] ) ) <EOL> def test_handles_returned_date_format_correctly_when_v3 ( self ) : <EOL> date_data = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> assert_that ( <EOL> V3Parser . to_datetime ( <EOL> date_data ) , <EOL> equal_to ( dt ( <NUM_LIT> , <NUM_LIT:0> <NUM_LIT:3> , <NUM_LIT:0> <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" ) ) ) <EOL> def test_handles_full_returned_date_format_correctly_when_v3 ( self ) : <EOL> date_data = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> assert_that ( <EOL> V3Parser . to_datetime ( <EOL> date_data ) , <EOL> equal_to ( dt ( <NUM_LIT> , <NUM_LIT:0> <NUM_LIT:3> , <NUM_LIT:0> <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" ) ) ) <EOL> def test_handles_no_data_in_period_when_v3 ( self ) : <EOL> options = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> data_type = "<STR_LIT>" <EOL> parser = V3Parser ( options , data_type ) <EOL> no_data_response = [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> "<STR_LIT>" : <NUM_LIT:0> <EOL> } , <EOL> "<STR_LIT>" : [ ] <EOL> } <EOL> ] <EOL> results = list ( parser . parse ( [ no_data_response ] ) ) <EOL> assert_that ( results , equal_to ( [ ] ) ) <EOL> def test_handles_none_data_in_period_when_v3 ( self ) : <EOL> options = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> data_type = "<STR_LIT>" <EOL> parser = V3Parser ( options , data_type ) <EOL> no_data_response = [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> "<STR_LIT>" : <NUM_LIT:0> <EOL> } , <EOL> "<STR_LIT>" : None <EOL> } <EOL> ] <EOL> results = list ( parser . parse ( [ no_data_response ] ) ) <EOL> assert_that ( results , equal_to ( [ ] ) ) <EOL> def test_parses_data_correctly_when_v3 ( self ) : <EOL> posted_data = [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : datetime . datetime ( <EOL> <NUM_LIT> , <NUM_LIT:0> <NUM_LIT:3> , <NUM_LIT:0> <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , tzinfo = pytz . UTC ) , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:test>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : datetime . datetime ( <EOL> <NUM_LIT> , <NUM_LIT:0> <NUM_LIT:3> , <NUM_LIT:0> <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , tzinfo = pytz . UTC ) , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:test>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> options = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] } <EOL> data_type = "<STR_LIT>" <EOL> parser = V3Parser ( options , data_type ) <EOL> results = list ( parser . parse ( [ get_fake_response ( <EOL> '<STR_LIT>' ) [ '<STR_LIT:data>' ] ] ) ) <EOL> assert_that ( results [ <NUM_LIT:0> ] , has_entries ( posted_data [ <NUM_LIT:0> ] ) ) <EOL> assert_that ( results [ <NUM_LIT:1> ] , has_entries ( posted_data [ <NUM_LIT:1> ] ) ) </s>
