# pylint: disable=too-many-statements import pytest from dapd_db_schema.schemas.etl import DimPlaylistMeta, DimTrack, FactPlaylistTrackDynamics from dapd_transformation_service.entities.dimension_meta import TrackPositions from dapd_transformation_service.services.facts.fact_track_dynamics import FactTrackDynamicsService pytestmark = pytest.mark.integration @pytest.mark.skip('Playlist track dynamics table removed') def test_process_playlist__new_playlist(db, dim_playlist, dim_market, dim_dsp): total_tracks = 3 track_positions = TrackPositions() for idx in range(0, total_tracks): track_id = f'track_id_{idx}' track_positions.add(position=idx, track_id=track_id) instance = DimTrack( **{ 'dsp_id': dim_dsp.dsp_id, 'market_id': dim_market.market_id, 'track_id': track_id, 'track_name': '062 - Spuk am Himmel - Teil 02', 'track_duration': 104, 'isrc': 'DEC711904690', 'audio_features': None, 'composer': None, 'upc': None, 'updated_at': '2020-10-24 22:11:00.385252', 'created_at': '2020-10-24 22:11:00.385252', 'release_date': '2020-10-24' } ) db.add(instance) db.flush() dim_playlist.metadata.track_positions = track_positions assert db.query(DimPlaylistMeta).count() == 0 assert db.query(FactPlaylistTrackDynamics).count() == 0 service = FactTrackDynamicsService(db) service.process_dimension(dim_playlist, {}) assert db.query(FactPlaylistTrackDynamics).count() == total_tracks for idx, fact in enumerate(db.query(FactPlaylistTrackDynamics).all()): fact: FactPlaylistTrackDynamics assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == f'track_id_{idx}' assert fact.track_delta == 0 assert fact.track_position == idx assert fact.total_tracks == total_tracks assert fact.is_playlist_entry assert not fact.is_playlist_exit @pytest.mark.skip('Playlist track dynamics table removed') def test_process_playlist__two_tracks_swapped(db, dim_playlist, dim_market, dim_dsp): previous_track_positions = TrackPositions() previous_track_positions.add(position=1, track_id='track_id_1') previous_track_positions.add(position=2, track_id='track_id_2') previous_track_positions.add(position=3, track_id='track_id_3') current_track_positions = TrackPositions() current_track_positions.add(position=1, track_id='track_id_1') current_track_positions.add(position=3, track_id='track_id_2') current_track_positions.add(position=2, track_id='track_id_3') for idx in range(1, len(current_track_positions) + 1): dim_track = DimTrack( **{ 'dsp_id': dim_dsp.dsp_id, 'market_id': dim_market.market_id, 'track_id': f'track_id_{idx}', 'track_name': '062 - Spuk am Himmel - Teil 02', 'track_duration': 104, 'isrc': 'DEC711904690', 'audio_features': None, 'composer': None, 'upc': None, 'updated_at': '2020-10-24 22:11:00.385252', 'created_at': '2020-10-24 22:11:00.385252', 'release_date': '2020-10-24' } ) db.add(dim_track) dim_playlist_meta = DimPlaylistMeta( playlist_id=dim_playlist.playlist_id, market_id=dim_playlist.market_id, dsp_id=dim_playlist.dsp_id, track_positions=previous_track_positions.to_plain_list(), ) db.add(dim_playlist_meta) db.flush() dim_playlist.metadata.track_positions = current_track_positions assert db.query(FactPlaylistTrackDynamics).count() == 0 service = FactTrackDynamicsService(db) service.process_dimension(dim_playlist, {}) assert db.query(FactPlaylistTrackDynamics).count() == len(current_track_positions) # First track, position was not changed fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_1').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_1' assert fact.track_delta == 0 assert fact.track_position == 1 assert fact.total_tracks == len(current_track_positions) assert not fact.is_playlist_entry assert not fact.is_playlist_exit # Second track, position went down fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_2').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_2' assert fact.track_delta == -1 assert fact.track_position == 3 assert fact.total_tracks == len(current_track_positions) assert not fact.is_playlist_entry assert not fact.is_playlist_exit # Third track, position went up fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_3').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_3' assert fact.track_delta == 1 assert fact.track_position == 2 assert fact.total_tracks == len(current_track_positions) assert not fact.is_playlist_entry assert not fact.is_playlist_exit @pytest.mark.skip('Playlist track dynamics table removed') def test_process_playlist__one_exit_one_entered(db, dim_playlist, dim_market, dim_dsp): previous_track_positions = TrackPositions() previous_track_positions.add(position=1, track_id='track_id_1') previous_track_positions.add(position=2, track_id='track_id_2') previous_track_positions.add(position=3, track_id='track_id_3') current_track_positions = TrackPositions() current_track_positions.add(position=1, track_id='track_id_1') current_track_positions.add(position=2, track_id='track_id_3') current_track_positions.add(position=3, track_id='track_id_4') for idx in range(1, len(current_track_positions) + 2): dim_track = DimTrack( **{ 'dsp_id': dim_dsp.dsp_id, 'market_id': dim_market.market_id, 'track_id': f'track_id_{idx}', 'track_name': '062 - Spuk am Himmel - Teil 02', 'track_duration': 104, 'isrc': 'DEC711904690', 'audio_features': None, 'composer': None, 'upc': None, 'updated_at': '2020-10-24 22:11:00.385252', 'created_at': '2020-10-24 22:11:00.385252', 'release_date': '2020-10-24' } ) db.add(dim_track) dim_playlist_meta = DimPlaylistMeta( playlist_id=dim_playlist.playlist_id, market_id=dim_playlist.market_id, dsp_id=dim_playlist.dsp_id, track_positions=previous_track_positions.to_plain_list(), ) db.add(dim_playlist_meta) db.flush() dim_playlist.metadata.track_positions = current_track_positions assert db.query(FactPlaylistTrackDynamics).count() == 0 service = FactTrackDynamicsService(db) service.process_dimension(dim_playlist, {}) assert db.query(FactPlaylistTrackDynamics).count() == len(current_track_positions) + 1 # First track, position was not changed fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_1').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_1' assert fact.track_delta == 0 assert fact.track_position == 1 assert fact.total_tracks == len(current_track_positions) assert not fact.is_playlist_entry assert not fact.is_playlist_exit # Second track, exited from playlist fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_2').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_2' assert fact.track_delta is None assert fact.track_position is None assert fact.total_tracks == len(current_track_positions) assert not fact.is_playlist_entry assert fact.is_playlist_exit # Third track, went from third position to second fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_3').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_3' assert fact.track_delta == 1 assert fact.track_position == 2 assert fact.total_tracks == len(current_track_positions) assert not fact.is_playlist_entry assert not fact.is_playlist_exit # Fourth track, entered the playlist fact: FactPlaylistTrackDynamics = db.query(FactPlaylistTrackDynamics) \ .filter(FactPlaylistTrackDynamics.track_id == 'track_id_4').one() assert fact.playlist_id == dim_playlist.playlist_id assert fact.market_id == dim_playlist.market_id assert fact.dsp_id == dim_playlist.dsp_id assert fact.track_id == 'track_id_4' assert fact.track_delta is None assert fact.track_position == 3 assert fact.total_tracks == len(current_track_positions) assert fact.is_playlist_entry assert not fact.is_playlist_exit @pytest.mark.skip('Playlist track dynamics table removed') def test_process_playlist__multiple_meta_items(db, dim_playlist, dim_market, dim_dsp): previous_track_positions = TrackPositions() previous_track_positions.add(position=1, track_id='track_id_1') previous_track_positions.add(position=2, track_id='track_id_2') previous_track_positions.add(position=3, track_id='track_id_3') current_track_positions = TrackPositions() current_track_positions.add(position=1, track_id='track_id_1') current_track_positions.add(position=3, track_id='track_id_2') current_track_positions.add(position=2, track_id='track_id_3') for idx in range(1, len(current_track_positions) + 1): dim_track = DimTrack( **{ 'dsp_id': dim_dsp.dsp_id, 'market_id': dim_market.market_id, 'track_id': f'track_id_{idx}', 'track_name': '062 - Spuk am Himmel - Teil 02', 'track_duration': 104, 'isrc': 'DEC711904690', 'audio_features': None, 'composer': None, 'upc': None, 'updated_at': '2020-10-24 22:11:00.385252', 'created_at': '2020-10-24 22:11:00.385252', 'release_date': '2020-10-24' } ) db.add(dim_track) for _ in range(0, 5): dim_playlist_meta = DimPlaylistMeta( playlist_id=dim_playlist.playlist_id, market_id=dim_playlist.market_id, dsp_id=dim_playlist.dsp_id, track_positions=previous_track_positions.to_plain_list(), ) db.add(dim_playlist_meta) db.flush() dim_playlist.metadata.track_positions = current_track_positions assert db.query(FactPlaylistTrackDynamics).count() == 0 service = FactTrackDynamicsService(db) service.process_dimension(dim_playlist, {}) assert db.query(FactPlaylistTrackDynamics).count() == len(current_track_positions)