<s> from nose . tools import * <EOL> from tests import ( <EOL> prepare_database , with_database , <EOL> TEST_1_FP_RAW , <EOL> TEST_1_LENGTH , <EOL> TEST_1A_FP_RAW , <EOL> TEST_1A_LENGTH , <EOL> TEST_1B_FP_RAW , <EOL> TEST_1B_LENGTH , <EOL> TEST_1C_FP_RAW , <EOL> TEST_1C_LENGTH , <EOL> TEST_1D_FP_RAW , <EOL> TEST_1D_LENGTH , <EOL> TEST_2_FP_RAW , <EOL> TEST_2_LENGTH , <EOL> ) <EOL> from acoustid import tables , const <EOL> from acoustid . data . meta import insert_meta <EOL> from acoustid . data . submission import insert_submission , import_submission , import_queued_submissions <EOL> @ with_database <EOL> def test_insert_submission ( conn ) : <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> assert_equals ( <NUM_LIT:1> , id ) <EOL> rows = conn . execute ( """<STR_LIT>""" , ( id , ) ) . fetchall ( ) <EOL> expected_rows = [ <EOL> ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ] , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) , <EOL> ] <EOL> assert_equals ( expected_rows , rows ) <EOL> @ with_database <EOL> def test_import_submission_with_foreignid ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1_FP_RAW , <EOL> '<STR_LIT>' : TEST_1_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> query = tables . track_foreignid . select ( tables . track_foreignid . c . track_id == fingerprint [ '<STR_LIT>' ] ) <EOL> track_foreignid = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:1> , track_foreignid [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission ( conn ) : <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1_FP_RAW , <EOL> '<STR_LIT>' : TEST_1_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:5> , fingerprint [ '<STR_LIT>' ] ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_true ( submission [ '<STR_LIT>' ] ) <EOL> query = tables . track_mbid . select ( tables . track_mbid . c . track_id == fingerprint [ '<STR_LIT>' ] ) <EOL> track_mbid = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:1> , track_mbid [ '<STR_LIT>' ] ) <EOL> query = tables . track_puid . select ( tables . track_puid . c . track_id == fingerprint [ '<STR_LIT>' ] ) <EOL> track_puid = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:1> , track_puid [ '<STR_LIT>' ] ) <EOL> query = tables . fingerprint . select ( tables . fingerprint . c . id == fingerprint [ '<STR_LIT:id>' ] ) <EOL> fingerprint = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT>' ] ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_2_FP_RAW , <EOL> '<STR_LIT>' : TEST_2_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> assert_equals ( <NUM_LIT:2> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:6> , fingerprint [ '<STR_LIT>' ] ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_true ( submission [ '<STR_LIT>' ] ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1_FP_RAW , <EOL> '<STR_LIT>' : TEST_1_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:5> , fingerprint [ '<STR_LIT>' ] ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_true ( submission [ '<STR_LIT>' ] ) <EOL> query = tables . track_mbid . select ( tables . track_mbid . c . track_id == fingerprint [ '<STR_LIT>' ] ) <EOL> track_mbid = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:2> , track_mbid [ '<STR_LIT>' ] ) <EOL> query = tables . track_puid . select ( tables . track_puid . c . track_id == fingerprint [ '<STR_LIT>' ] ) <EOL> track_puid = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:2> , track_puid [ '<STR_LIT>' ] ) <EOL> query = tables . fingerprint . select ( tables . fingerprint . c . id == fingerprint [ '<STR_LIT:id>' ] ) <EOL> fingerprint = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:2> , fingerprint [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission_reuse_fingerprint_97 ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" , dict ( fp = TEST_1A_FP_RAW , len = TEST_1A_LENGTH ) ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1B_FP_RAW , <EOL> '<STR_LIT>' : TEST_1B_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission_reuse_fingerprint_100 ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" , dict ( fp = TEST_1A_FP_RAW , len = TEST_1A_LENGTH ) ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1A_FP_RAW , <EOL> '<STR_LIT>' : TEST_1A_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission_reuse_track_93 ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" , dict ( fp = TEST_1A_FP_RAW , len = TEST_1A_LENGTH ) ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1C_FP_RAW , <EOL> '<STR_LIT>' : TEST_1C_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> try : <EOL> old_threshold = const . FINGERPRINT_MERGE_THRESHOLD <EOL> const . FINGERPRINT_MERGE_THRESHOLD = <NUM_LIT> <EOL> fingerprint = import_submission ( conn , submission ) <EOL> finally : <EOL> const . FINGERPRINT_MERGE_THRESHOLD = old_threshold <EOL> assert_equals ( <NUM_LIT:2> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission_new_track ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" , dict ( fp = TEST_1A_FP_RAW , len = TEST_1A_LENGTH ) ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1D_FP_RAW , <EOL> '<STR_LIT>' : TEST_1D_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> try : <EOL> old_threshold = const . TRACK_MERGE_THRESHOLD <EOL> const . TRACK_MERGE_THRESHOLD = <NUM_LIT> <EOL> fingerprint = import_submission ( conn , submission ) <EOL> finally : <EOL> const . TRACK_MERGE_THRESHOLD = old_threshold <EOL> assert_equals ( <NUM_LIT:2> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:5> , fingerprint [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission_new_track_different ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" , dict ( fp = TEST_1A_FP_RAW , len = TEST_1A_LENGTH ) ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_2_FP_RAW , <EOL> '<STR_LIT>' : TEST_2_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> fingerprint = import_submission ( conn , submission ) <EOL> assert_equals ( <NUM_LIT:2> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:5> , fingerprint [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_submission_merge_existing_tracks ( conn ) : <EOL> prepare_database ( conn , """<STR_LIT>""" , dict ( fp1 = TEST_1A_FP_RAW , len1 = TEST_1A_LENGTH , <EOL> fp2 = TEST_1B_FP_RAW , len2 = TEST_1B_LENGTH ) ) <EOL> id = insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1C_FP_RAW , <EOL> '<STR_LIT>' : TEST_1C_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> query = tables . submission . select ( tables . submission . c . id == id ) <EOL> submission = conn . execute ( query ) . fetchone ( ) <EOL> assert_false ( submission [ '<STR_LIT>' ] ) <EOL> try : <EOL> old_threshold = const . FINGERPRINT_MERGE_THRESHOLD <EOL> const . FINGERPRINT_MERGE_THRESHOLD = <NUM_LIT> <EOL> fingerprint = import_submission ( conn , submission ) <EOL> finally : <EOL> const . FINGERPRINT_MERGE_THRESHOLD = old_threshold <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT:id>' ] ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT>' ] ) <EOL> query = tables . fingerprint . select ( tables . fingerprint . c . id == <NUM_LIT:1> ) <EOL> fingerprint = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:1> , fingerprint [ '<STR_LIT>' ] ) <EOL> query = tables . track . select ( tables . track . c . id == <NUM_LIT:1> ) <EOL> track = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( None , track [ '<STR_LIT>' ] ) <EOL> query = tables . track . select ( tables . track . c . id == <NUM_LIT:2> ) <EOL> track = conn . execute ( query ) . fetchone ( ) <EOL> assert_equals ( <NUM_LIT:1> , track [ '<STR_LIT>' ] ) <EOL> @ with_database <EOL> def test_import_queued_submissions ( conn ) : <EOL> insert_meta ( conn , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1_FP_RAW , <EOL> '<STR_LIT>' : TEST_1_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_2_FP_RAW , <EOL> '<STR_LIT>' : TEST_2_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> insert_submission ( conn , { <EOL> '<STR_LIT>' : TEST_1_FP_RAW , <EOL> '<STR_LIT>' : TEST_1_LENGTH , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> import_queued_submissions ( conn ) <EOL> count = conn . execute ( "<STR_LIT>" ) . scalar ( ) <EOL> assert_equals ( <NUM_LIT:2> , count ) <EOL> count = conn . execute ( "<STR_LIT>" ) . scalar ( ) <EOL> assert_equals ( <NUM_LIT:2> , count ) </s>
