[2] | 1 | import unittest |
---|
| 2 | import galaxy.model.mapping as mapping |
---|
| 3 | from galaxy.model import directory_hash_id |
---|
| 4 | import os.path |
---|
| 5 | |
---|
| 6 | class MappingTests( unittest.TestCase ): |
---|
| 7 | def test_basic( self ): |
---|
| 8 | # Start the database and connect the mapping |
---|
| 9 | model = mapping.init( "/tmp", "sqlite:///:memory:", create_tables=True ) |
---|
| 10 | assert model.engine is not None |
---|
| 11 | # Make some changes and commit them |
---|
| 12 | u = model.User( email="james@foo.bar.baz", password="password" ) |
---|
| 13 | # gs = model.GalaxySession() |
---|
| 14 | h1 = model.History( name="History 1", user=u) |
---|
| 15 | #h1.queries.append( model.Query( "h1->q1" ) ) |
---|
| 16 | #h1.queries.append( model.Query( "h1->q2" ) ) |
---|
| 17 | h2 = model.History( name=( "H" * 1024 ) ) |
---|
| 18 | model.session.add_all( ( u, h1, h2 ) ) |
---|
| 19 | #q1 = model.Query( "h2->q1" ) |
---|
| 20 | d1 = model.HistoryDatasetAssociation( extension="interval", metadata=dict(chromCol=1,startCol=2,endCol=3 ), history=h2, create_dataset=True, sa_session=model.session ) |
---|
| 21 | #h2.queries.append( q1 ) |
---|
| 22 | #h2.queries.append( model.Query( "h2->q2" ) ) |
---|
| 23 | model.session.add( ( d1 ) ) |
---|
| 24 | model.session.flush() |
---|
| 25 | model.session.expunge_all() |
---|
| 26 | # Check |
---|
| 27 | users = model.session.query( model.User ).all() |
---|
| 28 | assert len( users ) == 1 |
---|
| 29 | assert users[0].email == "james@foo.bar.baz" |
---|
| 30 | assert users[0].password == "password" |
---|
| 31 | assert len( users[0].histories ) == 1 |
---|
| 32 | assert users[0].histories[0].name == "History 1" |
---|
| 33 | hists = model.session.query( model.History ).all() |
---|
| 34 | assert hists[0].name == "History 1" |
---|
| 35 | assert hists[1].name == ( "H" * 255 ) |
---|
| 36 | assert hists[0].user == users[0] |
---|
| 37 | assert hists[1].user is None |
---|
| 38 | assert hists[1].datasets[0].metadata.chromCol == 1 |
---|
| 39 | id = hists[1].datasets[0].id |
---|
| 40 | assert hists[1].datasets[0].file_name == os.path.join( "/tmp", *directory_hash_id( id ) ) + ( "/dataset_%d.dat" % id ) |
---|
| 41 | # Do an update and check |
---|
| 42 | hists[1].name = "History 2b" |
---|
| 43 | model.session.flush() |
---|
| 44 | model.session.expunge_all() |
---|
| 45 | hists = model.session.query( model.History ).all() |
---|
| 46 | assert hists[0].name == "History 1" |
---|
| 47 | assert hists[1].name == "History 2b" |
---|
| 48 | # gvk TODO need to ad test for GalaxySessions, but not yet sure what they should look like. |
---|
| 49 | |
---|
| 50 | def get_suite(): |
---|
| 51 | suite = unittest.TestSuite() |
---|
| 52 | suite.addTest( MappingTests( "test_basic" ) ) |
---|
| 53 | return suite |
---|