"""Unit tests for metric model.""" import datetime from oto import status import pytest from social_analytics.models import metric from tests.utils import db_snowflake @pytest.fixture def db_fixture(): """Drop and re-create all the SQLite tables and seed them.""" db_snowflake.create_tables() db_snowflake.seed_data() @pytest.fixture(params=[ { 'platform_id': 11872, 'metric_id': 2, 'start_date': datetime.datetime(2017, 9, 22, 11, 19, 15), 'end_date': datetime.datetime(2017, 9, 24, 11, 19, 15) }, { 'social_profile_id': 2, 'metric_id': 2, 'start_date': datetime.datetime(2017, 9, 22, 11, 19, 15), 'end_date': datetime.datetime(2017, 9, 24, 11, 19, 15) }, { 'social_profile_id': 2, 'platform_id': 11872, 'start_date': datetime.datetime(2017, 9, 22, 11, 19, 15), 'end_date': datetime.datetime(2017, 9, 24, 11, 19, 15) }, { 'social_profile_id': 2, 'platform_id': 11872, 'metric_id': 2, 'end_date': datetime.datetime(2017, 9, 24, 11, 19, 15) }, { 'social_profile_id': 2, 'platform_id': 11872, 'metric_id': 2, 'start_date': datetime.datetime(2017, 9, 22, 11, 19, 15) } ]) def bad_params(request): """Fixture for metric time series data bad params.""" return request.param @pytest.fixture def time_series_query_params(): """Fixture for metric time series data.""" return { 'social_profile_id': 2, 'platform_id': 1, 'metric_id': 2, 'start_date': datetime.datetime(2017, 9, 22, 11, 19, 15), 'end_date': datetime.datetime(2017, 9, 24, 11, 19, 15) } @pytest.fixture def time_series_query_params_not_found(): """Fixture for metric time series data not found.""" return { 'social_profile_id': 12, 'platform_id': 1, 'metric_id': 2, 'start_date': datetime.datetime(2017, 9, 22, 11, 19, 15), 'end_date': datetime.datetime(2017, 9, 24, 11, 19, 15) } def test_get_metrics_by_social_profile_id(db_fixture): """Test get metrics by social profile id.""" social_profile_id = 5 result = metric.get_metrics_by_social_profile_id(social_profile_id) assert result metrics = result.message assert len(metrics['items']) == 2 assert metrics['items'][0]['social_profile_id'] == 5 def test_get_metrics_by_social_profile_id_no_id(): """Test get metrics by social profile id, no social profile id given.""" result = metric.get_metrics_by_social_profile_id(None) assert result.status == status.BAD_REQUEST def test_get_metrics_by_social_profile_id_not_found(db_fixture): """Test get metrics by social profile id not found.""" social_profile_id = 15 result = metric.get_metrics_by_social_profile_id(social_profile_id) assert not result def test_get_time_series_data_for_period(db_fixture, time_series_query_params): """Test get time series data for period.""" result = metric.get_time_series_data_for_period(time_series_query_params) assert result def test_get_time_series_data_for_period_bad_params(bad_params): """Test get time series data for period bad params.""" result = metric.get_time_series_data_for_period(bad_params) assert not result assert result.status == status.BAD_REQUEST def test_get_time_series_data_for_period_not_found( db_fixture, time_series_query_params_not_found): """Test get time series data for period not found.""" result = metric.get_time_series_data_for_period( time_series_query_params_not_found) assert not result def test_get_metrics_by_social_profile_id_platform_id_and_metric_id( db_fixture): """Test get metrics by social profile id, platform id and metric id.""" social_profile_id = 5 platform_id = 1 metric_id = 2 expected_result = [{ 'social_profile_id': 5, 'platform_id': 1, 'label_id': 1, 'metric_id': 2, 'metric_value': 3898077.0, 'processed_datetime': '2017-09-22 11:19:10', 'activity_datetime': '2017-09-22 11:19:10', }, { 'social_profile_id': 5, 'platform_id': 1, 'label_id': 1, 'metric_id': 2, 'metric_value': 3898087.0, 'processed_datetime': '2017-09-22 11:19:12', 'activity_datetime': '2017-09-22 11:19:12', }] result = ( metric.get_metrics_by_social_profile_id_platform_id_and_metric_id( social_profile_id=social_profile_id, platform_id=platform_id, metric_id=metric_id)) assert result metrics = result.message['items'] assert len(metrics) == 2 assert metrics == expected_result def test_get_metrics_by_social_profile_id_platform_id_and_metric_id_not_found( db_fixture): """Test get metrics by social profile id, platform id and metric id.""" social_profile_id = 1 platform_id = 11872 metric_id = 2 result = ( metric.get_metrics_by_social_profile_id_platform_id_and_metric_id( social_profile_id=social_profile_id, platform_id=platform_id, metric_id=metric_id)) assert not result def test_get_metrics_by_social_profile_id_platform_id_and_metric_id_bad_params( db_fixture): """Test get metrics by social profile id, platform id and metric id.""" social_profile_id = None platform_id = 11872 metric_id = 2 result = ( metric.get_metrics_by_social_profile_id_platform_id_and_metric_id( social_profile_id=social_profile_id, platform_id=platform_id, metric_id=metric_id)) assert not result assert result.status == status.BAD_REQUEST def test_get_latest_metric_by_social_profile_id(db_fixture): """Test get total followers by social profile ids.""" social_profile_id = db_snowflake.facts[2]['social_profile_id'] expected = { 'social_profile_id': 3, 'platform': 'facebook', 'metric': 'fan_count', 'metric_value': 4598087.0 } result = metric.get_latest_metric_by_social_profile_id(social_profile_id) assert result assert result.message == expected def test_get_latest_metric_by_social_profile_id_no_id(): """Test get total followers by social profile ids no ids.""" social_profile_ids = [] result = metric.get_latest_metric_by_social_profile_id(social_profile_ids) assert not result assert result.status == status.BAD_REQUEST def test_get_latest_metric_by_social_profile_id_not_found(db_fixture): """Test get total followers by social profile ids.""" social_profile_id = 15 result = metric.get_latest_metric_by_social_profile_id(social_profile_id) assert not result assert result.status == status.NOT_FOUND