Back to index

python-biopython  1.60
test_BioSQL.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 # This code is part of the Biopython distribution and governed by its
00003 # license.  Please see the LICENSE file that should have been included
00004 # as part of this package.
00005 """Tests for dealing with storage of biopython objects in a relational db.
00006 """
00007 # standard library
00008 import os
00009 import unittest
00010 from StringIO import StringIO
00011 
00012 # local stuff
00013 from Bio import MissingExternalDependencyError
00014 from Bio.Seq import Seq, MutableSeq
00015 from Bio.SeqFeature import SeqFeature
00016 from Bio import Alphabet
00017 from Bio import SeqIO
00018 from Bio.SeqRecord import SeqRecord
00019 
00020 from BioSQL import BioSeqDatabase
00021 from BioSQL import BioSeq
00022 
00023 # This testing suite should try to detect whether a valid database
00024 # installation exists on this computer.  Only run the tests if it
00025 # does.
00026 try:
00027     from setup_BioSQL import DBDRIVER, DBTYPE
00028     from setup_BioSQL import DBHOST, DBUSER, DBPASSWD, TESTDB
00029     from setup_BioSQL import DBSCHEMA, SQL_FILE
00030 except (NameError, ImportError):
00031     message = "Check settings in Tests/setup_BioSQL.py "\
00032               "if you plan to use BioSQL."
00033     raise MissingExternalDependencyError(message)
00034 
00035 try:
00036     if DBDRIVER in ["sqlite3"]:
00037         server = BioSeqDatabase.open_database(driver = DBDRIVER, db = TESTDB)
00038     else:
00039         server = BioSeqDatabase.open_database(driver = DBDRIVER,
00040                                               user = DBUSER, passwd = DBPASSWD,
00041                                               host = DBHOST)
00042     server.close()
00043     del server
00044 except Exception, e:
00045     message = "Connection failed, check settings in Tests/setup_BioSQL.py "\
00046               "if you plan to use BioSQL: %s" % str(e)
00047     raise MissingExternalDependencyError(message)
00048 
00049 from seq_tests_common import compare_record, compare_records
00050 
00051 def _do_db_create():
00052     """Do the actual work of database creation. Relevant for MySQL and PostgreSQL
00053     """
00054     # first open a connection to create the database
00055     server = BioSeqDatabase.open_database(driver = DBDRIVER,
00056                                           user = DBUSER, passwd = DBPASSWD,
00057                                           host = DBHOST)
00058 
00059     if DBDRIVER == "pgdb":
00060         # The pgdb postgres driver does not support autocommit, so here we 
00061         # commit the current transaction so that 'drop database' query will
00062         # be outside a transaction block
00063         server.adaptor.cursor.execute("COMMIT")
00064     else:
00065         # Auto-commit: postgresql cannot drop database in a transaction
00066         try:
00067             server.adaptor.autocommit()
00068         except AttributeError:
00069             pass
00070 
00071     # drop anything in the database
00072     try:
00073         # with Postgres, can get errors about database still being used and
00074         # not able to be dropped. Wait briefly to be sure previous tests are
00075         # done with it.
00076         import time
00077         time.sleep(1)
00078 
00079         sql = r"DROP DATABASE " + TESTDB
00080         server.adaptor.cursor.execute(sql, ())
00081     except (server.module.OperationalError,
00082             server.module.DatabaseError), e: # the database doesn't exist
00083         pass
00084     except (server.module.IntegrityError,
00085             server.module.ProgrammingError), e: # ditto--perhaps
00086         if str(e).find('database "%s" does not exist' % TESTDB) == -1:
00087             raise
00088     # create a new database
00089     sql = r"CREATE DATABASE " + TESTDB
00090     server.adaptor.execute(sql, ())
00091     server.close()
00092 
00093 def create_database():
00094     """Delete any existing BioSQL test database, then (re)create an empty BioSQL database."""
00095     if DBDRIVER in ["sqlite3"]: 
00096         if os.path.exists(TESTDB):
00097             os.remove(TESTDB)
00098     else:
00099         _do_db_create()
00100 
00101     # now open a connection to load the database
00102     server = BioSeqDatabase.open_database(driver = DBDRIVER,
00103                                           user = DBUSER, passwd = DBPASSWD,
00104                                           host = DBHOST, db = TESTDB)
00105     server.load_database_sql(SQL_FILE)
00106     server.commit()
00107     server.close()
00108 
00109 def load_database(gb_handle):
00110     """Load a GenBank file into a new BioSQL database.
00111     
00112     This is useful for running tests against a newly created database.
00113     """
00114 
00115     create_database()
00116     # now open a connection to load the database
00117     db_name = "biosql-test"
00118     server = BioSeqDatabase.open_database(driver = DBDRIVER,
00119                                           user = DBUSER, passwd = DBPASSWD,
00120                                           host = DBHOST, db = TESTDB)
00121     db = server.new_database(db_name)
00122     
00123     # get the GenBank file we are going to put into it
00124     iterator = SeqIO.parse(gb_handle, "gb")
00125     # finally put it in the database
00126     count = db.load(iterator)
00127     server.commit()
00128     server.close()
00129     return count
00130 
00131 class ReadTest(unittest.TestCase):
00132     """Test reading a database from an already built database.
00133     """
00134     loaded_db = 0
00135     
00136     def setUp(self):
00137         """Connect to and load up the database.
00138         """
00139         gb_file = "GenBank/cor6_6.gb"
00140         gb_handle = open(gb_file, "r")
00141         load_database(gb_handle)
00142         gb_handle.close()
00143             
00144         self.server = BioSeqDatabase.open_database(driver = DBDRIVER,
00145                                               user = DBUSER, 
00146                                               passwd = DBPASSWD,
00147                                               host = DBHOST, db = TESTDB)
00148             
00149         self.db = self.server["biosql-test"]
00150 
00151     def tearDown(self):
00152         self.server.close()
00153         del self.db
00154         del self.server
00155 
00156     def test_server(self):
00157         """Check BioSeqDatabase methods"""
00158         server = self.server
00159         self.assertTrue("biosql-test" in server)
00160         self.assertEqual(1, len(server))
00161         self.assertEqual(["biosql-test"], server.keys())
00162         #Check we can delete the namespace...
00163         del server["biosql-test"]
00164         self.assertEqual(0, len(server))
00165         try:
00166             del server["non-existant-name"]
00167             assert False, "Should have raised KeyError"
00168         except KeyError:
00169             pass
00170 
00171     def test_get_db_items(self):
00172         """Check list, keys, length etc"""
00173         db = self.db
00174         items = db.values()
00175         keys = db.keys()
00176         l = len(items)
00177         self.assertEqual(l, len(db))
00178         self.assertEqual(l, len(list(db.iteritems())))
00179         self.assertEqual(l, len(list(db.iterkeys())))
00180         self.assertEqual(l, len(list(db.itervalues())))
00181         for (k1,r1), (k2,r2) in zip(zip(keys, items), db.iteritems()):
00182             self.assertEqual(k1, k2)
00183             self.assertEqual(r1.id, r2.id)
00184         for k in keys:
00185             del db[k]
00186         self.assertEqual(0, len(db))
00187         try:
00188             del db["non-existant-name"]
00189             assert False, "Should have raised KeyError"
00190         except KeyError:
00191             pass
00192 
00193     def test_lookup_items(self):
00194         """Test retrieval of items using various ids.
00195         """
00196         item = self.db.lookup(accession = "X62281")
00197         try:
00198             item = self.db.lookup(accession = "Not real")
00199             raise Assertionerror("No problem on fake id retrieval")
00200         except IndexError:
00201             pass
00202         item = self.db.lookup(display_id = "ATKIN2")
00203         try:
00204             item = self.db.lookup(display_id = "Not real")
00205             raise AssertionError("No problem on fake id retrieval")
00206         except IndexError:
00207             pass
00208         
00209         # primary id retrieval
00210         item = self.db.lookup(primary_id = "16353")
00211         try:
00212             item = self.db.lookup(primary_id = "Not Real")
00213             raise AssertionError("No problem on fake primary id retrieval")
00214         except IndexError:
00215             pass
00216 
00217 class SeqInterfaceTest(unittest.TestCase):
00218     """Make sure the BioSQL objects implement the expected biopython interfaces
00219     """
00220     def setUp(self):
00221         """Load a database.
00222         """
00223         gb_file = os.path.join(os.getcwd(), "GenBank", "cor6_6.gb")
00224         gb_handle = open(gb_file, "r")
00225         load_database(gb_handle)
00226         gb_handle.close()
00227 
00228         self.server = BioSeqDatabase.open_database(driver = DBDRIVER,
00229                                               user = DBUSER, passwd = DBPASSWD,
00230                                               host = DBHOST, db = TESTDB)
00231         self.db = self.server["biosql-test"]
00232         self.item = self.db.lookup(accession = "X62281")
00233 
00234     def tearDown(self):
00235         self.server.close()
00236         del self.db
00237         del self.item
00238         del self.server
00239     
00240     def test_seq_record(self):
00241         """Make sure SeqRecords from BioSQL implement the right interface.
00242         """
00243         test_record = self.item
00244         self.assertTrue(isinstance(test_record.seq, BioSeq.DBSeq))
00245         self.assertEqual(test_record.id, "X62281.1", test_record.id)
00246         self.assertEqual(test_record.name, "ATKIN2")
00247         self.assertEqual(test_record.description, "A.thaliana kin2 gene.")
00248         annotations = test_record.annotations
00249         # XXX should do something with annotations once they are like
00250         # a dictionary
00251         for feature in test_record.features:
00252             self.assertTrue(isinstance(feature, SeqFeature))
00253         s = str(test_record) #shouldn't cause any errors!
00254 
00255     def test_seq(self):
00256         """Make sure Seqs from BioSQL implement the right interface.
00257         """
00258         test_seq = self.item.seq
00259         alphabet = test_seq.alphabet
00260         self.assertTrue(isinstance(alphabet, Alphabet.Alphabet))
00261         data = test_seq.data
00262         self.assertEqual(type(data), type(""))
00263         string_rep = test_seq.tostring()
00264         self.assertEqual(string_rep, str(test_seq)) #check __str__ too
00265         self.assertEqual(type(string_rep), type(""))
00266         self.assertEqual(len(test_seq), 880)
00267         
00268     def test_convert(self):
00269         """Check can turn a DBSeq object into a Seq or MutableSeq."""
00270         test_seq = self.item.seq
00271 
00272         other = test_seq.toseq()
00273         self.assertEqual(str(test_seq), str(other))
00274         self.assertEqual(test_seq.alphabet, other.alphabet)
00275         self.assertTrue(isinstance(other, Seq))
00276 
00277         other = test_seq.tomutable()
00278         self.assertEqual(str(test_seq), str(other))
00279         self.assertEqual(test_seq.alphabet, other.alphabet)
00280         self.assertTrue(isinstance(other, MutableSeq))
00281 
00282     def test_addition(self):
00283         """Check can add DBSeq objects together."""
00284         test_seq = self.item.seq
00285         for other in [Seq("ACGT",test_seq.alphabet),
00286                       MutableSeq("ACGT",test_seq.alphabet),
00287                       "ACGT",
00288                       test_seq]:
00289             test = test_seq + other
00290             self.assertEqual(str(test), str(test_seq) + str(other))
00291             self.assertTrue(isinstance(test, Seq))
00292             test = other + test_seq
00293             self.assertEqual(str(test), str(other) + str(test_seq))
00294 
00295     def test_seq_slicing(self):
00296         """Check that slices of sequences are retrieved properly.
00297         """
00298         test_seq = self.item.seq
00299         new_seq = test_seq[:10]
00300         self.assertTrue(isinstance(new_seq, BioSeq.DBSeq))
00301         # simple slicing
00302         self.assertEqual(test_seq[:5].tostring(), 'ATTTG')
00303         self.assertEqual(test_seq[0:5].tostring(), 'ATTTG')
00304         self.assertEqual(test_seq[2:3].tostring(), 'T')
00305         self.assertEqual(test_seq[2:4].tostring(), 'TT')
00306         self.assertEqual(test_seq[870:].tostring(), 'TTGAATTATA')
00307         # getting more fancy
00308         self.assertEqual(test_seq[-1], 'A')
00309         self.assertEqual(test_seq[1], 'T')
00310         self.assertEqual(test_seq[-10:][5:].tostring(), "TTATA")
00311         self.assertEqual(str(test_seq[-10:][5:]), "TTATA")
00312 
00313     def test_seq_features(self):
00314         """Check SeqFeatures of a sequence.
00315         """
00316         test_features = self.item.features
00317         cds_feature = test_features[6]
00318         self.assertEqual(cds_feature.type, "CDS")
00319         self.assertEqual(str(cds_feature.location), "[103:579](+)")
00320         for sub_feature in cds_feature.sub_features:
00321             self.assertEqual(sub_feature.type, "CDS")
00322             self.assertEqual(sub_feature.location_operator, "join")
00323 
00324         try:
00325             self.assertEqual(cds_feature.qualifiers["gene"], ["kin2"])
00326             self.assertEqual(cds_feature.qualifiers["protein_id"], ["CAA44171.1"])
00327             self.assertEqual(cds_feature.qualifiers["codon_start"], ["1"])
00328         except KeyError:
00329             raise KeyError("Missing expected entries, have %s" \
00330                            % repr(cds_feature.qualifiers))
00331         
00332         self.assertTrue("db_xref" in cds_feature.qualifiers)
00333         multi_ann = cds_feature.qualifiers["db_xref"]
00334         self.assertEqual(len(multi_ann), 2)
00335         self.assertTrue("GI:16354" in multi_ann)
00336         self.assertTrue("SWISS-PROT:P31169" in multi_ann)
00337 
00338 class LoaderTest(unittest.TestCase):
00339     """Load a database from a GenBank file.
00340     """
00341     def setUp(self):
00342         # create TESTDB
00343         create_database()
00344         
00345         # load the database
00346         db_name = "biosql-test"
00347         self.server = BioSeqDatabase.open_database(driver = DBDRIVER,
00348                                               user = DBUSER, passwd = DBPASSWD,
00349                                               host = DBHOST, db = TESTDB)
00350         
00351         # remove the database if it already exists
00352         try:
00353             self.server[db_name]
00354             self.server.remove_database(db_name)
00355         except KeyError:
00356             pass
00357         
00358         self.db = self.server.new_database(db_name)
00359 
00360         # get the GenBank file we are going to put into it
00361         input_file = os.path.join(os.getcwd(), "GenBank", "cor6_6.gb")
00362         handle = open(input_file, "r")
00363         self.iterator = SeqIO.parse(handle, "gb")
00364 
00365     def tearDown(self):
00366         self.server.close()
00367         del self.db
00368         del self.server
00369 
00370     def test_load_database(self):
00371         """Load SeqRecord objects into a BioSQL database.
00372         """
00373         self.db.load(self.iterator)
00374 
00375         # do some simple tests to make sure we actually loaded the right
00376         # thing. More advanced tests in a different module.
00377         items = self.db.values()
00378         self.assertEqual(len(items), 6)
00379         self.assertEqual(len(self.db), 6)
00380         item_names = []
00381         item_ids = []
00382         for item in items:
00383             item_names.append(item.name)
00384             item_ids.append(item.id)
00385         item_names.sort()
00386         item_ids.sort()
00387         self.assertEqual(item_names, ['AF297471', 'ARU237582', 'ATCOR66M',
00388                                       'ATKIN2', 'BNAKINI', 'BRRBIF72'])
00389         self.assertEqual(item_ids, ['AF297471.1', 'AJ237582.1', 'L31939.1',
00390                                     'M81224.1', 'X55053.1', 'X62281.1'])
00391 
00392 class DupLoadTest(unittest.TestCase):
00393     """Check a few duplicate conditions fail."""
00394     def setUp(self):
00395         #drop any old database and create a new one:
00396         create_database()
00397         #connect to new database:
00398         self.server = BioSeqDatabase.open_database(driver = DBDRIVER,
00399                                               user = DBUSER, passwd = DBPASSWD,
00400                                               host = DBHOST, db = TESTDB)
00401         #Create new namespace within new empty database:
00402         self.db = self.server.new_database("biosql-test")
00403 
00404     def tearDown(self):
00405         self.server.rollback()
00406         self.server.close()
00407         del self.db
00408         del self.server
00409 
00410     def test_duplicate_load(self):
00411         """Make sure can't import a single record twice (in one go)."""
00412         record = SeqRecord(Seq("ATGCTATGACTAT", Alphabet.generic_dna),id="Test1")
00413         try:
00414             count = self.db.load([record,record])
00415         except Exception, err:
00416             #Good!
00417             #Note we don't do a specific exception handler because the
00418             #exception class will depend on which DB back end is in use.            
00419             self.assertTrue(err.__class__.__name__ in ["IntegrityError",
00420                                                        "OperationalError"],
00421                             err.__class__.__name__)
00422             return
00423         raise Exception("Should have failed! Loaded %i records" % count)
00424 
00425     def test_duplicate_load2(self):
00426         """Make sure can't import a single record twice (in steps)."""
00427         record = SeqRecord(Seq("ATGCTATGACTAT", Alphabet.generic_dna),id="Test2")
00428         count = self.db.load([record])
00429         self.assertEqual(count,1)
00430         try:
00431             count = self.db.load([record])
00432         except Exception, err:
00433             #Good!
00434             self.assertEqual("IntegrityError", err.__class__.__name__)
00435             return
00436         raise Exception("Should have failed! Loaded %i records" % count)
00437 
00438     def test_duplicate_id_load(self):
00439         """Make sure can't import records with same ID (in one go)."""
00440         record1 = SeqRecord(Seq("ATGCTATGACTAT", Alphabet.generic_dna),id="TestA")
00441         record2 = SeqRecord(Seq("GGGATGCGACTAT", Alphabet.generic_dna),id="TestA")
00442         try:
00443             count = self.db.load([record1,record2])
00444         except Exception, err:
00445             #Good!
00446             self.assertEqual("IntegrityError", err.__class__.__name__)
00447             return
00448         raise Exception("Should have failed! Loaded %i records" % count)
00449 
00450 class ClosedLoopTest(unittest.TestCase):
00451     """Test file -> BioSQL -> file."""
00452     #NOTE - For speed I don't bother to create a new database each time,
00453     #simple a new unique namespace is used for each test.
00454     
00455     def test_NC_005816(self):
00456         """GenBank file to BioSQL and back to a GenBank file, NC_005816."""
00457         self.loop("GenBank/NC_005816.gb", "gb")
00458 
00459     def test_NC_000932(self):
00460         """GenBank file to BioSQL and back to a GenBank file, NC_000932."""
00461         self.loop("GenBank/NC_000932.gb", "gb")
00462 
00463     def test_NT_019265(self):
00464         """GenBank file to BioSQL and back to a GenBank file, NT_019265."""
00465         self.loop("GenBank/NT_019265.gb", "gb")
00466 
00467     def test_protein_refseq2(self):
00468         """GenBank file to BioSQL and back to a GenBank file, protein_refseq2."""
00469         self.loop("GenBank/protein_refseq2.gb", "gb")
00470 
00471     def test_no_ref(self):
00472         """GenBank file to BioSQL and back to a GenBank file, noref."""
00473         self.loop("GenBank/noref.gb", "gb")
00474 
00475     def test_one_of(self):
00476         """GenBank file to BioSQL and back to a GenBank file, one_of."""
00477         self.loop("GenBank/one_of.gb", "gb")
00478 
00479     def test_cor6_6(self):
00480         """GenBank file to BioSQL and back to a GenBank file, cor6_6."""
00481         self.loop("GenBank/cor6_6.gb", "gb")
00482 
00483     def test_arab1(self):
00484         """GenBank file to BioSQL and back to a GenBank file, arab1."""
00485         self.loop("GenBank/arab1.gb", "gb")
00486 
00487     def loop(self, filename, format):
00488         original_records = list(SeqIO.parse(open(filename, "rU"), format))
00489         # now open a connection to load the database
00490         server = BioSeqDatabase.open_database(driver = DBDRIVER,
00491                                               user = DBUSER, passwd = DBPASSWD,
00492                                               host = DBHOST, db = TESTDB)
00493         db_name = "test_loop_%s" % filename #new namespace!
00494         db = server.new_database(db_name)
00495         count = db.load(original_records)
00496         self.assertEqual(count, len(original_records))
00497         server.commit()
00498         #Now read them back...
00499         biosql_records = [db.lookup(name=rec.name) \
00500                           for rec in original_records]
00501         #And check they agree
00502         self.assertTrue(compare_records(original_records, biosql_records))
00503         #Now write to a handle...
00504         handle = StringIO()
00505         SeqIO.write(biosql_records, handle, "gb")
00506         #Now read them back...
00507         handle.seek(0)
00508         new_records = list(SeqIO.parse(handle, "gb"))
00509         #And check they still agree
00510         self.assertEqual(len(new_records), len(original_records))
00511         for old, new in zip(original_records, new_records):
00512             #TODO - remove this hack because we don't yet write these (yet):
00513             for key in ["comment", "references", "db_source"]:
00514                 if key in old.annotations and key not in new.annotations:
00515                     del old.annotations[key]
00516             self.assertTrue(compare_record(old, new))
00517         #Done
00518         server.close()
00519 
00520 class TransferTest(unittest.TestCase):
00521     """Test file -> BioSQL, BioSQL -> BioSQL."""
00522     #NOTE - For speed I don't bother to create a new database each time,
00523     #simple a new unique namespace is used for each test.
00524     
00525     def test_NC_005816(self):
00526         """GenBank file to BioSQL, then again to a new namespace, NC_005816."""
00527         self.trans("GenBank/NC_005816.gb", "gb")
00528 
00529     def test_NC_000932(self):
00530         """GenBank file to BioSQL, then again to a new namespace, NC_000932."""
00531         self.trans("GenBank/NC_000932.gb", "gb")
00532 
00533     def test_NT_019265(self):
00534         """GenBank file to BioSQL, then again to a new namespace, NT_019265."""
00535         self.trans("GenBank/NT_019265.gb", "gb")
00536 
00537     def test_protein_refseq2(self):
00538         """GenBank file to BioSQL, then again to a new namespace, protein_refseq2."""
00539         self.trans("GenBank/protein_refseq2.gb", "gb")
00540 
00541     def test_no_ref(self):
00542         """GenBank file to BioSQL, then again to a new namespace, noref."""
00543         self.trans("GenBank/noref.gb", "gb")
00544 
00545     def test_one_of(self):
00546         """GenBank file to BioSQL, then again to a new namespace, one_of."""
00547         self.trans("GenBank/one_of.gb", "gb")
00548 
00549     def test_cor6_6(self):
00550         """GenBank file to BioSQL, then again to a new namespace, cor6_6."""
00551         self.trans("GenBank/cor6_6.gb", "gb")
00552 
00553     def test_arab1(self):
00554         """GenBank file to BioSQL, then again to a new namespace, arab1."""
00555         self.trans("GenBank/arab1.gb", "gb")
00556 
00557     def trans(self, filename, format):
00558         original_records = list(SeqIO.parse(open(filename, "rU"), format))
00559         # now open a connection to load the database
00560         server = BioSeqDatabase.open_database(driver = DBDRIVER,
00561                                               user = DBUSER, passwd = DBPASSWD,
00562                                               host = DBHOST, db = TESTDB)
00563         db_name = "test_trans1_%s" % filename #new namespace!
00564         db = server.new_database(db_name)
00565         count = db.load(original_records)
00566         self.assertEqual(count, len(original_records))
00567         server.commit()
00568         #Now read them back...
00569         biosql_records = [db.lookup(name=rec.name) \
00570                           for rec in original_records]
00571         #And check they agree
00572         self.assertTrue(compare_records(original_records, biosql_records))
00573         #Now write to a second name space...
00574         db_name = "test_trans2_%s" % filename #new namespace!
00575         db = server.new_database(db_name)
00576         count = db.load(biosql_records)
00577         self.assertEqual(count, len(original_records))
00578         #Now read them back again,
00579         biosql_records2 = [db.lookup(name=rec.name) \
00580                           for rec in original_records]
00581         #And check they also agree
00582         self.assertTrue(compare_records(original_records, biosql_records2))
00583         #Done
00584         server.close()
00585 
00586 
00587 class InDepthLoadTest(unittest.TestCase):
00588     """Make sure we are loading and retreiving in a semi-lossless fashion.
00589     """
00590     def setUp(self):
00591         gb_file = os.path.join(os.getcwd(), "GenBank", "cor6_6.gb")
00592         gb_handle = open(gb_file, "r")
00593         load_database(gb_handle)
00594         gb_handle.close()
00595 
00596         self.server = BioSeqDatabase.open_database(driver = DBDRIVER,
00597                                               user = DBUSER, passwd = DBPASSWD,
00598                                               host = DBHOST, db = TESTDB)
00599         self.db = self.server["biosql-test"]
00600 
00601     def tearDown(self):
00602         self.server.close()
00603         del self.db
00604         del self.server
00605 
00606     def test_transfer(self):
00607         """Make sure can load record into another namespace."""
00608         #Should be in database already...
00609         db_record = self.db.lookup(accession = "X55053")
00610         #Make a new namespace
00611         db2 = self.server.new_database("biosql-test-alt")
00612         #Should be able to load this DBSeqRecord there...
00613         count = db2.load([db_record])
00614         self.assertEqual(count,1)
00615 
00616     def test_reload(self):
00617         """Make sure can't reimport existing records."""
00618         gb_file = os.path.join(os.getcwd(), "GenBank", "cor6_6.gb")
00619         gb_handle = open(gb_file, "r")
00620         record = SeqIO.parse(gb_handle, "gb").next()
00621         gb_handle.close()
00622         #Should be in database already...
00623         db_record = self.db.lookup(accession = "X55053")
00624         self.assertEqual(db_record.id, record.id)
00625         self.assertEqual(db_record.name, record.name)
00626         self.assertEqual(db_record.description, record.description)
00627         self.assertEqual(str(db_record.seq), str(record.seq))
00628         #Good... now try reloading it!
00629         try:
00630             count = self.db.load([record])
00631         except Exception, err:
00632             #Good!
00633             self.assertEqual("IntegrityError", err.__class__.__name__)
00634             return
00635         raise Exception("Should have failed! Loaded %i records" % count)
00636 
00637     def test_record_loading(self):
00638         """Make sure all records are correctly loaded.
00639         """
00640         test_record = self.db.lookup(accession = "X55053")
00641         self.assertEqual(test_record.name, "ATCOR66M")
00642         self.assertEqual(test_record.id, "X55053.1")
00643         self.assertEqual(test_record.description, "A.thaliana cor6.6 mRNA.")
00644         self.assertTrue(isinstance(test_record.seq.alphabet, Alphabet.DNAAlphabet))
00645         self.assertEqual(test_record.seq[:10].tostring(), 'AACAAAACAC')
00646 
00647         test_record = self.db.lookup(accession = "X62281")
00648         self.assertEqual(test_record.name, "ATKIN2")
00649         self.assertEqual(test_record.id, "X62281.1")
00650         self.assertEqual(test_record.description, "A.thaliana kin2 gene.")
00651         self.assertTrue(isinstance(test_record.seq.alphabet, Alphabet.DNAAlphabet))
00652         self.assertEqual(test_record.seq[:10].tostring(), 'ATTTGGCCTA')
00653 
00654     def test_seq_feature(self):
00655         """Indepth check that SeqFeatures are transmitted through the db.
00656         """
00657         test_record = self.db.lookup(accession = "AJ237582")
00658         features = test_record.features
00659         self.assertEqual(len(features), 7)
00660        
00661         # test single locations
00662         test_feature = features[0]
00663         self.assertEqual(test_feature.type, "source")
00664         self.assertEqual(str(test_feature.location), "[0:206](+)")
00665         self.assertEqual(len(test_feature.qualifiers.keys()), 3)
00666         self.assertEqual(test_feature.qualifiers["country"], ["Russia:Bashkortostan"])
00667         self.assertEqual(test_feature.qualifiers["organism"], ["Armoracia rusticana"])
00668         self.assertEqual(test_feature.qualifiers["db_xref"], ["taxon:3704"])
00669 
00670         # test split locations
00671         test_feature = features[4]
00672         self.assertEqual(test_feature.type, "CDS")
00673         self.assertEqual(str(test_feature.location), "[0:206](+)")
00674         self.assertEqual(len(test_feature.sub_features), 2)
00675         self.assertEqual(str(test_feature.sub_features[0].location), "[0:48](+)")
00676         self.assertEqual(test_feature.sub_features[0].type, "CDS")
00677         self.assertEqual(test_feature.sub_features[0].location_operator, "join")
00678         self.assertEqual(str(test_feature.sub_features[1].location), "[142:206](+)")
00679         self.assertEqual(test_feature.sub_features[1].type, "CDS")
00680         self.assertEqual(test_feature.sub_features[1].location_operator, "join")
00681         self.assertEqual(len(test_feature.qualifiers.keys()), 6)
00682         self.assertEqual(test_feature.qualifiers["gene"], ["csp14"])
00683         self.assertEqual(test_feature.qualifiers["codon_start"], ["2"])
00684         self.assertEqual(test_feature.qualifiers["product"],
00685                          ["cold shock protein"])
00686         self.assertEqual(test_feature.qualifiers["protein_id"], ["CAB39890.1"])
00687         self.assertEqual(test_feature.qualifiers["db_xref"], ["GI:4538893"])
00688         self.assertEqual(test_feature.qualifiers["translation"],
00689                          ["DKAKDAAAAAGASAQQAGKNISDAAAGGVNFVKEKTG"])
00690 
00691         # test passing strand information
00692         # XXX We should be testing complement as well
00693         test_record = self.db.lookup(accession = "AJ237582")
00694         test_feature = test_record.features[4] # DNA, no complement
00695         self.assertEqual(test_feature.strand, 1)
00696         for sub_feature in test_feature.sub_features:
00697             self.assertEqual(sub_feature.strand, 1)
00698 
00699         test_record = self.db.lookup(accession = "X55053")
00700         test_feature = test_record.features[0]
00701         # mRNA, so really cDNA, so the strand should be 1 (not complemented)
00702         self.assertEqual(test_feature.strand, 1)
00703 
00704 #Some of the unit tests don't create their own database,
00705 #so just in case there is no database already:
00706 create_database()
00707 
00708 if __name__ == "__main__":
00709     runner = unittest.TextTestRunner(verbosity = 2)
00710     unittest.main(testRunner=runner)