"""Unit tests for conflict_csv_to_json.py.""" import io import json import pytest from yt_conflict_elasticsearch.flows.elasticsearch_export \ import conflict_csv_to_json @pytest.fixture def csv_columns(): """Test csv columns.""" return [ 'conflict_id', 'isrc', 'tuid', 'vendor_id', 'subaccount_id', 'subaccount_name', 'territory', 'territory_standard', 'conflicting_owner', 'conflict_date', 'resolved_datetime', 'daily_average_views', 'track_name', 'version', 'artist_names', 'product_id', 'product_name', 'display_upc', 'status', 'es_id', 'product_delivered_version', 'views_in_conflict' ] @pytest.fixture def conflict_dict_fields(): """Test conflict dictionary fields.""" return [ 'conflict_id', 'isrc', 'tuid', 'vendor_id', 'subaccount_id', 'subaccount_name', 'territory', 'territory_standard', 'conflicting_owner', 'conflict_date', 'resolved_datetime', 'daily_average_views', 'track_name', 'product_id', 'product_name', 'display_upc', 'status', 'track_artists', 'es_id', 'views_in_conflict' ] def test_process_csv_file(mocker, save_gzipped_file_with_csv_conflicts, territories_success_response): """Test process csv to file.""" mocker.patch( 'yt_conflict_elasticsearch.flows.elasticsearch_export.models.' 'ows_territories.request.process', return_value=territories_success_response) test_fp = io.StringIO() num_conflicts = conflict_csv_to_json.process_csv_file( 'tests/test_files/yt_conflict_new.csv.gz', test_fp) test_fp.seek(0) conflicts = [json.loads(ln) for ln in test_fp.readlines()] assert num_conflicts == 2 assert conflicts == [ { 'territory_standard': 'ISO_3166_1_2016', 'conflict_date': '2013-09-04', 'resolved_datetime': None, 'tuid': 8449, 'isrc': 'USA370330320', 'product_id': 12112, 'conflicting_owner': 'adrevuppm', 'track_name': 'Day By Day', 'vendor_id': 2, 'track_artists': ['Dave Bush & Food'], 'display_upc': '669910333228', 'territories': [ { 'conflict_id': 4971125, 'code': 'EC', 'continent_name': 'South America', 'name': 'Ecuador' }, { 'conflict_id': 5033762, 'code': 'AR', 'continent_name': 'South America', 'name': 'Argentina' } ], 'product_name': 'The Big One', 'subaccount_name': '', 'subaccount_id': None, 'daily_average_views': 0, 'status': 'NEW', 'es_id': 'dDzkgWcB9u7hUka8A72K', 'views_in_conflict': 2 }, { 'territory_standard': 'ISO_3166_1_2016', 'conflict_date': '2017-04-26', 'resolved_datetime': None, 'tuid': 111911, 'isrc': 'USA370376686', 'product_id': 24434, 'conflicting_owner': 'rumblefish_cdbaby', 'track_name': 'Amazing Grace (remix)', 'vendor_id': 36, 'track_artists': ['Just Plain Darin', 'Mama Mia'], 'display_upc': '669910788622', 'territories': [ { 'conflict_id': 31487132, 'code': 'BQ', 'continent_name': 'North America', 'name': 'Bonaire' } ], 'product_name': 'Back To The Basics (Live)', 'subaccount_name': 'Hot Potato Productions', 'subaccount_id': 15, 'daily_average_views': 0, 'status': 'NEW', 'es_id': '', 'views_in_conflict': 2 }] @pytest.mark.parametrize( 'data, expected_result', [ (['4971125', 'USA370330320', '8449', '2', '\\\\N', '\\N', 'EC', 'ISO_3166_1_2016', 'adrevuppm', '2013-09-04', '\\N', '0', 'Day By Day', '', 'Dave Bush & Food', '12112', 'The Big One', '669910333228', 'NEW', 'dDzkgWcB9u7hUka8A72K', '\\N', 2], {'territory': 'EC', 'resolved_datetime': None, 'isrc': 'USA370330320', 'display_upc': '669910333228', 'subaccount_id': None, 'subaccount_name': '', 'product_name': 'The Big One', 'tuid': 8449, 'daily_average_views': 0, 'vendor_id': 2, 'conflict_id': 4971125, 'territory_standard': 'ISO_3166_1_2016', 'conflict_date': '2013-09-04', 'product_id': 12112, 'track_name': 'Day By Day', 'track_artists': ['Dave Bush & Food'], 'conflicting_owner': 'adrevuppm', 'status': 'NEW', 'es_id': 'dDzkgWcB9u7hUka8A72K', 'views_in_conflict': 2}), (['4971125', 'USA370330320', '8449', '2', '\\\\N', '\\N', 'EC', 'ISO_3166_1_2016', 'adrevuppm', '2013-09-04', '\\N', '0', 'Day By Day', '\\N', 'Dave Bush & Food', '12112', 'The Big One', '669910333228', 'NEW', 'dDzkgWcB9u7hUka8A72K', '\\N', 2], {'territory': 'EC', 'resolved_datetime': None, 'isrc': 'USA370330320', 'display_upc': '669910333228', 'subaccount_id': None, 'subaccount_name': '', 'product_name': 'The Big One', 'tuid': 8449, 'daily_average_views': 0, 'vendor_id': 2, 'conflict_id': 4971125, 'territory_standard': 'ISO_3166_1_2016', 'conflict_date': '2013-09-04', 'product_id': 12112, 'track_name': 'Day By Day', 'track_artists': ['Dave Bush & Food'], 'conflicting_owner': 'adrevuppm', 'status': 'NEW', 'es_id': 'dDzkgWcB9u7hUka8A72K', 'views_in_conflict': 2})]) def test_convert_row_to_dict_no_version( data, expected_result, csv_columns, conflict_dict_fields): """.Tests convert row to dict output.""" result = conflict_csv_to_json._convert_row_to_dict(data, csv_columns) assert result for field_name in conflict_dict_fields: assert result[field_name] == expected_result[field_name] def test_convert_row_to_dict_with_version(csv_columns): """Test convert row updates track_name with version.""" data = [ '4971125', 'USA370330320', '8449', '2', '\\\\N', '\\N', 'EC', 'ISO_3166_1_2016', 'adrevuppm', '2013-09-04', '\\N', '0', 'Day By Day', 'remix', 'Dave Bush & Food', '12112', 'The Big One', '669910333228', 'NEW', 'dDzkgWcB9u7hUka8A72K', '\\N', 2] result = conflict_csv_to_json._convert_row_to_dict(data, csv_columns) assert result['track_name'] == 'Day By Day (remix)' assert 'version' not in result def test_convert_row_to_dict_without_version(csv_columns): """Test convert row updates track_name without version.""" data = [ '4971125', 'USA370330320', '8449', '2', '\\\\N', '\\N', 'EC', 'ISO_3166_1_2016', 'adrevuppm', '2013-09-04', '\\N', '0', 'Day By Day', 'remix', 'Dave Bush & Food', '12112', 'The Big One', '669910333228', 'NEW', 'dDzkgWcB9u7hUka8A72K', 'Live', 2] result = conflict_csv_to_json._convert_row_to_dict(data, csv_columns) assert result['product_name'] == 'The Big One (Live)' assert 'product_display_version' not in result @pytest.mark.parametrize( 'value,result', [ ('0', 0), ('1', 1), ('NULL', None), ('\\N', None), ('\\\\N', None) ]) def test_to_int_valid_values(value, result): """Verify valid values are parsed correctly.""" assert conflict_csv_to_json._to_int(value) == result @pytest.mark.parametrize( 'value,result', [ ('2013-09-04', '2013-09-04'), ('NULL', None), ('\\N', None), ('\\\\N', None) ]) def test_to_date_valid_values(value, result): """Verify valid values are parsed correctly.""" assert conflict_csv_to_json._to_date(value) == result @pytest.mark.parametrize( 'value,result', [ ('Wed, 10 Jan 2018 16:07:14 Z', '2018-01-10T16:07:14'), ('NULL', None), ('\\N', None), ('\\\\N', None) ]) def test_to_datetime_valid_values(value, result): """Verify valid values are parsed correctly.""" assert conflict_csv_to_json._to_datetime(value) == result