"""Tests ddex utility functions.""" from datetime import datetime, timezone from unittest.mock import patch import pytest from freezegun import freeze_time from soundrecording_utils.ddex.utils import control_chars, primary_track from soundrecording_utils.ddex.utils import rules as utils_rules from soundrecording_utils.metadata.types import ( FingerprintRule, HasFingerprintRules, Label, LabelType, OrchardSoundRecording, Product, Track, TrackConnection, ) def test_remove_control_chars(): """Test removal of control characters from a string.""" input_1 = 'Thư Em \x03' expected_1 = 'Thư Em' input_2 = 'Những Đứa Con Của \x08Mẹ' expected_2 = 'Những Đứa Con Của Mẹ' input_3 = 'Lời Giới \x08Thiệu' expected_3 = 'Lời Giới Thiệu' input_4 = 'Clean String' expected_4 = 'Clean String' assert expected_1 == control_chars.remove_control_chars(input_1) assert expected_2 == control_chars.remove_control_chars(input_2) assert expected_3 == control_chars.remove_control_chars(input_3) assert expected_4 == control_chars.remove_control_chars(input_4) def test_get_primary_track( mock_orchard_sound_recording: OrchardSoundRecording, mock_track: Track ): """Test selection of primary track.""" mock_orchard_sound_recording = mock_orchard_sound_recording._replace( track_connection=TrackConnection( tracks=[ mock_track._replace(name='Track 1', primary=False), mock_track._replace(name='Track 2', primary=True), ] ) ) result = primary_track.get_primary_track( mock_orchard_sound_recording, 'lambda-sr-delivery-tiktok' ) assert result.name == 'Track 2' @patch('soundrecording_utils.ddex.utils.primary_track.get_primary_track_isrc_and_tuid') @patch('soundrecording_utils.ddex.utils.primary_track.get_track') def test_no_tracks_enriched_from_graphql( mock_get_track, mock_get_primary_track_isrc_and_tuid, mock_orchard_sound_recording: OrchardSoundRecording, ): """Test that the no-tracks fallback maps GraphQL data into a Track.""" mock_get_primary_track_isrc_and_tuid.return_value = ('track-isrc-123', 'track-tuid-123') mock_get_track.return_value = { 'trackName': 'Fire', 'labelSoundRecording': { 'label': {'name': 'La Reserve Records'}, }, 'product': { 'subgenre': {'genre': {'name': 'Rock'}}, }, 'participations': [ { 'participant': {'name': 'Artist A', 'uuid': '1234'}, 'participated_as': 'performer', }, ], } osr = mock_orchard_sound_recording._replace( track_connection=TrackConnection(tracks=[]), ) result = primary_track.get_primary_track(osr, 'test-app') mock_get_primary_track_isrc_and_tuid.assert_called_once_with(osr.isrc, 'test-app') mock_get_track.assert_called_once_with('track-isrc-123', 'track-tuid-123', 'test-app') assert result.name == 'Fire' assert result.product.label.name == 'La Reserve Records' assert result.product.subgenre.genre.name == 'Rock' assert len(result.participations) == 1 assert result.participations[0].participant.name == 'Artist A' assert result.participations[0].participated_as == 'performer' @patch('soundrecording_utils.ddex.utils.primary_track.get_primary_track_isrc_and_tuid') @patch('soundrecording_utils.ddex.utils.primary_track.get_track') def test_no_tracks_graphql_returns_none( mock_get_track, mock_get_primary_track_isrc_and_tuid, mock_orchard_sound_recording: OrchardSoundRecording, ): """Test that when get_track returns None a dummy Track with empty fields is returned.""" mock_get_primary_track_isrc_and_tuid.return_value = ('track-isrc-123', 'track-tuid-123') mock_get_track.return_value = None osr = mock_orchard_sound_recording._replace( track_connection=TrackConnection(tracks=[]), ) result = primary_track.get_primary_track(osr, 'test-app') mock_get_primary_track_isrc_and_tuid.assert_called_once_with(osr.isrc, 'test-app') mock_get_track.assert_called_once_with('track-isrc-123', 'track-tuid-123', 'test-app') assert result.name == '' assert result.product.label.name == '' assert result.product.subgenre.genre.name == '' assert result.participations == [] def test_no_primary_tracks( mock_orchard_sound_recording: OrchardSoundRecording, mock_track: Track ): """Test exception on no primary track.""" mock_orchard_sound_recording = mock_orchard_sound_recording._replace( track_connection=TrackConnection( tracks=[ mock_track._replace(name='Track 1', primary=False), mock_track._replace(name='Track 2', primary=False), ] ) ) with pytest.raises(StopIteration): primary_track.get_primary_track( mock_orchard_sound_recording, 'lambda-sr-delivery-tiktok' ) def test_get_fingerprint_rules_by_track_id( mock_orchard_sound_recording: OrchardSoundRecording, mock_track: Track, mock_product: Product, mock_label: Label, mock_vendor_rules: HasFingerprintRules, mock_fingerprint_rule: FingerprintRule, ): """Test formatting/filtering all rules level by tuid.""" t1_vendor_rules = mock_vendor_rules._replace( rules=[ mock_fingerprint_rule._replace( policy='monetize', territory='FR', service='tiktok' ), mock_fingerprint_rule._replace( policy='monetize', territory='US', service='tiktok', active=False ), ] ) t1_subaccount_rules = [ mock_fingerprint_rule._replace( policy='monetize', territory='CA', service='tiktok' ), ] t1_rules = [ mock_fingerprint_rule._replace( policy='carveout', territory='FR', service='tiktok', start_date=None ), ] t2_vendor_rules = [ mock_fingerprint_rule._replace( policy='monetize', territory='FR', service='tiktok' ), mock_fingerprint_rule._replace( policy='monetize', territory='US', service='tiktok', active=False ), mock_fingerprint_rule._replace( policy='carveout', territory='IT', service='tiktok', start_date=None ), ] t2_rules = [ mock_fingerprint_rule._replace( policy='carveout', territory='FR', service='tiktok', start_date=None ), ] mock_orchard_sound_recording = mock_orchard_sound_recording._replace( track_connection=TrackConnection( tracks=[ mock_track._replace( name='Track 1', tuid='1', product=mock_product._replace( release_date='2019-06-15', label=mock_label._replace( rules=t1_subaccount_rules, vendor=t1_vendor_rules, type=LabelType('Subaccount'), ), ), rules=t1_rules, ), mock_track._replace( name='Track 2', tuid='2', product=mock_product._replace( release_date='2019-06-14', label=mock_label._replace( rules=t2_vendor_rules, type=LabelType('Vendor') ), ), rules=t2_rules, ), ] ) ) result = utils_rules.get_fingerprint_rules_by_track_id( mock_orchard_sound_recording.track_connection.tracks ) expected_t1_rules = { 'vendor_rules': [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'FR', 'level': 'vendor', } # noqa:E501 ], 'subaccount_rules': [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'CA', 'level': 'subaccount', } # noqa:E501 ], 'track_rules': [], 'vendor_carveouts': [], 'subaccount_carveouts': [], 'track_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'FR', } # noqa:E501 ], } expected_t2_rules = { 'vendor_rules': [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'FR', 'level': 'vendor', } # noqa:E501 ], 'subaccount_rules': [], 'track_rules': [], 'vendor_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'IT', } # noqa:E501 ], 'subaccount_carveouts': [], 'track_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'FR', } # noqa:E501 ], } assert result['1']['rules'] == expected_t1_rules assert result['2']['rules'] == expected_t2_rules result_keys = list(result.keys()) assert len(result_keys) == 2 assert result_keys[0] == '2' assert result_keys[1] == '1' def test_filter_rules_by_service(): """Test filtering all rules level by tuid by service.""" data = { '1': { 'rules': { 'vendor_rules': [ { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'meta', 'territory': 'CA', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 ], 'subaccount_rules': [ { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'meta', 'territory': 'CA', }, # noqa:E501 ], 'track_rules': [ { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'meta', 'territory': 'CA', }, # noqa:E501 ], 'vendor_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'meta', 'territory': 'CA', }, # noqa:E501 ], 'subaccount_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'meta', 'territory': 'CA', }, # noqa:E501 ], 'track_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'meta', 'territory': 'CA', }, # noqa:E501 ], } } } expected_result = { '1': { 'rules': { 'vendor_rules': [ { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', } # noqa:E501 ], 'subaccount_rules': [ { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', } # noqa:E501 ], 'track_rules': [ { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', } # noqa:E501 ], 'vendor_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'CA', } # noqa:E501 ], 'subaccount_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'CA', } # noqa:E501 ], 'track_carveouts': [ { 'policy': 'carveout', 'start_date': None, 'end_date': None, 'service': 'tiktok', 'territory': 'CA', } # noqa:E501 ], } } } result = utils_rules._filter_rules_by_service(data, 'tiktok') assert result == expected_result def test_expand_rules_star_service_star_territory( mock_fingerprint_rule: FingerprintRule, ): """Test expand rules when * service and * territory.""" result = utils_rules._expand_rules([mock_fingerprint_rule]) tiktok_rules = [x for x in result if x['service'] == 'tiktok'] meta_rules = [x for x in result if x['service'] == 'meta'] youtube_rules = [x for x in result if x['service'] == 'youtube'] # 249 territories x 3 services assert len(result) == 747 assert len(tiktok_rules) == 249 assert len(meta_rules) == 249 assert len(youtube_rules) == 249 def test_expand_rules_star_service(mock_fingerprint_rule: FingerprintRule): """Test expand rules when * service.""" expected_result = [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'meta', 'territory': 'FR', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'youtube', 'territory': 'FR', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'FR', }, # noqa:E501 ] result = utils_rules._expand_rules([mock_fingerprint_rule._replace(territory='FR')]) assert result == expected_result def test_expand_rules_star_territory(mock_fingerprint_rule: FingerprintRule): """Test expand rules when * territory.""" result = utils_rules._expand_rules( [mock_fingerprint_rule._replace(service='tiktok')] ) # 249 territories assert len(result) == 249 def test_expand_rules_no_star(mock_fingerprint_rule: FingerprintRule): """Test expand rules when no *.""" result = utils_rules._expand_rules( [mock_fingerprint_rule._replace(service='tiktok', territory='FR')] ) expected_result = [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'FR', } # noqa:E501 ] assert result == expected_result @freeze_time('2024-01-01 12:00:00') def test_find_active_rule_by_date(mock_fingerprint_rule: FingerprintRule): """Test get active claimed and blocked rules.""" rules = [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc), 'service': 'meta', 'territory': 'CA', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'FR', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 2, 11, 0, 0, tzinfo=timezone.utc), 'service': 'tiktok', 'territory': 'UK', }, # noqa:E501 { 'policy': 'block_access', 'start_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc), 'service': 'tiktok', 'territory': 'US', }, # noqa:E501 { 'policy': 'block_access', 'start_date': datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'service': 'tiktok', 'territory': 'CA', }, # noqa:E501 { 'policy': 'block_access', 'start_date': datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'FR', }, # noqa:E501 { 'policy': 'block_access', 'start_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'AL', }, # noqa:E501 ] expected_result = [ { 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': None, 'territory': 'CA', 'policy': 'monetize', }, # noqa:E501 { 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': None, 'territory': 'FR', 'policy': 'monetize', }, # noqa:E501 { 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': datetime(2024, 1, 2, 11, 0, 0, tzinfo=timezone.utc), 'territory': 'UK', 'policy': 'monetize', }, # noqa:E501 { 'start': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end': datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc), 'territory': 'US', 'policy': 'block_access', }, # noqa:E501 { 'start': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end': None, 'territory': 'AL', 'policy': 'block_access', }, # noqa:E501 ] result = utils_rules._find_active_rule_by_date(rules) assert result == expected_result def test_deduplicate_territories(): """Test dedup claimed and blocked rules.""" tracks_and_rules = { '123': { 'claimed_rules': [ {'territory': 'AA', 'start': '2024-01-01', 'end': None}, {'territory': 'BB', 'start': '2024-01-01', 'end': None}, ], 'blocked_rules': [{'territory': 'DD', 'start': '2024-01-01', 'end': None}], }, '456': { 'claimed_rules': [ {'territory': 'AA', 'start': '2024-01-01', 'end': None}, {'territory': 'BB', 'start': '2024-01-01', 'end': None}, {'territory': 'CC', 'start': '2024-01-01', 'end': None}, ], 'blocked_rules': [ {'territory': 'DD', 'start': '2024-01-01', 'end': None}, {'territory': 'EE', 'start': '2024-01-01', 'end': None}, ], }, '789': { 'claimed_rules': [ {'territory': 'AA', 'start': '2024-01-01', 'end': None}, {'territory': 'CC', 'start': '2024-01-01', 'end': None}, ], 'blocked_rules': [ {'territory': 'DD', 'start': '2024-01-01', 'end': '2025-01-01'} ], }, } expected_result = { '123': { 'claimed_rules': [ {'territory': 'AA', 'start': '2024-01-01', 'end': None}, {'territory': 'BB', 'start': '2024-01-01', 'end': None}, ], 'blocked_rules': [{'territory': 'DD', 'start': '2024-01-01', 'end': None}], }, '456': { 'claimed_rules': [{'territory': 'CC', 'start': '2024-01-01', 'end': None}], 'blocked_rules': [{'territory': 'EE', 'start': '2024-01-01', 'end': None}], }, '789': { 'claimed_rules': [], 'blocked_rules': [ {'territory': 'DD', 'start': '2024-01-01', 'end': '2025-01-01'} ], }, } result = utils_rules._deduplicate_territories(tracks_and_rules) assert result == expected_result def test_extract_service_territory_rules(): """Test extract service/territory rules.""" rules = [ { 'policy': 'monetize', 'start_date': '2024-01-01T00:00:00.000Z', 'end_date': '2024-01-01T11:00:00.000Z', 'service': 'meta', 'territory': 'CA', 'level': 'track', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', 'level': 'vendor', }, # noqa:E501 { 'policy': 'block_access', 'start_date': '2024-01-01T00:00:00.000Z', 'end_date': '2024-01-02T00:00:00.000Z', 'service': 'tiktok', 'territory': 'US', 'level': 'track', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T00:00:00.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'US', 'level': 'vendor', }, # noqa:E501 { 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': '2024-01-01T11:00:00.000Z', 'territory': 'FR', 'policy': 'monetize', 'level': 'track', }, # noqa:E501 { 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'territory': 'FR', 'policy': 'monetize', 'level': 'subaccount', }, # noqa:E501 { 'start_date': '2023-01-01T10:01:40.000Z', 'end_date': None, 'territory': 'FR', 'policy': 'monetize', 'level': 'vendor', }, # noqa:E501 ] expected_result = { 'tiktok-CA': [ { 'policy': 'monetize', 'start_date': '2024-01-01T00:00:00.000Z', 'end_date': '2024-01-01T11:00:00.000Z', 'service': 'meta', 'territory': 'CA', 'level': 'track', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'CA', 'level': 'vendor', }, # noqa:E501 ], 'tiktok-US': [ { 'policy': 'block_access', 'start_date': '2024-01-01T00:00:00.000Z', 'end_date': '2024-01-02T00:00:00.000Z', 'service': 'tiktok', 'territory': 'US', 'level': 'track', }, # noqa:E501 { 'policy': 'monetize', 'start_date': '2024-01-01T00:00:00.000Z', 'end_date': None, 'service': 'tiktok', 'territory': 'US', 'level': 'vendor', }, # noqa:E501 ], 'tiktok-FR': [ { 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': '2024-01-01T11:00:00.000Z', 'territory': 'FR', 'policy': 'monetize', 'level': 'track', }, # noqa:E501 { 'start_date': '2024-01-01T10:01:40.000Z', 'end_date': None, 'territory': 'FR', 'policy': 'monetize', 'level': 'subaccount', }, # noqa:E501 { 'start_date': '2023-01-01T10:01:40.000Z', 'end_date': None, 'territory': 'FR', 'policy': 'monetize', 'level': 'vendor', }, # noqa:E501 ], } result = utils_rules._extract_service_territory_rules('tiktok', rules) assert result == expected_result @freeze_time('2024-01-01 12:00:00') def test_extract_rules_by_level_and_date(): """Test extract rules based on level and date.""" service_territory_rules = { 'tiktok-CA': [ { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc), 'service': 'meta', 'territory': 'CA', 'level': 'track', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'CA', 'level': 'vendor', }, # noqa:E501 ], 'tiktok-US': [ { 'policy': 'block_access', 'start_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc), 'service': 'tiktok', 'territory': 'US', 'level': 'track', }, # noqa:E501 { 'policy': 'monetize', 'start_date': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end_date': None, 'service': 'tiktok', 'territory': 'US', 'level': 'vendor', }, # noqa:E501 ], 'tiktok-FR': [ { 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': datetime(2024, 1, 1, 11, 0, 0, tzinfo=timezone.utc), 'territory': 'FR', 'policy': 'monetize', 'level': 'track', }, # noqa:E501 { 'start_date': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'territory': 'FR', 'policy': 'monetize', 'level': 'subaccount', }, # noqa:E501 { 'start_date': datetime(2023, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end_date': None, 'territory': 'FR', 'policy': 'monetize', 'level': 'vendor', }, # noqa:E501 ], } expected_result = [ { 'policy': 'monetize', 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': None, 'territory': 'CA', }, # noqa:E501 { 'policy': 'block_access', 'start': datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), 'end': datetime(2024, 1, 2, 0, 0, 0, tzinfo=timezone.utc), 'territory': 'US', }, # noqa:E501 { 'policy': 'monetize', 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': None, 'territory': 'FR', }, # noqa:E501 ] result = utils_rules._extract_rules_by_level_and_date(service_territory_rules) assert result == expected_result @freeze_time('2024-01-03 12:00:00') def test_get_territory_claims_by_rules( mock_track: Track, mock_product: Product, mock_label: Label, mock_vendor_rules: HasFingerprintRules, mock_fingerprint_rule: FingerprintRule, ): """Test generate active claims and blocked rules.""" t1_vendor_rules = mock_vendor_rules._replace( rules=[ mock_fingerprint_rule._replace( policy='carveout', territory='*', service='tiktok' ) ] ) t1_subaccount_rules = [ mock_fingerprint_rule._replace( policy='monetize', territory='CA', service='tiktok' ), ] t1_rules = [ mock_fingerprint_rule._replace( policy='carveout', territory='FR', service='tiktok' ), mock_fingerprint_rule._replace( policy='monetize', territory='US', service='tiktok' ), ] t2_vendor_rules = mock_vendor_rules._replace( rules=[ mock_fingerprint_rule._replace( policy='monetize', territory='*', service='tiktok' ) ] ) t2_subaccount_rules = [ mock_fingerprint_rule._replace( policy='carveout', territory='*', service='tiktok' ), ] t2_rules: list = [] t3_vendor_rules = mock_vendor_rules._replace( rules=[ mock_fingerprint_rule._replace( policy='monetize', territory='*', service='tiktok' ) ] ) t3_subaccount_rules: list = [] t3_rules = [ mock_fingerprint_rule._replace( policy='carveout', territory='*', service='tiktok' ) ] t4_vendor_rules = mock_vendor_rules._replace( rules=[ mock_fingerprint_rule._replace( policy='monetize', territory='US', service='tiktok' ), mock_fingerprint_rule._replace( policy='monetize', territory='FR', service='tiktok' ), mock_fingerprint_rule._replace( policy='monetize', territory='AL', service='tiktok' ), ] ) t4_subaccount_rules: list = [] t4_rules = [ mock_fingerprint_rule._replace( policy='carveout', territory='FR', service='tiktok' ), mock_fingerprint_rule._replace( policy='block_access', territory='US', service='tiktok' ), mock_fingerprint_rule._replace( policy='block_access', territory='AL', service='tiktok', start_date=datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), end_date=datetime(2024, 1, 2, 10, 1, 40, tzinfo=timezone.utc), ), ] tracks = [ # Generate full territory carveout at vendor level mock_track._replace( name='Track 1', tuid='1', product=mock_product._replace( release_date='2019-06-15', label=mock_label._replace( rules=t1_subaccount_rules, vendor=t1_vendor_rules, type=LabelType('Subaccount'), ), ), rules=t1_rules, ), # Generate full territory carveout at subaccount level mock_track._replace( name='Track 2', tuid='2', product=mock_product._replace( release_date='2019-06-14', label=mock_label._replace( rules=t2_subaccount_rules, vendor=t2_vendor_rules, type=LabelType('Subaccount'), ), ), rules=t2_rules, ), # Generate full territory carveout at track level mock_track._replace( name='Track 3', tuid='3', product=mock_product._replace( release_date='2019-06-14', label=mock_label._replace( rules=t3_subaccount_rules, vendor=t3_vendor_rules, type=LabelType('Subaccount'), ), ), rules=t3_rules, ), # Generate claimed and block rules mock_track._replace( name='Track 4', tuid='4', product=mock_product._replace( release_date='2019-06-14', label=mock_label._replace( rules=t4_subaccount_rules, vendor=t4_vendor_rules, type=LabelType('Subaccount'), ), ), rules=t4_rules, ), ] track_and_rules_by_tuid = utils_rules.get_fingerprint_rules_by_track_id(tracks) # noqa:E501 result = utils_rules.get_territory_claims_by_rules(track_and_rules_by_tuid, 'tiktok') # TUID 1: No claimed/blocked rules because of full Vendor carveout assert result['1']['claimed_rules'] == [] assert result['1']['blocked_rules'] == [] # TUID 2: No claimed/blocked rules because of full Subaccount carveout assert result['2']['claimed_rules'] == [] assert result['2']['blocked_rules'] == [] # TUID 3: No claimed/blocked rules because of full Track carveout assert result['3']['claimed_rules'] == [] assert result['3']['blocked_rules'] == [] # TUID 4: monetize AL from Vendor because block access rule expired # TUID 4: No monetize FR from Vendor because carveout rule at the Track # TUID 4: No monetize US from Vendor because block rule at the Track assert result['4']['claimed_rules'] == [ { 'policy': 'monetize', 'territory': 'AL', 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': None, } ] assert result['4']['blocked_rules'] == [ { 'policy': 'block_access', 'territory': 'US', 'start': datetime(2024, 1, 1, 10, 1, 40, tzinfo=timezone.utc), 'end': None, } ] @freeze_time('2024-01-03 12:00:00') def test_get_territory_claims_by_rules_filters_by_service( mock_track: Track, mock_product: Product, mock_label: Label, mock_fingerprint_rule: FingerprintRule, ): """Test that only rules for the specified service are used.""" youtube_rule = mock_fingerprint_rule._replace( policy='monetize', territory='US', service='youtube' ) tiktok_rule = mock_fingerprint_rule._replace( policy='monetize', territory='FR', service='tiktok' ) track = mock_track._replace( tuid='1', product=mock_product._replace( release_date='2019-06-15', label=mock_label._replace( rules=[youtube_rule, tiktok_rule], vendor=None, type=LabelType('Vendor'), ), ), rules=[], ) track_and_rules_by_tuid = utils_rules.get_fingerprint_rules_by_track_id([track]) result = utils_rules.get_territory_claims_by_rules(track_and_rules_by_tuid, 'youtube') territories = [r['territory'] for r in result['1']['claimed_rules']] assert 'US' in territories assert 'FR' not in territories