bauble 0.9.7-2 / usr / share / bauble / plugins / garden / test.py

import sys
import unittest

from sqlalchemy import *
from sqlalchemy.exc import *

from bauble.error import CheckConditionError, check
from bauble.test import BaubleTestCase
import bauble.utils as utils
from bauble.plugins.garden.accession import Accession, AccessionEditor, \
    dms_to_decimal, decimal_to_dms, longitude_to_dms, latitude_to_dms
from bauble.plugins.garden.donor import Donor, DonorEditor
from bauble.plugins.garden.source import Donation, Collection
from bauble.plugins.garden.plant import Plant, PlantEditor
from bauble.plugins.garden.location import Location, LocationEditor
from bauble.plugins.plants.family import Family
from bauble.plugins.plants.genus import Genus
from bauble.plugins.plants.species_model import Species
import bauble.plugins.plants.test as plants_test
from bauble.plugins.garden.institution import Institution

# TODO: create a test to make sure that if you delete an accession then the
# plants that are "children" of this accession are also deleted
from datetime import datetime
accession_test_data = ({'id':1 , 'code': u'1.1', 'species_id': 1,
                        'date': datetime.today(),
                        'source_type': u'Donation'},
                       {'id':2 , 'code': u'2.2', 'species_id': 2,
                        'source_type': u'Collection'},
                       )

plant_test_data = ({'id':1 , 'code': u'1', 'accession_id': 1,
                    'location_id': 1},
                   )

location_test_data = ({'id': 1, 'site': u'Somewhere Over The Rainbow'},
                      )

donor_test_data = ({'id': 1, 'name': u'SomeDonor'},
                   )

donation_test_data = ({'id': 1, 'accession_id': 1, 'donor_id': 1},
                      )

collection_test_data = ({'id': 1, 'accession_id': 2, 'locale': u'Somewhere'},
                        )

test_data_table_control = ((Accession, accession_test_data),
                           (Location, location_test_data),
                           (Plant, plant_test_data),
                           (Donor, donor_test_data),
                           (Donation, donation_test_data),
                           (Collection, collection_test_data))

def setUp_data():
    """
    create_test_data()
    #if this method is called again before tearDown_test_data is called you
    #will get an error about the test data rows already existing in the database
    """
    for cls, data in test_data_table_control:
        table = cls.__table__
        for row in data:
            table.insert().execute(row).close()
        for col in table.c:
            utils.reset_sequence(col)
    i = Institution()
    i.name = u'TestInstitution'
    i.technical_contact = u'TestTechnicalContact Name'
    i.email = u'contact@test.com'
    i.contact = u'TestContact Name'
    i.code = u'TestCode'



# TODO: things to create tests for
#
# - test all cascading works as expected
# - need to test that the Donor doesn't get deleted if it is orphaned since
# we don't want to ever throw out donor information

# TODO: if we ever get a GUI tester then do the following
# test all possible combinations of entering data into the accession editor
# 1. new accession without source
# 2. new accession with source
# 3. existing accession without source
# 4. existing accession with new source
# 5. existing accession with existing source
# - create test for parsing latitude/longitude entered into the lat/lon entries


class GardenTestCase(BaubleTestCase):

    def __init__(self, *args):
        super(GardenTestCase, self).__init__(*args)

    def setUp(self):
        super(GardenTestCase, self).setUp()
        plants_test.setUp_data()
        #setUp_test_data()
        self.family = Family(family=u'fam')
        self.genus = Genus(family=self.family, genus=u'gen')
        self.species = Species(genus=self.genus, sp=u'sp')
        self.session.add_all([self.family, self.genus, self.species])
        self.session.commit()

    # def tearDown(self):
    #     print >>sys.stderr, 'GardenTestCase.tearDown()'
    #     super(GardenTestCase, self).tearDown()

    def create(self, class_, **kwargs):
        obj = class_(**kwargs)
        self.session.add(obj)
        return obj


class DonorTests(GardenTestCase):

    def __init__(self, *args):
        super(DonorTests, self).__init__(*args)

    def test_delete(self):
        acc = self.create(Accession, species=self.species, code=u'1')
        donor = Donor(name=u'name')
        donation = Donation()
        donation.donor = donor
        acc.source = donation
        self.session.commit()
        self.session.close()

        # test that we can't delete a donor if it has corresponding donations
        import bauble
        session = bauble.Session()
        donor = session.query(Donor).filter_by(name=u'name').one()
        # shouldn't be allowed to delete donor if it has donations,
        # what is happening here is that when deleting the donor the
        # corresponding donations.donor_id's are being be set to null which
        # isn't allowed by the scheme....is this the best we can do? or can we
        # get some sort of error when creating a dangling reference
        session.delete(donor)
        self.assertRaises(SQLError, session.commit)


    def itest_donor_editor(self):
        """
        Interactively test the PlantEditor
        """
        loc = self.create(Donor, name=u'some donor')
        editor = DonorEditor(model=loc)
        editor.start()
        del editor
        assert utils.gc_objects_by_type('DonorEditor') == [], \
            'DonorEditor not deleted'
        assert utils.gc_objects_by_type('DonorEditorPresenter') == [], \
            'DonorEditorPresenter not deleted'
        assert utils.gc_objects_by_type('DonorEditorView') == [], \
            'DonorEditorView not deleted'



class PlantTests(GardenTestCase):

    def __init__(self, *args):
        super(PlantTests, self).__init__(*args)

    def setUp(self):
        super(PlantTests, self).setUp()

    def tearDown(self):
        super(PlantTests, self).tearDown()


    def test_constraints(self):
        acc = self.create(Accession, species=self.species, code=u'1')
        location = Location(site=u'site')
        plant = Plant(accession=acc, location=location, code=u'1')
        self.session.commit()

        # test that we can't have duplicate codes with the same accession
        plant2 = Plant(accession=acc, location=location, code=u'1')
        self.session.add(plant2)
        self.assertRaises(IntegrityError, self.session.commit)

    def test_delete(self):
        """
        Test that when a plant is deleted...
        """
        pass

    def itest_plant_editor(self):
        """
        Interactively test the PlantEditor
        """
        acc = self.create(Accession, species=self.species, code=u'1')
        location = Location(site=u'site')
        plant = Plant(accession=acc, location=location, code=u'1')
        self.session.commit()
        # create a second uncommited plant so that we can check for
        # uniqueness with an existing plant, i.e 1.1
        plant = Plant(accession=acc, location=location, code=u'2')
        editor = PlantEditor(model=plant)
        editor.start()
        del editor

        assert utils.gc_objects_by_type('PlantEditor') == [], \
            'PlantEditor not deleted'
        assert utils.gc_objects_by_type('PlantEditorPresenter') == [], \
            'PlantEditorPresenter not deleted'
        assert utils.gc_objects_by_type('PlantEditorView') == [], \
            'PlantEditorView not deleted'


class AccessionTests(GardenTestCase):

    def __init__(self, *args):
        super(AccessionTests, self).__init__(*args)

    def setUp(self):
        super(AccessionTests, self).setUp()


    def tearDown(self):
        super(AccessionTests, self).tearDown()


    def test_species_str(self):
        """
        Test Accesion.species_str()
        """
        acc = self.create(Accession, species=self.species, code=u'1')
        s = 'gen sp'
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' % (s, sp_str))
        acc.id_qual = '?'
        s = 'gen sp(?)'
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' % (s, sp_str))

        acc.id_qual = 'aff.'
        acc.id_qual_rank = 'sp'
        s = 'gen aff. sp'
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' %(s, sp_str))

        # here species.infrasp is None but we still allow the string
        acc.id_qual = 'cf.'
        acc.id_qual_rank = 'infrasp'
        s = 'gen sp cf. None'
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' %(s, sp_str))

        # species.infrasp is still none but these just get pasted on
        # the end so it doesn't matter
        acc.id_qual = 'incorrect'
        acc.id_qual_rank = 'infrasp'
        s = 'gen sp(incorrect)'
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' %(s, sp_str))

        acc.id_qual = 'forsan'
        acc.id_qual_rank = 'sp'
        s = 'gen sp(forsan)'
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' %(s, sp_str))

        acc.species.infrasp_rank = u'cv.'
        acc.species.infrasp = u'Cultivar'
        acc.id_qual = u'cf.'
        acc.id_qual_rank = u'infrasp'
        s = "gen sp cf. 'Cultivar'"
        sp_str = acc.species_str()
        self.assert_(s == sp_str, '%s == %s' %(s, sp_str))


        # test that the cached string is returned

        # have to commit because the cached string won't be returned
        # on dirty species
        self.session.commit()
        s2 = acc.species_str()
        assert id(sp_str) == id(s2), '%s(%s) == %s(%s)' % (sp_str, id(sp_str),
                                                           s2, id(s2))

        # this used to test that if the id_qual was set but the
        # id_qual_rank wasn't then we would get an error. now we just
        # show an warning and put the id_qual on the end of the string
#         acc.id_qual = 'aff.'
#         acc.id_qual_rank = None
#         self.assertRaises(CheckConditionError, acc.species_str)


    def test_delete(self):
        """
        Test that when an accession is deleted any orphaned rows are
        cleaned up.
        """
        acc = self.create(Accession, species=self.species, code=u'1')
        plant = self.create(Plant, accession=acc,
                            location=Location(site=u'site'), code=u'1')
        self.session.commit()

        # test that the plant is deleted after being orphaned
        plant_id = plant.id
        self.session.delete(acc)
        self.session.commit()
        self.assert_(not self.session.query(Plant).get(plant_id))

        # test that the donation and collection is deleted after being orphaned
        #is done in test_source
#         acc = acc = self.create(Accession, species=self.species, code=u'1')
#         coll = Collection(locale=u'locale')
#         acc.source = coll
#         self.session.add(coll)
#         self.session.commit()
#         coll_id = coll.id
#         self.session.delete(acc)
#         self.session.commit()
#         self.assert_(not self.session.query(Plant).get(coll_id))
        # test that the collection is orphaned after being deleted

    def test_constraints(self):
        """
        Test the constraints on the accession table.
        """
        acc = Accession(species=self.species, code=u'1')
        self.session.add(acc)
        self.session.commit()

        # test that accession.code is unique
        acc = Accession(species=self.species, code=u'1')
        self.session.add(acc)
        self.assertRaises(IntegrityError, self.session.commit)


    def test_set_source(self):
        #acc = self.session.query(Accession).get(1)
        #donor = self.session.query(Donor).get(1)
        acc = Accession(code=u'1', species=self.species)
        donor = Donor(name=u'me')
        self.session.add_all([acc, donor])
        self.session.commit()

        # set source on accession as a Donation
        donation = Donation()
        donation.donor = donor
        acc.source = donation
        self.session.flush()
        #self.session.expire(acc)
        self.session.refresh(acc)
        self.assertEquals(acc.source.id, donation.id)
        self.assertEquals(acc.source_type, u'Donation')

        # create a new Donation and set that as the source, this should
        # delete the old donation object since it's an orphan,
        old_donation_id = donation.id
        donation2 = Donation()
        donation2.donor = donor
        acc.source = donation2
        self.session.flush()
        self.session.expire(acc)
        self.assertEquals(acc.source.id, donation2.id)
        self.assertEquals(acc.source_type, u'Donation')

        # set the same source twice to make sure the source isn't
        # deleted before setting it again
        acc.source = donation2
        self.session.commit()
        self.assert_(acc.source)

        # delete all the donations
        # TODO: the donor should never be deleted if a donation is
        # deleted and a donation should never get deleted if a donor
        # is deleted, an error should be reaised if you attempt to
        # delete a donor that has donations but should an error be
        # raised if you attempt to delete a donation that has a donor,
        # i don't think so

        # make sure the old donation gets deleted since it's an orphan
        print self.session.query(Donation).get(old_donation_id)
        self.assert_(self.session.query(Donation).get(old_donation_id) == None)

        # delete the source by setting acc.source=None
        donation = Donation()
        donation.donor = donor
        acc.source = donation
        acc.source = None
        self.session.flush()
        self.session.expire(acc)
        old_donation_id = donation2.id
        self.assertEquals(acc.source, None)
        self.assertEquals(acc.source_type, None)

        # delete the source 2
        donation = Donation()
        donation.donor = donor
        acc.source = donation
        del acc.source
        self.session.flush()
        self.session.expire(acc)
        old_donation_id = donation2.id
        self.assertEquals(acc.source, None)
        self.assertEquals(acc.source_type, None)

        # make sure the orphaned donation get's deleted
        self.assert_(not self.session.query(Donation).get(old_donation_id))

        # set accession.source to a Collection
        collection = Collection(locale=u'TestAccLocale')
        acc.source = collection
        self.session.flush()
        self.session.expire(acc)
        self.assertEquals(acc.source.id, collection.id)
        self.assertEquals(acc.source_type, u'Collection')

        # changed source from collection to donation
        old_collection_id = collection.id
        donation3 = Donation()
        donation3.donor = donor
        acc.source = donation3
        self.session.flush()
        self.session.expire(acc)
        self.assertEquals(acc.source.id, donation3.id)
        self.assertEquals(acc.source_type, u'Donation')

        # make sure the orphaned collection get's deleted
        self.assert_(not self.session.query(Collection).get(old_collection_id))

        # change source from donation to collection
        old_donation_id = donation3.id
        collection2 = Collection(locale=u'TestAccLocale2')
        acc.source = collection2
        self.session.flush()
        self.session.expire(acc)
        self.assertEquals(acc.source.id, collection2.id)
        self.assertEquals(acc.source_type, u'Collection')

        # change source without flushing
        donation4 = Donation()
        acc.source = donation4
        collection3 = Collection(locale=u'TestAccLocale3')
        acc.source = collection3
        self.session.flush()
#        utils.log.echo(False)

        # make sure the orphaned donation get's deleted
        self.assert_(not self.session.query(Donation).get(old_donation_id))


    def test_double_commit(self):
        """
        This tests a bug with SQLAlchemy that was tentatively fixed
        after SQ 0.4.4 was released in r4264.  There is a reference to
        this in the SA mailing list.

        The bug is here just to check if this ever gets fixed.
        """
        import bauble.utils.log as log
        sp = self.session.query(Species).get(1)
        acc = Accession()
        self.session.add(acc)
        acc.species = sp
        acc.code = u"3"
        # not donor_id, should raise an IntegrityError
        donation = Donation()
        acc.source = donation
        try:
            self.session.commit()
        except IntegrityError:
            self.session.rollback()
            # before SA 0.4.5 this would give and InvalidRequestError
            # about not being able to refresh Accession after a rollback
            try:
                self.session.commit()
            except InvalidRequestError, e:
                # we get here in SA pre-0.4.5, we can't use those
                # versions for bauble
                raise
            except IntegrityError:
                # it should raise an integrity error because there is
                # still no donor_id on donation
                pass


    def itest_accession_editor(self):
        """
        Interactively test the PlantEditor
        """
        donor = self.create(Donor, name=u'test')
        self.session.commit()
        acc = self.create(Accession, species=self.species, code=u'1')
        acc.id_qual = 'aff.'
        editor = AccessionEditor(model=acc)
        #editor = AccessionEditor()
        editor.start()
        del editor
        assert utils.gc_objects_by_type('AccessionEditor') == [], \
            'AccessionEditor not deleted'
        assert utils.gc_objects_by_type('AccessionEditorPresenter') == [], \
            'AccessionEditorPresenter not deleted'
        assert utils.gc_objects_by_type('AccessionEditorView') == [], \
            'AccessionEditorView not deleted'


class LocationTests(GardenTestCase):

    def __init__(self, *args):
        super(LocationTests, self).__init__(*args)

    def setUp(self):
        super(LocationTests, self).setUp()


    def tearDown(self):
        super(LocationTests, self).tearDown()

    def itest_location_editor(self):
        """
        Interactively test the PlantEditor
        """
        loc = self.create(Location, site=u'some site')
        editor = LocationEditor(model=loc)
        editor.start()
        del editor
        assert utils.gc_objects_by_type('LocationEditor') == [], \
            'LocationEditor not deleted'
        assert utils.gc_objects_by_type('LocationEditorPresenter') == [], \
            'LocationEditorPresenter not deleted'
        assert utils.gc_objects_by_type('LocationEditorView') == [], \
            'LocationEditorView not deleted'

# latitude: deg[0-90], min[0-59], sec[0-59]
# longitude: deg[0-180], min[0-59], sec[0-59]

ALLOWED_DECIMAL_ERROR = 5
THRESHOLD = .01
DMS = 0 # DMS
DEG_MIN_DEC = 1 # Deg with minutes decimal
DEG_DEC = 2 # Degrees decimal
UTM = 3 # Datum(wgs84/nad83 or nad27), UTM Zone, Easting, Northing

# decimal points to accuracy in decimal degrees
# 1 +/- 8000m
# 2 +/- 800m
# 3 +/- 80m
# 4 +/- 8m
# 5 +/- 0.8m
# 6 +/- 0.08m

from decimal import Decimal
dec = Decimal
conversion_test_data = (
                        ((('N', 17, 21, dec(59)), # dms
                          ('W', 89, 1, 41)),
                         ((dec(17), dec('21.98333333')), # deg min_dec
                          (dec(-89), dec('1.68333333'))),
                         (dec('17.366389'), dec('-89.028056')), # dec deg
                         (('wgs84', 16, 284513, 1921226))), # utm
                        ((('S', 50, 19, dec('32.59')), # dms
                          ('W', 74, 2, dec('11.6'))),
                         ((dec(-50), dec('19.543166')), # deg min_dec
                          (dec(-74), dec('2.193333'))),
                         (dec('-50.325719'), dec('-74.036556')), # dec deg
                          (('wgs84', 18, 568579, 568579)),
                          (('nad27', 18, 568581, 4424928))),
                        ((('N', 9, 0, dec('4.593384')),
                          ('W', 78, 3, dec('28.527984'))),
                         ((9, dec('0.0765564')),
                          (-78, dec('3.4754664'))),
                         (dec('9.00127594'), dec('-78.05792444')))
                        )

#parse_lat_lon_data = ('17, 21, 59', '17 21 59', '17:21:59',
#                      '17, 21.98333333', '17 21.98333333',
#                      '17.36638889',
#                      '50, 19, 32.59', '50 19 32.59', '50:19:32.59',
#                      '-50 19.543166', '-50, 19.543166',
#                      '-50.325719')
parse_lat_lon_data = ('17 21 59', '17 21.98333333', '17.03656',
                      '50 19 32.59', '-50 19.543166', '-50.32572')


class DMSConversionTests(unittest.TestCase):

    # test coordinate conversions
    def test_dms_to_decimal(self):
        # test converting DMS to degrees decimal
        ALLOWED_ERROR = 6
        for data_set in conversion_test_data:
            dms_data = data_set[DMS]
            dec_data = data_set[DEG_DEC]
            lat_dec = dms_to_decimal(*dms_data[0])
            lon_dec = dms_to_decimal(*dms_data[1])
            self.assertAlmostEqual(lat_dec, dec_data[0], ALLOWED_ERROR)
            self.assertAlmostEqual(lon_dec, dec_data[1], ALLOWED_ERROR)



    def test_decimal_to_dms(self):
        # test converting degrees decimal to dms, allow a certain
        # amount of error in the seconds
        ALLOWABLE_ERROR = 2
        for data_set in conversion_test_data:
            dms_data = data_set[DMS]
            dec_data = data_set[DEG_DEC]

            # convert to DMS
            lat_dms = latitude_to_dms(dec_data[0])
            self.assertEqual(lat_dms[0:2], dms_data[0][0:2])
            # test seconds with allowable error
            self.assertAlmostEqual(lat_dms[3], dms_data[0][3], ALLOWABLE_ERROR)

            lon_dms = longitude_to_dms(dec_data[1])
            self.assertEqual(lon_dms[0:2], dms_data[1][0:2])
            # test seconds with allowable error
            self.assertAlmostEqual(lon_dms[3], dms_data[1][3], ALLOWABLE_ERROR)


    def test_parse_lat_lon(self):
        for data in parse_lat_lon_data:
            pass