"""Tests for Processor.""" from typing import Any from unittest.mock import MagicMock, patch import pytest from gridgen.logic import processor from gridgen.models.metadata import ( ArtistMeta, GridMeta, Release, ReleaseIdMeta, ReleaseMeta, TrackIdMeta, TrackMeta, ) # grid meta tuple with a dummy GRid and product number GRID_META_TEST1 = GridMeta("GRid#00001", "Prod#00001") # grid meta tuple with a different GRid and product number GRID_META_TEST2 = GridMeta("GRid#00002", "Prod#00002") # grid meta tuple with GRid and product number produced by mock generator GRID_META_TEST_MOCK = GridMeta("A10426HABCDE12345Z", "G04ABCDE12345Z") # release meta tuple with a GRid and product number RELEASE_META_TEST1 = ReleaseMeta( 1001, "1234567890", GRID_META_TEST1.grid, GRID_META_TEST1.product_no, "release1", "info1", "config1", "owner1", "artist1", "complete", "digital", ) # release meta tuple with a different GRid and product number RELEASE_META_TEST2 = ReleaseMeta( 1001, "1234567890", GRID_META_TEST2.grid, GRID_META_TEST2.product_no, "release2", "info2", "config2", "owner2", "artist2", "complete", "digital", ) # release meta tuple with diff release_name meta. Should show that error RELEASE_META_TEST3 = ReleaseMeta( 1001, "1234567890", GRID_META_TEST2.grid, GRID_META_TEST2.product_no, "release3", "info2", "config2", "owner3", "artist2", "complete", "physical", ) # release meta tuple with no GRid or product number RELEASE_META_EMPTY_PROD_NUM = ReleaseMeta( 1001, "1234567890", None, None, "release3", "info3", "config3", "owner3", "artist3", "complete", "digital", ) # release meta tuple with no GRid or product number RELEASE_META_EMPTY_PROD_NUM2 = ReleaseMeta( 1001, "1234567890", None, None, "release3", "info3", "config3", "owner3", "artist3", "complete", "physical", ) # release meta tuple with no values set RELEASE_META_EMPTY = ReleaseMeta( None, "", None, None, None, None, None, None, None, None, None ) # track meta tuple with a GRid and product number TRACK_META_TEST1 = TrackMeta( 2001, "01234567890", "isrc", "track1", "info1", "pline1", "type1", 130, {ArtistMeta("artist1", "role1")}, {GRID_META_TEST1}, ) # track meta tuple with a different GRid and product number TRACK_META_TEST2 = TrackMeta( 2001, "01234567890", "isrc", "track2", "info2", "pline2", "type2", 140, {ArtistMeta("artist2", "role2")}, {GRID_META_TEST2}, ) # track meta tuple with the same track metadata. should have no errors TRACK_META_TEST3 = TrackMeta( 2001, "01234567890", "isrc", "track3", "info2", "pline2", "type2", 145, {ArtistMeta("Artist2", "Role2")}, {GRID_META_TEST2}, ) # track meta tuple with no 'p' in front of pline. Should still match TRACK_META_TEST4 = TrackMeta( 2001, "01234567890", "isrc", "track3", "info2", "(P) PLine2", "type2", 140, {ArtistMeta("artist2", "role2")}, {GRID_META_TEST2}, ) # track meta tuple with no GRid or product number TRACK_META_EMPTY_PROD_NUM = TrackMeta( 2001, "01234567890", "isrc", "track3", "info3", "pline3", "type3", 140, {ArtistMeta("artist3", "role3")}, set(), ) # track meta tuple with no values set TRACK_META_EMPTY = TrackMeta(None, None, "", "", "", "", "", 0, set(), set()) @pytest.mark.parametrize( "test_release, expected", [ (RELEASE_META_TEST1, GRID_META_TEST1), (RELEASE_META_EMPTY_PROD_NUM, processor.EMPTY_GRID_META), (None, processor.EMPTY_GRID_META), ], ) def test_extract_release_grid(test_release: ReleaseMeta, expected: GridMeta) -> None: """Test extracting grid data from release. Args: test_release (ReleaseMeta): the test release data. expected (GridMeta): the expected grid metadata tuple """ assert processor.extract_release_grid(test_release) == expected @pytest.mark.parametrize( "test_track, expected", [ (TRACK_META_TEST1, GRID_META_TEST1), (TRACK_META_EMPTY_PROD_NUM, processor.EMPTY_GRID_META), (None, processor.EMPTY_GRID_META), ], ) def test_extract_track_grid(test_track: TrackMeta, expected: GridMeta) -> None: """Test extracting grid data from track. Args: test_track (TrackMeta): the test track data. expected (GridMeta): the expected grid metadata tuple """ assert processor.extract_track_grid(test_track) == expected @pytest.mark.parametrize( ( "ar_grid_meta", "gras_grid_meta", "release_id_meta", "expected_result", "expected_save_call", ), [ pytest.param( GRID_META_TEST1, processor.EMPTY_GRID_META, ReleaseIdMeta(12345, "1235467890"), True, GRID_META_TEST1, id="use existing AR grid", ), pytest.param( GRID_META_TEST1, GRID_META_TEST2, ReleaseIdMeta(12345, "01234567890"), False, GRID_META_TEST2, id="use GRAS grid - both present", ), pytest.param( processor.EMPTY_GRID_META, GRID_META_TEST1, ReleaseIdMeta(12345, "01235467890"), False, GRID_META_TEST1, id="use GRAS grid - AR empty", ), pytest.param( processor.EMPTY_GRID_META, processor.EMPTY_GRID_META, ReleaseIdMeta(12345, "1235467890"), True, None, id="use new generated grid", ), ], ) @patch("gridgen.models.datasource.orchard.save_track", return_value=None) @patch("gridgen.models.datasource.orchard.save_release", return_value=None) def test_save_grid_release( save_release_mock: MagicMock, save_track_mock: MagicMock, ar_grid_meta: GridMeta, gras_grid_meta: GridMeta, release_id_meta: ReleaseIdMeta, expected_result: bool, expected_save_call: GridMeta | None, ) -> None: """Test save_grid for releases across all grid selection scenarios.""" result = processor.save_grid(ar_grid_meta, gras_grid_meta, release_id_meta) assert result == expected_result assert not save_track_mock.called if expected_save_call is not None: save_release_mock.assert_called_with(release_id_meta, expected_save_call) else: assert not save_release_mock.called @pytest.mark.parametrize( ("ar_grid_meta", "gras_grid_meta", "expected_result", "expected_save_call"), [ pytest.param( GRID_META_TEST1, processor.EMPTY_GRID_META, True, GRID_META_TEST1, id="use existing AR grid", ), pytest.param( GRID_META_TEST1, GRID_META_TEST2, False, GRID_META_TEST2, id="use GRAS grid - both present", ), pytest.param( processor.EMPTY_GRID_META, GRID_META_TEST1, False, GRID_META_TEST1, id="use GRAS grid - AR empty", ), pytest.param( processor.EMPTY_GRID_META, processor.EMPTY_GRID_META, True, None, id="use new generated grid", ), ], ) @patch("gridgen.models.datasource.orchard.save_track", return_value=None) @patch("gridgen.models.datasource.orchard.save_release", return_value=None) def test_save_grid_track( save_release_mock: MagicMock, save_track_mock: MagicMock, ar_grid_meta: GridMeta, gras_grid_meta: GridMeta, expected_result: bool, expected_save_call: GridMeta | None, ) -> None: """Test save_grid for tracks across all grid selection scenarios.""" track_id_meta = TrackIdMeta(12345, "ISRC98765", "1235467890") result = processor.save_grid(ar_grid_meta, gras_grid_meta, track_id_meta) assert result == expected_result assert not save_release_mock.called if expected_save_call is not None: save_track_mock.assert_called_with(track_id_meta, expected_save_call) else: assert not save_track_mock.called @pytest.mark.parametrize( "ar_release_meta, gras_release_meta", [ (RELEASE_META_TEST1, RELEASE_META_TEST1), (RELEASE_META_TEST1, RELEASE_META_TEST2), (RELEASE_META_EMPTY_PROD_NUM, RELEASE_META_TEST1), ], ) @patch("gridgen.models.datasource.orchard.save_release", return_value=None) def test_synch_release_with_gras( save_mock: MagicMock, ar_release_meta: ReleaseMeta, gras_release_meta: ReleaseMeta ) -> None: """Test synch release where gras release exists.""" release_id_meta = ReleaseIdMeta(ar_release_meta.release_id, ar_release_meta.upc) ar_release = Release(ar_release_meta) gras_release = Release(gras_release_meta) assert not processor.synch_release(ar_release, gras_release) save_mock.assert_called_with( release_id_meta, GridMeta(gras_release_meta.grid, gras_release_meta.product_number), ) @pytest.mark.parametrize( ("ar_release_meta", "expected_result", "expected_save_call"), [ pytest.param( RELEASE_META_TEST1, True, ( ReleaseIdMeta(RELEASE_META_TEST1.release_id, RELEASE_META_TEST1.upc), GRID_META_TEST1, ), id="empty gras", ), pytest.param(None, False, None, id="empty ar"), pytest.param(RELEASE_META_EMPTY_PROD_NUM, True, None, id="new grid"), pytest.param( RELEASE_META_EMPTY_PROD_NUM2, True, ( ReleaseIdMeta( RELEASE_META_EMPTY_PROD_NUM2.release_id, RELEASE_META_EMPTY_PROD_NUM2.upc, ), GridMeta(None, "123456789"), ), id="physical", ), ], ) @patch("gridgen.models.datasource.orchard.save_release", return_value=None) def test_synch_release_no_gras( save_mock: MagicMock, ar_release_meta: ReleaseMeta | None, expected_result: bool, expected_save_call: tuple[Any, ...] | None, ) -> None: """Test synch release scenarios where gras release is empty.""" ar_release = Release(ar_release_meta) gras_release = Release(None) assert processor.synch_release(ar_release, gras_release) == expected_result if expected_save_call is None: assert not save_mock.called else: save_mock.assert_called_with(*expected_save_call) @patch( "gridgen.models.datasource.orchard.lookup_track", return_value=TRACK_META_TEST1, ) @patch("gridgen.models.datasource.orchard.save_release", return_value=None) def test_synch_release_single_track( save_mock: MagicMock, lookup_mock: MagicMock ) -> None: """Test synch release generate new grid and product id.""" release_id_meta = ReleaseIdMeta( RELEASE_META_EMPTY_PROD_NUM.release_id, RELEASE_META_EMPTY_PROD_NUM.upc ) ar_release = Release(RELEASE_META_EMPTY_PROD_NUM, {TRACK_META_TEST1}) gras_release = Release(None) assert not processor.synch_release(ar_release, gras_release) lookup_mock.assert_called_with( TrackIdMeta( TRACK_META_TEST1.track_id, TRACK_META_TEST1.isrc, release_id_meta.upc ) ) save_mock.assert_called_with(release_id_meta, GRID_META_TEST1) @pytest.mark.parametrize( ("ar_track_meta", "gras_track_meta", "expected_result", "expected_save_call"), [ pytest.param( TRACK_META_TEST1, TRACK_META_TEST1, False, GRID_META_TEST1, id="with gras - same meta", ), pytest.param( TRACK_META_TEST1, TRACK_META_TEST2, False, GRID_META_TEST2, id="with gras - different meta", ), pytest.param( TRACK_META_EMPTY_PROD_NUM, TRACK_META_TEST1, False, GRID_META_TEST1, id="with gras - ar no product num", ), pytest.param(TRACK_META_TEST1, None, True, GRID_META_TEST1, id="empty gras"), ], ) @patch("gridgen.models.datasource.orchard.save_track", return_value=None) def test_synch_track_with_save( save_mock: MagicMock, ar_track_meta: TrackMeta, gras_track_meta: TrackMeta | None, expected_result: bool, expected_save_call: GridMeta, ) -> None: """Test synch_track scenarios where save_track is expected to be called.""" track_id_meta = TrackIdMeta(ar_track_meta.track_id, ar_track_meta.isrc, "") ar_release = Release(None, {ar_track_meta}) gras_release = ( Release(None, {gras_track_meta}) if gras_track_meta else Release(None) ) assert ( processor.synch_track(ar_release, gras_release, track_id_meta) == expected_result ) save_mock.assert_called_with(track_id_meta, expected_save_call) @pytest.mark.parametrize( ("ar_track_meta", "gras_track_meta", "expected_result", "expected_save_called"), [ pytest.param(None, TRACK_META_TEST1, False, False, id="empty ar"), # TRACK_META_EMPTY_PROD_NUM has no product_numbers, so extract_track_grid # returns EMPTY_GRID_META for both ar and gras — save_grid exits early without # calling save_track. pytest.param(TRACK_META_EMPTY_PROD_NUM, None, True, False, id="new grid"), ], ) @patch("gridgen.models.datasource.orchard.save_track", return_value=None) def test_synch_track_no_save_assert( save_mock: MagicMock, ar_track_meta: TrackMeta | None, gras_track_meta: TrackMeta | None, expected_result: bool, expected_save_called: bool, ) -> None: """Test synch_track scenarios where save_track should not be called.""" track_source = ar_track_meta if ar_track_meta else gras_track_meta assert track_source is not None track_id_meta = TrackIdMeta(track_source.track_id, track_source.isrc, "") ar_release = Release(None, {ar_track_meta}) if ar_track_meta else Release(None) gras_release = ( Release(None, {gras_track_meta}) if gras_track_meta else Release(None) ) assert ( processor.synch_track(ar_release, gras_release, track_id_meta) == expected_result ) assert save_mock.called == expected_save_called @pytest.mark.parametrize( ("ar_release", "gras_release", "expected"), [ pytest.param( RELEASE_META_TEST1, RELEASE_META_TEST2, [ { "error_code": "M01", "field": "release_name", "message": RELEASE_META_TEST2.release_name, }, { "error_code": "M01", "field": "display_artist_name", "message": RELEASE_META_TEST2.display_artist_name, }, ], id="fields differ", ), pytest.param(RELEASE_META_TEST1, RELEASE_META_EMPTY, [], id="gras empty"), ], ) def test_compare_release_fields( ar_release: ReleaseMeta, gras_release: ReleaseMeta, expected: list[dict[str, Any]], ) -> None: """Test compare release metadata.""" assert processor.compare_release_fields(ar_release, gras_release) == expected @pytest.mark.parametrize( ("ar_track", "gras_track", "expected"), [ pytest.param( TRACK_META_TEST1, TRACK_META_TEST3, [ { "error_code": "M01", "field": "main_participants", "message": ",".join( f"{a.name} ({a.role})" for a in TRACK_META_TEST3.main_participants ), }, { "error_code": "M01", "field": "track_name", "message": TRACK_META_TEST3.track_name, }, { "error_code": "M01", "field": "track_supplemental_info", "message": TRACK_META_TEST3.track_supplemental_info, }, { "error_code": "M01", "field": "p_line", "message": TRACK_META_TEST3.p_line, }, { "error_code": "M01", "field": "duration", "message": TRACK_META_TEST3.duration, }, { "error_code": "M01", "field": "track_type", "message": TRACK_META_TEST3.track_type, }, ], id="all fields differ", ), pytest.param( TRACK_META_TEST2, TRACK_META_TEST3, [ { "error_code": "M01", "field": "track_name", "message": TRACK_META_TEST3.track_name, } ], id="only track name differs", ), pytest.param( TRACK_META_TEST3, TRACK_META_TEST4, [], id="pline with (P) prefix - no errors", ), pytest.param(TRACK_META_TEST1, TRACK_META_EMPTY, [], id="gras empty"), ], ) def test_compare_track_fields( ar_track: TrackMeta, gras_track: TrackMeta, expected: list[dict[str, Any]], ) -> None: """Test compare track metadata.""" assert processor.compare_track_fields(ar_track, gras_track) == expected def test_validate_release() -> None: """Test compare release metadata.""" result: dict[str, Any] = { "id": RELEASE_META_TEST1.release_id, "data": { "upc": RELEASE_META_TEST1.upc, "product_no": RELEASE_META_TEST1.product_number, "grid": RELEASE_META_TEST1.grid, "errors": {"metadata": [], "general": []}, }, } ar_release = Release(RELEASE_META_TEST1) gras_release = Release(RELEASE_META_TEST1) assert processor.validate_release(ar_release, gras_release) == result def test_validate_release_diff_release_name() -> None: """Test compare release metadata with different release name.""" result = { "id": RELEASE_META_TEST2.release_id, "data": { "upc": RELEASE_META_TEST2.upc, "product_no": RELEASE_META_TEST2.product_number, "grid": RELEASE_META_TEST2.grid, "errors": { "metadata": [ { "error_code": "M01", "field": "release_name", "message": "release3", } ], "general": [], }, }, } ar_release = Release(RELEASE_META_TEST2) gras_release = Release(RELEASE_META_TEST3) assert processor.validate_release(ar_release, gras_release) == result def test_validate_release_ar_empty() -> None: """Test compare release metadata where art relations release is empty.""" ar_release = Release(None) gras_release = Release(RELEASE_META_TEST1) assert processor.validate_release(ar_release, gras_release) == {} def test_validate_release_gras_empty() -> None: """Test compare release metadata where gras release is empty.""" ar_release = Release(RELEASE_META_TEST1) gras_release = Release(None) assert processor.validate_release(ar_release, gras_release) == {} def test_validate_track() -> None: """Test compare release metadata.""" participants = [] for artist in TRACK_META_TEST1.main_participants: participants.append("{} ({})".format(artist.name, artist.role)) grid_meta = next(iter(TRACK_META_TEST1.product_numbers)) track_id_meta = TrackIdMeta(TRACK_META_TEST1.track_id, TRACK_META_TEST1.isrc, "") result: dict[str, Any] = { "id": TRACK_META_TEST1.track_id, "data": { "isrc": TRACK_META_TEST1.isrc, "product_no": grid_meta.product_no, "grid": grid_meta.grid, "errors": {"metadata": [], "general": []}, }, } ar_release = Release(None, {TRACK_META_TEST1}) gras_release = Release(None, {TRACK_META_TEST1}) assert processor.validate_track(ar_release, gras_release, track_id_meta) == result def test_validate_track_ar_empty() -> None: """Test compare release metadata where art relations release is empty.""" track_id_meta = TrackIdMeta(TRACK_META_TEST1.track_id, TRACK_META_TEST1.isrc, "") ar_release = Release(None) gras_release = Release(None, {TRACK_META_TEST1}) assert processor.validate_track(ar_release, gras_release, track_id_meta) == {} def test_validate_track_gras_empty() -> None: """Test compare release metadata where gras release is empty.""" track_id_meta = TrackIdMeta(TRACK_META_TEST1.track_id, TRACK_META_TEST1.isrc, "") ar_release = Release(None, {TRACK_META_TEST1}) gras_release = Release(None) assert processor.validate_track(ar_release, gras_release, track_id_meta) == {}