collections: tests and data migration

ccdb/collections_ccdb/factories.py

@@ -0,0 +1,88 @@
+from datetime import date, time
+
+from factory import DjangoModelFactory, Sequence, SubFactory, LazyFunction
+from factory.fuzzy import FuzzyText, FuzzyDate, FuzzyInteger
+from factory.django import FileField
+
+from .models import CollectionType, CollectionMethod, Flaw, ADFGPermit, Collection, \
+    DatasheetAttachment, CollectionTrap
+from ..projects.factories import ProjectFactory
+from ..locations.factories import StudyLocationFactory, StorageLocationFactory
+from ..processing.factories import ProcessTypeFactory, ReagentFactory
+
+
+class CollectionTypeFactory(DjangoModelFactory):
+    class Meta:
+        model = CollectionType
+
+    name = Sequence(lambda n: 'collection_type{}'.format(n))
+    code = Sequence(lambda n: 'ct{}'.format(n))
+    sort_order = Sequence(lambda n: n)
+
+
+class CollectionMethodFactory(DjangoModelFactory):
+    class Meta:
+        model = CollectionMethod
+
+    name = Sequence(lambda n: 'collection_method{}'.format(n))
+    code = Sequence(lambda n: 'cm{}'.format(n))
+    collection_method_class = FuzzyText(length=50)
+    sort_order = Sequence(lambda n: n)
+
+
+class FlawFactory(DjangoModelFactory):
+    class Meta:
+        model = Flaw
+
+    name = Sequence(lambda n: 'flaw{}'.format(n))
+    description = FuzzyText(length=255)
+    sort_order = Sequence(lambda n: n)
+
+
+class ADFGPermitFactory(DjangoModelFactory):
+    class Meta:
+        model = ADFGPermit
+
+    name = Sequence(lambda n: 'adfg_permit{}'.format(n))
+    sort_order = Sequence(lambda n: n)
+
+
+class CollectionFactory(DjangoModelFactory):
+    class Meta:
+        model = Collection
+
+    project = SubFactory(ProjectFactory)
+    study_location = SubFactory(StudyLocationFactory)
+    collection_type = SubFactory(CollectionTypeFactory)
+    collection_method = SubFactory(CollectionMethodFactory)
+    number_of_traps = FuzzyInteger(0)
+    collection_start_date = FuzzyDate(date(2012, 1, 1))
+    collection_start_time = None
+    collection_end_date = FuzzyDate(date(2015, 1, 1))
+    collection_end_time = None
+    storage_location = SubFactory(StorageLocationFactory)
+    specimen_state = FuzzyText(length=50)
+    process_type = SubFactory(ProcessTypeFactory)
+    reagent = SubFactory(ReagentFactory)
+    adfg_permit = SubFactory(ADFGPermitFactory)
+    flaw = SubFactory(FlawFactory)
+
+
+class DatasheetAttachmentFactory(DjangoModelFactory):
+    class Meta:
+        model = DatasheetAttachment
+
+    collection = SubFactory(CollectionFactory)
+    datasheet = FileField()
+
+
+class CollectionTrapFactory(DjangoModelFactory):
+    class Meta:
+        model = CollectionTrap
+
+    collection = SubFactory(CollectionFactory)
+    number_of_traps = FuzzyInteger(0)
+    date_opened = FuzzyDate(date(2012, 1, 1))
+    time_opened = LazyFunction(time)
+    date_closed = FuzzyDate(date(2015, 1, 1))
+    time_closed = LazyFunction(time)

ccdb/collections_ccdb/migrations/0005_DATA_initial.py

@@ -0,0 +1,107 @@
+from django.db import migrations
+from django.forms import modelform_factory
+
+from ccdb.utils.data import get_data_sources
+
+
+class Migration(migrations.Migration):
+    def migrate(apps, schema_editor):
+        sources = get_data_sources()
+        if not sources:
+            return
+
+        c = sources['db0']
+
+        CollectionType = apps.get_model('collections_ccdb', 'CollectionType')
+        CollectionMethod = apps.get_model('collections_ccdb', 'CollectionMethod')
+        Flaw = apps.get_model('collections_ccdb', 'Flaw')
+        ADFGPermit = apps.get_model('collections_ccdb', 'ADFGPermit')
+        Collection = apps.get_model('collections_ccdb', 'Collection')
+        DatasheetAttachment = apps.get_model('collections_ccdb', 'DatasheetAttachment')
+        CollectionTrap = apps.get_model('collections_ccdb', 'CollectionTrap')
+
+        for model in [CollectionTrap, Collection, Flaw, DatasheetAttachment,
+                      CollectionMethod, CollectionType, ADFGPermit]:
+            model.objects.all().delete()
+
+        Project = apps.get_model('projects', 'Project')
+
+        CollectionTypeForm = modelform_factory(CollectionType, fields='__all__')
+        CollectionMethodForm = modelform_factory(CollectionMethod, fields='__all__')
+        ADFGPermitForm = modelform_factory(ADFGPermit, fields='__all__')
+        CollectionForm = modelform_factory(Collection, fields='__all__')
+
+        for r in c.execute('SELECT * FROM tbl_lu_collection_types;'):
+            form = CollectionTypeForm(dict(name=r[1], code=r[2],
+                                           sort_order=int(r[3]) if r[3] else None))
+            if form.is_valid():
+                CollectionType.objects.create(id=r[0], **form.cleaned_data)
+            else:
+                print('collection type', r[0:], form.errors.as_data())
+
+        for r in c.execute('SELECT * FROM tbl_lu_collection_methods;'):
+            form = CollectionMethodForm(dict(name=r[1], code=r[2],
+                                             collection_method_class=r[3],
+                                             sort_order=int(r[4]) if r[4] else None))
+            if form.is_valid():
+                CollectionMethod.objects.create(id=r[0], **form.cleaned_data)
+            else:
+                print('collection method', r[0:], form.errors.as_data())
+
+        for i, r in enumerate(c.execute('SELECT DISTINCT ADFG_Permit FROM tbl_collections;')):
+            form = ADFGPermitForm(dict(name=r[0], sort_order=i))
+            if form.is_valid():
+                form.save()
+            else:
+                print('adfg permit', r[0:], form.errors.as_data())
+
+        for r in c.execute('''
+                SELECT *,
+                    collection_start_date AS "collection_start_date [dtdt]",
+                    collection_start_time AS "collection_start_time [dtdt]",
+                    collection_end_date   AS "collection_end_date [dtdt]",
+                    collection_end_time   AS "collection_end_time [dtdt]"
+                FROM tbl_collections;
+                '''):
+            permit = None
+            if r[14] is not '':
+                permit = ADFGPermit.objects.get(name=r[14]).pk
+            form = CollectionForm(dict(project=r[0], study_location=r[2],
+                                       collection_type=r[3], collection_method=r[4],
+                                       number_of_traps=r[5],
+                                       collection_start_date=r[17],
+                                       collection_start_time=r[18].time() if r[18] else None,
+                                       collection_end_date=r[19],
+                                       collection_end_time=r[20].time() if r[20] else None,
+                                       storage_location=r[10], specimen_state=r[11],
+                                       process_type=r[12], reagent=r[13],
+                                       adfg_permit=permit))
+            if form.is_valid():
+                project = Project.objects.get(id=r[0])
+                d = "{}_{}_{}_{}".format(project, form.cleaned_data['collection_end_date'],
+                                         form.cleaned_data['study_location'],
+                                         form.cleaned_data['collection_type'])
+                Collection.objects.create(id=r[1], display_name=d, **form.cleaned_data)
+            else:
+                print('collection', r[0:], form.errors.as_data())
+
+    def rollback(apps, schema_editor):
+        CollectionType = apps.get_model('collections_ccdb', 'CollectionType')
+        CollectionMethod = apps.get_model('collections_ccdb', 'CollectionMethod')
+        Flaw = apps.get_model('collections_ccdb', 'Flaw')
+        ADFGPermit = apps.get_model('collections_ccdb', 'ADFGPermit')
+        Collection = apps.get_model('collections_ccdb', 'Collection')
+        DatasheetAttachment = apps.get_model('collections_ccdb', 'DatasheetAttachment')
+        CollectionTrap = apps.get_model('collections_ccdb', 'CollectionTrap')
+
+        for model in [CollectionTrap, Collection, Flaw, DatasheetAttachment,
+                      CollectionMethod, CollectionType, ADFGPermit]:
+            model.objects.all().delete()
+
+    dependencies = [
+        ('collections_ccdb', '0004_collections_ordering'),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate, rollback),
+    ]

ccdb/collections_ccdb/models.py

@@ -78,7 +78,7 @@ class Collection(models.Model):
 
     def save(self, *args, **kwargs):
         self.display_name = "{}_{}_{}_{}".format(self.project,
-            self.collection_end_date.date(), self.study_location,
+            self.collection_end_date, self.study_location,
             self.collection_type)
         super(Collection, self).save(*args, **kwargs)
 
@@ -110,4 +110,5 @@ class CollectionTrap(models.Model):
             self.collection, self.number_of_traps, self.date_opened, self.date_closed)
 
     class Meta:
-        unique_together = ('collection', 'date_opened', 'time_opened', 'date_closed', 'time_closed')
+        unique_together = ('collection', 'date_opened', 'time_opened',
+                           'date_closed', 'time_closed')

ccdb/collections_ccdb/tests/test_models.py

@@ -0,0 +1,92 @@
+from django.test import TestCase
+from django.db import IntegrityError, transaction
+
+from ..models import CollectionType, CollectionMethod, Flaw, ADFGPermit, \
+    Collection, DatasheetAttachment, CollectionTrap
+from ..factories import CollectionTypeFactory, CollectionMethodFactory, \
+    FlawFactory, ADFGPermitFactory, CollectionFactory, DatasheetAttachmentFactory, \
+    CollectionTrapFactory
+
+
+class CollectionTypeTestCase(TestCase):
+    def test_creation(self):
+        c = CollectionTypeFactory()
+        self.assertTrue(isinstance(c, CollectionType))
+        self.assertEqual(c.__str__(), c.name)
+
+    def test_uniqueness(self):
+        c1 = CollectionTypeFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            CollectionTypeFactory(name=c1.name, code=c1.code)
+        c3 = CollectionTypeFactory()
+        self.assertTrue(isinstance(c3, CollectionType))
+
+
+class CollectionMethodTestCase(TestCase):
+    def test_creation(self):
+        c = CollectionMethodFactory()
+        self.assertTrue(isinstance(c, CollectionMethod))
+        self.assertEqual(c.__str__(), c.name)
+
+    def test_uniqueness(self):
+        c1 = CollectionMethodFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            CollectionMethodFactory(name=c1.name, code=c1.code)
+        c3 = CollectionMethodFactory()
+        self.assertTrue(isinstance(c3, CollectionMethod))
+
+
+class FlawTestCase(TestCase):
+    def test_creation(self):
+        f = FlawFactory()
+        self.assertTrue(isinstance(f, Flaw))
+        self.assertEqual(f.__str__(), f.name)
+
+
+class ADFGPermitTestCase(TestCase):
+    def test_creation(self):
+        a = ADFGPermitFactory()
+        self.assertTrue(isinstance(a, ADFGPermit))
+        self.assertEqual(a.__str__(), a.name)
+
+
+class CollectionTestCase(TestCase):
+    def test_creation(self):
+        c = CollectionFactory()
+        self.assertTrue(isinstance(c, Collection))
+        self.assertEqual(c.__str__(), c.display_name)
+
+    def test_uniqueness(self):
+        c1 = CollectionFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            CollectionFactory(project=c1.project, study_location=c1.study_location,
+                              collection_type=c1.collection_type,
+                              collection_start_date=c1.collection_start_date,
+                              collection_end_date=c1.collection_end_date,
+                              collection_method=c1.collection_method)
+        c3 = CollectionFactory()
+        self.assertTrue(isinstance(c3, Collection))
+
+
+class DatasheetAttachmentTestCase(TestCase):
+    def test_creation(self):
+        d = DatasheetAttachmentFactory()
+        self.assertTrue(isinstance(d, DatasheetAttachment))
+
+
+class CollectionTrapTestCase(TestCase):
+    def test_creation(self):
+        c = CollectionTrapFactory()
+        self.assertTrue(isinstance(c, CollectionTrap))
+        name = "{} # Traps: {} {} {}".format(c.collection, c.number_of_traps,
+                                             c.date_opened, c.date_closed)
+        self.assertEqual(c.__str__(), name)
+
+    def test_uniqueness(self):
+        c1 = CollectionTrapFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            CollectionTrapFactory(collection=c1.collection, date_opened=c1.date_opened,
+                                  time_opened=c1.time_opened, date_closed=c1.date_closed,
+                                  time_closed=c1.time_closed)
+        c3 = CollectionTrapFactory()
+        self.assertTrue(isinstance(c3, CollectionTrap))