"""Tests for gn3.db.sample_data""" import pytest import gn3 from gn3.db.sample_data import __extract_actions from gn3.db.sample_data import delete_sample_data from gn3.db.sample_data import get_case_attributes from gn3.db.sample_data import insert_sample_data from gn3.db.sample_data import update_sample_data from gn3.db.sample_data import get_trait_csv_sample_data @pytest.mark.unit_test def test_insert_sample_data(mocker): """Test that inserts work properly""" mock_conn = mocker.MagicMock() strain_id, data_id, inbredset_id = 1, 17373, 20 with mock_conn.cursor() as cursor: cursor.fetchone.side_effect = ( 0, [ 19, ], 0, 0, ) mocker.patch( "gn3.db.sample_data.get_sample_data_ids", return_value=(strain_id, data_id, inbredset_id), ) insert_sample_data( conn=mock_conn, trait_name=35, data="BXD1,18,3,0,Red,M", csv_header="Strain Name,Value,SE,Count,Color,Sex (13)", phenotype_id=10007, ) calls = [ mocker.call( "SELECT Id FROM PublishData where Id = %s AND StrainId = %s", (17373, 1), ), mocker.call( "INSERT INTO PublishData (StrainId, Id, value) " "VALUES (%s, %s, %s)", (1, 17373, "18"), ), mocker.call( "INSERT INTO PublishSE (StrainId, DataId, error) VALUES " "(%s, %s, %s)", (1, 17373, "3"), ), mocker.call( "INSERT INTO NStrain (StrainId, DataId, count) VALUES " "(%s, %s, %s)", (1, 17373, "0"), ), mocker.call( "SELECT Id FROM CaseAttribute WHERE Name = %s", ("Color",) ), mocker.call( "SELECT StrainId FROM CaseAttributeXRefNew WHERE " "StrainId = %s AND CaseAttributeId = %s AND InbredSetId = %s", (1, 19, 20), ), mocker.call( "INSERT INTO CaseAttributeXRefNew (StrainId, " "CaseAttributeId, Value, InbredSetId) VALUES " "(%s, %s, %s, %s)", (1, 19, "Red", 20), ), mocker.call( "SELECT StrainId FROM CaseAttributeXRefNew WHERE " "StrainId = %s AND CaseAttributeId = %s AND " "InbredSetId = %s", (1, "13", 20), ), mocker.call( "INSERT INTO CaseAttributeXRefNew (StrainId, " "CaseAttributeId, Value, InbredSetId) VALUES " "(%s, %s, %s, %s)", (1, "13", "M", 20), ), ] cursor.execute.assert_has_calls(calls, any_order=False) @pytest.mark.unit_test def test_delete_sample_data(mocker): """Test that deletes work properly""" mock_conn = mocker.MagicMock() strain_id, data_id, inbredset_id = 1, 17373, 20 with mock_conn.cursor() as cursor: mocker.patch( "gn3.db.sample_data.get_sample_data_ids", return_value=(strain_id, data_id, inbredset_id), ) delete_sample_data( conn=mock_conn, trait_name=35, data="BXD1,18,3,0,Red,M", csv_header="Strain Name,Value,SE,Count,Color,Sex (17)", phenotype_id=10007, ) calls = [ mocker.call( "DELETE FROM PublishData WHERE StrainId = %s AND Id = %s", (strain_id, data_id), ), mocker.call( "DELETE FROM PublishSE WHERE StrainId = %s AND DataId = %s", (strain_id, data_id), ), mocker.call( "DELETE FROM NStrain WHERE StrainId = %s AND DataId = %s", (strain_id, data_id), ), mocker.call( "DELETE FROM CaseAttributeXRefNew WHERE " "StrainId = %s AND CaseAttributeId = " "(SELECT CaseAttributeId FROM " "CaseAttribute WHERE Name = %s) " "AND InbredSetId = %s", (strain_id, "Color", inbredset_id), ), mocker.call( "DELETE FROM CaseAttributeXRefNew WHERE " "StrainId = %s AND CaseAttributeId = %s " "AND InbredSetId = %s", (strain_id, "17", inbredset_id), ), ] cursor.execute.assert_has_calls(calls, any_order=False) @pytest.mark.unit_test def test_extract_actions(): """Test that extracting the correct dict of 'actions' work properly""" assert __extract_actions( original_data="BXD1,18,x,0,x", updated_data="BXD1,x,2,1,F", csv_header="Strain Name,Value,SE,Count,Sex", ) == { "delete": {"data": "BXD1,18", "csv_header": "Strain Name,Value"}, "insert": {"data": "BXD1,2,F", "csv_header": "Strain Name,SE,Sex"}, "update": {"data": "BXD1,1", "csv_header": "Strain Name,Count"}, } assert __extract_actions( original_data="BXD1,18,x,0,x", updated_data="BXD1,19,2,1,F", csv_header="Strain Name,Value,SE,Count,Sex", ) == { "delete": None, "insert": {"data": "BXD1,2,F", "csv_header": "Strain Name,SE,Sex"}, "update": { "data": "BXD1,19,1", "csv_header": "Strain Name,Value,Count", }, } @pytest.mark.unit_test def test_update_sample_data(mocker): """Test that updates work properly""" mock_conn = mocker.MagicMock() strain_id, data_id, inbredset_id = 1, 17373, 20 with mock_conn.cursor() as cursor: mocker.patch( "gn3.db.sample_data.get_sample_data_ids", return_value=(strain_id, data_id, inbredset_id), ) mocker.patch("gn3.db.sample_data.insert_sample_data", return_value=1) mocker.patch("gn3.db.sample_data.delete_sample_data", return_value=1) update_sample_data( conn=mock_conn, trait_name=35, original_data="BXD1,18,x,0,x,M,x,Red", updated_data="BXD1,x,2,1,2,F,2,Green", csv_header="Strain Name,Value,SE,Count,pH,Sex (13),Age,Color", phenotype_id=10007, ) # pylint: disable=[E1101] gn3.db.sample_data.insert_sample_data.assert_called_once_with( conn=mock_conn, trait_name=35, data="BXD1,2,2,2", csv_header="Strain Name,SE,pH,Age", phenotype_id=10007, ) # pylint: disable=[E1101] gn3.db.sample_data.delete_sample_data.assert_called_once_with( conn=mock_conn, trait_name=35, data="BXD1,18", csv_header="Strain Name,Value", phenotype_id=10007, ) cursor.execute.assert_has_calls( [ mocker.call( "UPDATE NStrain SET count = %s " "WHERE StrainId = %s AND DataId = %s", ("1", strain_id, data_id), ), mocker.call( "UPDATE CaseAttributeXRefNew SET Value = %s " "WHERE StrainId = %s AND CaseAttributeId = %s AND " "InbredSetId = %s", ("F", strain_id, "13", inbredset_id), ), mocker.call( "UPDATE CaseAttributeXRefNew SET Value = %s " "WHERE StrainId = %s AND CaseAttributeId = " "(SELECT CaseAttributeId FROM " "CaseAttribute WHERE Name = %s) " "AND InbredSetId = %s", ("Green", strain_id, "Color", inbredset_id), ), ], any_order=False, ) @pytest.mark.unit_test def test_get_case_attributes(mocker): """Test that case attributes work well""" mock_conn = mocker.MagicMock() with mock_conn.cursor() as cursor: cursor.fetchall.return_value = ( (1, "Condition", None), (2, "Tissue", None), (3, "Age", "Cum sociis natoque penatibus et magnis dis"), (4, "Condition", "Description A"), (5, "Condition", "Description B"), ) results = get_case_attributes(mock_conn) cursor.execute.assert_called_once_with( "SELECT Id, Name, Description FROM CaseAttribute" ) assert results == { "Condition (1)": "", "Tissue": "", "Age": "Cum sociis natoque penatibus et magnis dis", "Condition (4)": "Description A", "Condition (5)": "Description B", } @pytest.mark.unit_test def test_get_trait_csv_sample_data(mocker): """Test fetching trait sample data""" mock_conn = mocker.MagicMock() with mock_conn.cursor() as cursor: cursor.fetchall.return_value = [ ["JL00005,896.000000,x,x", "Age", "896"], ["JL00005,896.000000,x,x", "DOB", "4/22/04"], ["JL00005,896.000000,x,x", "Date Geno", "x"], ["JL00005,896.000000,x,x", "Drug/Site", "4OHPBN_J"], ["JL00005,896.000000,x,x", "MOD", "Oct"], ["JL00005,896.000000,x,x", "Platform", "x"], [ "JL00005,896.000000,x,x", "SNP", """0 """, ], ["JL00005,896.000000,x,x", "Sex", "M"], ["JL00005,896.000000,x,x", "Site", "JL"], ["JL00005,896.000000,x,x", "Tx", "1"], ["JL00005,896.000000,x,x", "Year", "2004"], ["JL00058,686.000000,x,x", "Age", "686"], ["JL00058,686.000000,x,x", "DOB", "4/22/04"], ["JL00058,686.000000,x,x", "Date Geno", "2017.06"], ["JL00058,686.000000,x,x", "Drug/Site", "Cont_04_J"], ["JL00058,686.000000,x,x", "MOD", "Mar"], ["JL00058,686.000000,x,x", "Platform", "M"], [ "JL00058,686.000000,x,x", "SNP", """2 """, ], ["JL00058,686.000000,x,x", "Sex", "M"], ["JL00058,686.000000,x,x", "Site", "JL"], ["JL00058,686.000000,x,x", "Tx", "0"], ["JL00058,686.000000,x,x", "Year", "2004"], ] assert get_trait_csv_sample_data( conn=mock_conn, trait_name=10007, phenotype_id=35 ) == ( "Strain Name,Value,SE,Count,Age,DOB," "Date Geno,Drug/Site,MOD," "Platform,SNP,Sex,Site,Tx,Year\n" "JL00005,896.000000,x,x," "896,4/22/04,x,4OHPBN_J,Oct,x,0,M,JL,1,2004\n" "JL00058,686.000000,x,x," "686,4/22/04,2017.06,Cont_04_J,Mar,M,2,M,JL,0,2004\n" )