"""Tests for project_transfer_job model (PORT-67).""" from contextlib import contextmanager from unittest.mock import patch from sqlalchemy import exc as sa_exc, text from project_manager.connector import mysql from project_manager.models import project_transfer_job as model _PROJECT_ID = 1 _DESTINATION_VENDOR_ID = 9999 def _read_project_vendor_id(project_id): with mysql.pm_session_scope() as s: row = s.execute( text('SELECT vendor_id FROM project WHERE project_id = :pid'), {'pid': project_id}, ).one() return row['vendor_id'] def test_execute_content_transfer_rolls_back_on_mid_transaction_error( db_fixture, projects, test_request_context): """A failure during the releases UPDATE rolls back the project change. Verifies that all writes in execute_content_transfer are wrapped in a single transaction: if the releases UPDATE raises an error after the project UPDATE has already executed, pm_session_scope rolls back and the project row is left unchanged. """ assert _read_project_vendor_id(_PROJECT_ID) == 7123 original_scope = mysql.pm_session_scope @contextmanager def scope_with_releases_failure(): # Count execute calls so we can fail on the releases UPDATE (2nd call: # UPDATE project, UPDATE releases). call_n = [0] with original_scope() as session: orig_exec = session.execute def patched_exec(stmt, *args, **kwargs): call_n[0] += 1 if call_n[0] == 2: raise sa_exc.SQLAlchemyError('simulated releases update failure') return orig_exec(stmt, *args, **kwargs) session.execute = patched_exec yield session with patch.object(mysql, 'pm_session_scope', scope_with_releases_failure): result = model.execute_content_transfer( job_id=42, project_id=_PROJECT_ID, destination_vendor_id=_DESTINATION_VENDOR_ID, destination_subaccount_id=None, destination_artist_id=77, products=[{ 'release_id': 1001, 'destination_artist_id': 77, 'destination_video_artist_id': None, }], ) # @wrap_db_errors converts the SQLAlchemyError to a fatal response assert result.status == 500 # The project UPDATE was rolled back; vendor_id must be unchanged assert _read_project_vendor_id(_PROJECT_ID) == 7123 def _set_release_artist_info_id(release_id, artist_info_id): with mysql.pm_session_scope() as s: s.execute( text('UPDATE release_artist SET artist_info_id = :aid WHERE release_id = :rid'), {'aid': artist_info_id, 'rid': release_id}, ) def _set_track_artist_info_id(track_id, artist_info_id): with mysql.pm_session_scope() as s: s.execute( text('UPDATE track_artist SET artist_info_id = :aid WHERE track_id = :tid'), {'aid': artist_info_id, 'tid': track_id}, ) def _set_track_writer_info_id(unique_track_id, artist_info_id): with mysql.pm_session_scope() as s: s.execute( text('UPDATE track_writer SET artist_info_id = :aid WHERE unique_track_id = :uid'), {'aid': artist_info_id, 'uid': unique_track_id}, ) def _read_release_artist_info_ids(release_id): with mysql.pm_session_scope() as s: rows = s.execute( text('SELECT artist_info_id FROM release_artist WHERE release_id = :rid'), {'rid': release_id}, ).fetchall() return [r['artist_info_id'] for r in rows] def _read_track_artist_info_ids(track_id): with mysql.pm_session_scope() as s: rows = s.execute( text('SELECT artist_info_id FROM track_artist WHERE track_id = :tid'), {'tid': track_id}, ).fetchall() return [r['artist_info_id'] for r in rows] def _read_track_writer_info_ids(unique_track_id): with mysql.pm_session_scope() as s: rows = s.execute( text('SELECT artist_info_id FROM track_writer WHERE unique_track_id = :uid'), {'uid': unique_track_id}, ).fetchall() return [r['artist_info_id'] for r in rows] def test_execute_content_transfer_updates_artist_info_ids( db_fixture, projects, track_tables, test_request_context): """artist_info_id is remapped on release_artist, track_artist, and track_writer. Rows matching source_artist_id are updated to destination_artist_id. Rows with a different artist_info_id (e.g. a featuring artist) are untouched. """ _SOURCE = 50 _DEST = 77 _FEATURING = 99 # must not be touched # release 1001 has tracks 5001 and 5002 (from track_tables fixture) _set_release_artist_info_id(1001, _SOURCE) _set_track_artist_info_id(5001, _SOURCE) _set_track_writer_info_id(5001, _SOURCE) # featuring artist on the same release — must stay untouched with mysql.pm_session_scope() as s: s.execute( text( "INSERT INTO release_artist (upc, role, release_id, artist_name, artist_info_id) " "VALUES (888831283041, 'featuring', 1001, 'Guest', :fid)" ), {'fid': _FEATURING}, ) s.execute( text( "INSERT INTO track_artist (track_id, type, name, artist_info_id) " "VALUES (5001, 'featuring', 'Guest', :fid)" ), {'fid': _FEATURING}, ) s.execute( text( "INSERT INTO track_writer (writer_name, upc, cd, track_id, unique_track_id, " "artist_info_id) VALUES ('Guest', 888831283041, 9, 1, 5001, :fid)" ), {'fid': _FEATURING}, ) result = model.execute_content_transfer( job_id=42, project_id=_PROJECT_ID, destination_vendor_id=_DESTINATION_VENDOR_ID, destination_subaccount_id=None, destination_artist_id=_DEST, products=[{ 'release_id': 1001, 'source_artist_id': _SOURCE, 'destination_artist_id': _DEST, 'destination_video_artist_id': None, }], ) assert result['release_artist_rows_updated'] == 1 assert result['track_artist_rows_updated'] == 1 assert result['track_writer_rows_updated'] == 1 # primary artist rows updated assert _read_release_artist_info_ids(1001) == [_DEST, _FEATURING] assert _read_track_artist_info_ids(5001) == [_DEST, _FEATURING] assert _read_track_writer_info_ids(5001) == [_DEST, _FEATURING] # track 5002 on the same release also had source artist — not in products so untouched # (track_artist row for 5002 still has _SOURCE because no track_writer/track_artist was set) # release 1002 was not in the job — its rows must be untouched assert _read_release_artist_info_ids(1002) == [None]