# Copyright 2008-2016 by Peter Cock. All rights reserved. # This code is part of the Biopython distribution and governed by its # license. Please see the LICENSE file that should have been included # as part of this package. """Testing online code for fetching NCBI qblast. Uses Bio.Blast.NCBIWWW.qblast() to run some online blast queries, get XML blast results back, and then checks Bio.Blast.NCBIXML.parse() can read them. Goals: - Make sure that all retrieval is working as expected. - Make sure we can parse the latest XML format being used by the NCBI. If an internet connection is available and run_tests.py has not given the argument '--offline', these tests will run online (and can take a long time). Otherwise, the module runs offline using a mocked version of urllib.request.urlopen, which returns file contents instead of internet responses. IMPORTANT: If you add new tests (or change existing tests) you must provide new 'mock' xml files which fulfill the requirements of the respective tests. These need to be added to the 'response_list' within the 'mock_response' function. Note: The tests are run in alphabetical order, so you must place your mock file at the correct position. """ import unittest from io import BytesIO from unittest import mock from urllib.error import HTTPError import requires_internet from Bio import MissingExternalDependencyError # We want to test these: from Bio.Blast import NCBIWWW from Bio.Blast import NCBIXML NCBIWWW.email = "biopython@biopython.org" try: requires_internet.check() # this will set 'require_internet.check.available' (if not already done before) # it is False if we don't have internet or if we run the tests with --offline except MissingExternalDependencyError: # need to catch this error, because we want to run the tests even offline pass # 'requires_internet_check.available' is now False if not requires_internet.check.available: # if the tests are running with --offline or if no internet is available, # we mock urlopen, so that it returns files from the blast folder def mock_response(): """Mimic an NCBI qblast response.""" # Each use of NCBIWWW.qblast makes two urlopen calls with different responses: # a. the 'wait' page, and b. the result. wait = ["Blast/mock_wait.html"] # This mimics the 'wait' page # These mimic the results. Add new mock files here, if you add more tests. # Note: The test are run in alphabetical order, so place new files at the # correct position. response_list = [ "Blast/mock_actin.xml", # result for test_blastp_nr_actin "Blast/mock_disco.xml", # result for test_discomegalast "Blast/mock_orchid.xml", # result for test_orchid_est "Blast/mock_pcr.xml", # result for test_pcr_primers "Blast/mock_short_empty.xml", # result for test_short_query # 1 "Blast/mock_short_result.xml", # result for test_short_query # 2 "Blast/mock_short_result.xml", # result for test_short_query # 3 ] # Generate a list of responses with the structure wait|result|wait|result... responses = ( BytesIO(open(a, "rb").read()) for b in zip(len(response_list) * wait, response_list) for a in b ) return responses NCBIWWW.time.sleep = mock.Mock() # we don't want to wait 20 sec for each test... NCBIWWW.urlopen = mock.Mock(side_effect=mock_response()) ##################################################################### # List of qblast requests stored as a tuple of parameters: # - program # - database # - query identifier or sequence # - expectation value threshold # - Entrez filter string (or None) # - list of hit identifiers expected to be found (or None if expect 0) class TestQblast(unittest.TestCase): def test_blastp_nr_actin(self): # Simple protein blast filtered for rat only, using protein # GI:160837788 aka NP_075631.2 # the actin related protein 2/3 complex, subunit 1B [Mus musculus] self.run_qblast( "blastp", "nr", "NP_075631.2", 0.001, "rat [ORGN]", {"megablast": "FALSE"}, [ "NP_112408.1", "AAH59131.1", "EDM14357.1", "NP_001008766.1", "NP_001102411.1", "EDL80109.1", "EDL80106.1", "NP_001100434.1", "AAI67084.1", ], ) def test_pcr_primers(self): # This next example finds PCR primer matches in Chimpanzees, e.g. BRCA1: self.run_qblast( "blastn", "nr", "GTACCTTGATTTCGTATTC" + ("N" * 30) + "GACTCTACTACCTTTACCC", 10, "pan [ORGN]", {"megablast": "FALSE"}, [ "XM_034941187.1", "XM_034941186.1", "XM_034941185.1", "XM_034941184.1", "XM_034941183.1", "XM_034941182.1", "XM_034941180.1", "XM_034941179.1", "XM_034941178.1", "XM_034941177.1", ], ) def test_orchid_est(self): # Try an orchid EST (nucleotide) sequence against NR using BLASTX self.run_qblast( "blastx", "nr", """>gi|116660609|gb|EG558220.1|EG558220 CR02019H04 Leaf CR02 cDNA library Catharanthus roseus cDNA clone CR02019H04 5', mRNA sequence CTCCATTCCCTCTCTATTTTCAGTCTAATCAAATTAGAGCTTAAAAGAATGAGATTTTTAACAAATAAAA AAACATAGGGGAGATTTCATAAAAGTTATATTAGTGATTTGAAGAATATTTTAGTCTATTTTTTTTTTTT TCTTTTTTTGATGAAGAAAGGGTATATAAAATCAAGAATCTGGGGTGTTTGTGTTGACTTGGGTCGGGTG TGTATAATTCTTGATTTTTTCAGGTAGTTGAAAAGGTAGGGAGAAAAGTGGAGAAGCCTAAGCTGATATT GAAATTCATATGGATGGAAAAGAACATTGGTTTAGGATTGGATCAAAAAATAGGTGGACATGGAACTGTA CCACTACGTCCTTACTATTTTTGGCCGAGGAAAGATGCTTGGGAAGAACTTAAAACAGTTTTAGAAAGCA AGCCATGGATTTCTCAGAAGAAAATGATTATACTTCTTAATCAGGCAACTGATATTATCAATTTATGGCA GCAGAGTGGTGGCTCCTTGTCCCAGCAGCAGTAATTACTTTTTTTTCTCTTTTTGTTTCCAAATTAAGAA ACATTAGTATCATATGGCTATTTGCTCAATTGCAGATTTCTTTCTTTTGTGAATG""", 0.0000001, None, {"megablast": "FALSE"}, [ "XP_021665344.1", "XP_021615158.1", "XP_017223689.1", "OMP06800.1", "XP_021634873.1", "XP_021299673.1", "XP_002311451.2", "XP_021976565.1", "OMO90244.1", ], ) def test_discomegablast(self): self.run_qblast( "blastn", "nr", """>some sequence ATGAAGATCTTCCAGATCCAGTGCAGCAGCTTCAAGGAGAGCAGGTGGCAGAAGAGCAAGTGCGACAACT GCCTGAAGTTCCACATCGACATCAACAACAACAGCAAGACCAGCAACACCGACACCGACTTCGACGCCAA CACCAACATCAACAGCAACATCAACAGCAACATCAACAGCAACATCAACATCAACAACAGCGGCAACAAC AACAAGAACAGCAACAACATCGAGATCACCGAGAACATCGACAACAAGGCCAAGATCATCAACAAGCACA TCAAGACCATCACCAACAGCAAGCCCATCCCCATCCCCATCCCCACCCCCACCCCCATCAGCATCAAGGA GAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGAAGGAGATG AAGAGCACCATCAACCTGAGGAGCGAGGACACCACCAGCAACAAGAGCACCATCGTGTTCACCGAGTGCC TGGAGTACAAGGGCCACCAGTGGAGGCCCAACATCTGCGTGACCTGCTTCAGCCCCAAGAACAAGCACAA GAACGTGCTGCCCGAGACCAGCACCCCCCTGATCAGCCAGAGCAGCCAGACCAGCACCATCACCCCCAGC AGCAGCAGCACCAGCACCAGCACCAGCAGCATCAGCACCCACAAGACCGCCAACAACAAGACCGTGATCA CCTACATCAGCAGCACCACCACCACCACCACCACCAGCAGCAGCAGCAGCAGCCCCCCCAGCAGCAGCAT CGCCGGCATCACCAACCCCACCAGCAGGAGCAGCAGCCCCATCCTGAAGAGCGTGCCCCCCAGCGCCTAC AGCAACGTGGTGATCCCCATCAACAACATCAACAACAGCAACAGCAACAGCAGCAGCGGCGGCGGCAACA ACAACAACAAGAGCATCAGCACCCCCAGCAGCCCCATCATCAGCAGGCCCATCACCAACAAGATCAACAA CAACAACAACAACAACCAGCCCCAGCTGCACTACAACCAGCCCCAGAGCAGCAGCGTGAGCACCACCAGC AGCCCCATCATCAGGCCCGTGCTGAGGAGGCAGTTCCAGAGCTTCCCCAGCAACCCCAAGATCAGCAAGG CCATCCTGGAGCAGTGCAACATCATCAACAACAACAGCAACAGCAACAACAGCAACAACAAGGACCCCGT GATCCTGTGCAAGTACACCATCGAGAGCCAGCCCAAGAGCAACATCAGCGTGCTGAAGCCCACCCTGGTG GAGTTCATCAACCAGCCCGACAGCAAGGACGACGAGAGCAGCGTGAAGAGCCCCCCCCTGCCCGTGGAGA GCCAGCCCATCTTCAACAGCAAGCAGAGCGCCACCATGGACGGCATCACCACCCACAAGAGCGTGAGCAT CACCATCAGCACCAGCACCAGCCCCAGCAGCACCACCACCACCACCAGCACCACCACCAGCATCATCGCC GAGGAGCCCAGCAGCCCCATCCTGCCCACCGCCAGCCCCAGCAGCAGCAGCAGCAGCATCATCACCACCG CCACCGCCAGCACCATCCCCATGAGCCCCAGCCTGCCCAGCATCCCCTTCCACGAGTTCGAGACCATGGA GAGCAGCACCACCACCACCCTGCTGAGCGAGAACAACGGCGGCGGCGGCGGCAGCAGCTGCAACGACAAC AGCAGGAGGAACAGCCTGAACATCCTGCCCCTGAGGCTGAAGAGCTTCAGCTTCAGCGCCCCCCAGAGCG ACAGCATGATCGAGCAGCCCGAGGACGACCCCTTCTTCGACTTCGAGGACCTGAGCGACGACGACGACAG CAACGACAACGACGACGAGGAGCTGAAGGAGATCAACGGCGAGAAGATCATCCAGCAGAACGACCTGACC CCCACCACCACCATCACCAGCACCACCACCATCCTGCAGAGCCCCACCCTGGAGAAGACCCTGAGCACCA CCACCACCACCACCATCCCCAGCCCCAGCACCAACAGCAGGAGCATCTGCAACACCCTGATGGACAGCAC CGACAGCATCAACAACACCAACACCAACACCAACACCAACACCAACACCAACACCAACACCAACACCAAC ACCAACACCAACACCAACACCAACGCCAACATCAACAACAAGGTGAGCACCACCACCACCACCACCACCA CCAAGAGGAGGAGCCTGAAGATGGACCAGTTCAAGGAGAAGGAGGACGAGTGGGACCAGGGCGTGGACCT GACCAGCTTCCTGAAGAGGAAGCCCACCCTGCAGAGGGACTTCAGCTACTGCAACAACAAGGTGATGGAG ATCAGCAGCGTGAAGGAGGAGGCCAAGAGGCTGCACGGCGGCACCGGCTACATCCACCAGTTCGCCTTCG AGGCCTTCAAGGACATCCTGGAGGCCAAGCAGACCCAGATCAACAGGGCCTTCTGCAGCCAGAAGATCGA CGCCCCCGACTGCGAGATGCTGATCAACGAGATCAACACCGCCAAGAAGCTGCTGGAGGACCTGCTGGAG CTGAACAGCAACAGCAGCGGCAGCGGCAACAACAGCAACGACAACAGCGGCAGCAGCAGCCCCAGCAGCA GCAAGACCAACACCCTGAACCAGCAGAGCATCTGCATCAAGAGCGAGATCCAACGATACGTTGAAATTCG CTTGTGTGCCACTGGTAAATCCACCCCCCCTAAGCCTCTAATAGGGAGACCTTAG""", 0.0000001, None, {"template_type": 0, "template_length": 18, "megablast": "on"}, # noqa 231 ["XM_635681.1", "XM_008496783.1"], ) def run_qblast( self, program, database, query, e_value, entrez_filter, additional_args, expected_hits, ): """Do qblast searches with given parameters and analyze results.""" try: if program == "blastn": # Check the megablast parameter is accepted handle = NCBIWWW.qblast( program, database, query, alignments=10, descriptions=10, hitlist_size=10, entrez_query=entrez_filter, expect=e_value, **additional_args, ) else: handle = NCBIWWW.qblast( program, database, query, alignments=10, descriptions=10, hitlist_size=10, entrez_query=entrez_filter, expect=e_value, **additional_args, ) except HTTPError: # e.g. a proxy error raise MissingExternalDependencyError("internet connection failed") from None record = NCBIXML.read(handle) if record.query == "No definition line": # We used a sequence as the query self.assertEqual(len(query), record.query_letters) elif query.startswith(">"): # We used a FASTA record as the query expected = query[1:].split("\n", 1)[0] self.assertEqual(expected, record.query) elif ( record.query_id.startswith("Query_") and len(query) == record.query_letters ): # We used a sequence as the entry and it was given a placeholder name pass else: # We used an identifier as the query self.assertIn( query, record.query_id.split("|"), f"Expected {query!r} within query_id {record.query_id!r}", ) # Check the recorded input parameters agree with those requested self.assertEqual(float(record.expect), e_value) self.assertEqual(record.application.lower(), program) self.assertLessEqual(len(record.alignments), 10) self.assertLessEqual(len(record.descriptions), 10) # Check the expected result(s) are found in the alignments if expected_hits is None: self.assertEqual(len(record.alignments), 0) # Expected no alignments! else: self.assertGreater(len(record.alignments), 0) # Expected some alignments! found_result = False for expected_hit in expected_hits: for alignment in record.alignments: if expected_hit in alignment.hit_id.split("|"): found_result = True break self.assertTrue( found_result, "Missing all expected hits (%s), instead have: %s" % ( ", ".join(expected_hits), ", ".join(a.hit_id for a in record.alignments), ), ) # Check the expected result(s) are found in the descriptions if expected_hits is None: # Expected no descriptions! self.assertEqual(len(record.descriptions), 0) else: # Expected some descriptions! self.assertGreater(len(record.descriptions), 0) found_result = False for expected_hit in expected_hits: for descr in record.descriptions: if ( expected_hit == descr.accession or expected_hit in descr.title.split(None, 1)[0].split("|") ): found_result = True break msg = f"Missing all of {expected_hit} in descriptions" self.assertTrue(found_result, msg=msg) def test_parse_qblast_ref_page(self): with open("Blast/html_msgid_29_blastx_001.html", "rb") as f: handle = BytesIO(f.read()) self.assertRaises(ValueError, NCBIWWW._parse_qblast_ref_page, handle) def test_short_query(self): """Test SHORT_QUERY_ADJUST parameter.""" # Should give no hits: my_search = NCBIWWW.qblast("blastp", "nr", "ICWENRMP", hitlist_size=5) my_hits = NCBIXML.read(my_search) my_search.close() self.assertEqual(len(my_hits.alignments), 0) # Should give hits: my_search = NCBIWWW.qblast( "blastp", "nr", "ICWENRMP", hitlist_size=5, short_query=True ) my_hits = NCBIXML.read(my_search) my_search.close() self.assertNotEqual(len(my_hits.alignments), 0) # Disabled for 1.81 release # Query does not seem to be returning anything so # maybe the test needs to be updated? # # with warnings.catch_warnings(record=True) as w: # # Cause all warnings to always be triggered. # warnings.simplefilter("always") # # Trigger a warning. # my_search = NCBIWWW.qblast( # "blastn", "nt", "ATGTCAACTTCAGAA", hitlist_size=5, short_query=True # ) # # Verify some things # self.assertEqual(w[0].category, BiopythonWarning) # self.assertIn("blastn", str(w[0].message)) # my_hits = NCBIXML.read(my_search) # my_search.close() # self.assertNotEqual(len(my_hits.alignments), 0) def test_error_conditions(self): """Test if exceptions were properly handled.""" self.assertRaises( ValueError, NCBIWWW.qblast, "megablast", "nt", "ATGCGTACGCAGCTAAAGTAAACCTATCGCGTCTCCT", ) if __name__ == "__main__": runner = unittest.TextTestRunner(verbosity=2) unittest.main(testRunner=runner)