<commit_before>from indra.statements import Phosphorylation
from indra.sources.phosphoelm.api import PhosphoElmProcessor, \
    _get_json_from_entry_rows

columns = ['acc', 'sequence', 'position', 'code', 'pmids', 'kinases',
           'source', 'species', 'entry_date']
non_human_no_kinase = ['O08539',
                       'MAEMGSKG', '6', 'S', '17114649', '', 'HTP',
                       'Mus musculus', '2005-03-14 12:16:11.108314+01']
human_no_kinase = ['O14543',
                   'MVTHSKFPAAGMSRPLDTSLRLKTFSSKSEYQL', '31', 'Y',
                   '12783885', '', 'LTP', 'Homo sapiens',
                   '2006-10-17 12:06:48.271076+02']
human_kinase1 = ['O14543',
                 'FAKESEQUENCESRPLDTSLRLKTFSSKSEYQL',
                 '31', 'Y', '12783885', 'Lck', 'LTP', 'Homo sapiens',
                 '2006-10-17 12:06:48.16767+02']
human_kinase2 = ['O14746', 'FAKEPRCRAVRSLL', '12', 'S', '10224060',
                 'PKB_group', 'LTP', 'Homo sapiens',
                 '2004-12-31 00:00:00+01']

raw_data = [columns, non_human_no_kinase, human_no_kinase, human_kinase1,
            human_kinase2]


def test_json_processing():
    test_json = _get_json_from_entry_rows(iter(raw_data))
    assert len(test_json) == len(raw_data) - 1, len(test_json)
    assert set(columns) == set(test_json[0].keys())


def test_keep_empty():
    stmts = PhosphoElmProcessor(
        file_dump_json=_get_json_from_entry_rows(iter(raw_data)),
        keep_empty=True
    ).statements
    assert len(stmts) == 3
    assert all(isinstance(st, Phosphorylation) for st in stmts)
    assert all(st.evidence[0].source_api == 'phospho.ELM' for st in stmts)
    assert all(len(st.evidence[0].annotations['sequence']) > 0
               for st in stmts)


def test_not_empty():
    stmts = PhosphoElmProcessor(
        file_dump_json=_get_json_from_entry_rows(iter(raw_data))
    ).statements
    assert len(stmts) == 2


<commit_msg>Test for reading from s3<commit_after>import csv
from indra.statements import Phosphorylation
from indra.sources.phosphoelm.api import PhosphoElmProcessor, \
    _get_json_from_entry_rows, _get_s3_client

columns = ['acc', 'sequence', 'position', 'code', 'pmids', 'kinases',
           'source', 'species', 'entry_date']
non_human_no_kinase = ['O08539',
                       'MAEMGSKG', '6', 'S', '17114649', '', 'HTP',
                       'Mus musculus', '2005-03-14 12:16:11.108314+01']
human_no_kinase = ['O14543',
                   'MVTHSKFPAAGMSRPLDTSLRLKTFSSKSEYQL', '31', 'Y',
                   '12783885', '', 'LTP', 'Homo sapiens',
                   '2006-10-17 12:06:48.271076+02']
human_kinase1 = ['O14543',
                 'FAKESEQUENCESRPLDTSLRLKTFSSKSEYQL',
                 '31', 'Y', '12783885', 'Lck', 'LTP', 'Homo sapiens',
                 '2006-10-17 12:06:48.16767+02']
human_kinase2 = ['O14746', 'FAKEPRCRAVRSLL', '12', 'S', '10224060',
                 'PKB_group', 'LTP', 'Homo sapiens',
                 '2004-12-31 00:00:00+01']

raw_data = [columns, non_human_no_kinase, human_no_kinase, human_kinase1,
            human_kinase2]


def test_download_from_s3():
    s3 = _get_s3_client()
    s3_bucket = 'bigmech'
    ppelm_s3_key = 'phosphoELM_data/phosphoELM_all_2015-04.dump'
    s3_obj = s3.get_object(Bucket=s3_bucket, Key=ppelm_s3_key)
    csv_reader = csv.reader(
        s3_obj['Body'].read().decode('utf8').splitlines(True),
        delimiter='\t'
    )
    ppelm_json = _get_json_from_entry_rows(csv_reader)
    assert ppelm_json


def test_json_processing():
    test_json = _get_json_from_entry_rows(iter(raw_data))
    assert len(test_json) == len(raw_data) - 1, len(test_json)
    assert set(columns) == set(test_json[0].keys())


def test_keep_empty():
    stmts = PhosphoElmProcessor(
        file_dump_json=_get_json_from_entry_rows(iter(raw_data)),
        keep_empty=True
    ).statements
    assert len(stmts) == 3
    assert all(isinstance(st, Phosphorylation) for st in stmts)
    assert all(st.evidence[0].source_api == 'phospho.ELM' for st in stmts)
    assert all(len(st.evidence[0].annotations['sequence']) > 0
               for st in stmts)


def test_not_empty():
    stmts = PhosphoElmProcessor(
        file_dump_json=_get_json_from_entry_rows(iter(raw_data))
    ).statements
    assert len(stmts) == 2



{'fraction_non_alphanumeric': 0.14009774261112404, 'fraction_numerical': 0.07470328135908774, 'mean_word_length': 3.0894386298763084, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5203785', 'n_tokens_mistral': 1823, 'n_tokens_neox': 1581, 'n_words': 272}