processing: tests and data migration

ccdb/processing/factories.py

@@ -0,0 +1,52 @@
+from datetime import datetime, date
+
+from factory import DjangoModelFactory, Sequence, SubFactory, LazyFunction
+from factory.fuzzy import FuzzyText, FuzzyDate, FuzzyFloat, FuzzyInteger
+
+from .models import ProcessType, Reagent, Flaw, Processing
+from ..misc.factories import ContainerFactory, MeasurementUnitFactory
+
+
+class ProcessTypeFactory(DjangoModelFactory):
+    class Meta:
+        model = ProcessType
+
+    name = Sequence(lambda n: 'process_type{}'.format(n))
+    code = Sequence(lambda n: 'pt{}'.format(n))
+    description = FuzzyText(length=255)
+    sort_order = Sequence(lambda n: n)
+
+
+class ReagentFactory(DjangoModelFactory):
+    class Meta:
+        model = Reagent
+
+    name = Sequence(lambda n: 'reagent{}'.format(n))
+    code = Sequence(lambda n: 'r{}'.format(n))
+    reagent_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 ProcessingFactory(DjangoModelFactory):
+    class Meta:
+        model = Processing
+
+    process_type = SubFactory(ProcessTypeFactory)
+    container = SubFactory(ContainerFactory)
+    container_label = FuzzyText(length=50)
+    process_date = FuzzyDate(date(2012, 1, 1))
+    process_time = LazyFunction(datetime.now().time)
+    reagent = SubFactory(ReagentFactory)
+    reagent_volume = FuzzyFloat(0.0)
+    measurement_unit = SubFactory(MeasurementUnitFactory)
+    minutes_in_reagent = FuzzyInteger(0)
+    flaw = SubFactory(FlawFactory)

ccdb/processing/migrations/0002_DATA_initial.py

@@ -0,0 +1,57 @@
+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']
+
+        ProcessType = apps.get_model('processing', 'ProcessType')
+        Reagent = apps.get_model('processing', 'Reagent')
+        Flaw = apps.get_model('processing', 'Flaw')
+        Processing = apps.get_model('processing', 'Processing')
+
+        for model in [Processing, Flaw, Reagent, ProcessType]:
+            model.objects.all().delete()
+
+        ProcessTypeForm = modelform_factory(ProcessType, fields='__all__')
+        ReagentForm = modelform_factory(Reagent, fields='__all__')
+
+        for r in c.execute('SELECT * FROM tbl_lu_process_types;'):
+            form = ProcessTypeForm(dict(name=r[1], code=r[2],
+                                    description=r[3],
+                                    sort_order=int(r[4]) if r[4] else None))
+            if form.is_valid():
+                ProcessType.objects.create(id=r[0], **form.cleaned_data)
+            else:
+                print('process type', r[0:], form.errors.as_data())
+
+        for r in c.execute('SELECT * FROM tbl_lu_reagents;'):
+            form = ReagentForm(dict(name=r[1], code=r[2],
+                                    reagent_class=r[3],
+                                    sort_order=int(r[4]) if r[4] else None))
+            if form.is_valid():
+                Reagent.objects.create(id=r[0], **form.cleaned_data)
+            else:
+                print('reagent', r[0:], form.errors.as_data())
+
+    def rollback(apps, schema_editor):
+        ProcessType = apps.get_model('processing', 'ProcessType')
+        Reagent = apps.get_model('processing', 'Reagent')
+
+        for model in [Reagent, ProcessType]:
+            model.objects.all().delete()
+
+    dependencies = [
+        ('processing', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.RunPython(migrate, rollback),
+    ]

ccdb/processing/models.py

@@ -59,8 +59,8 @@ class Processing(models.Model):
     flaw = models.ForeignKey(Flaw, blank=True, null=True)
 
     def __str__(self):
-        return "{process_date} {process_type} {container_label}".format(**self)
+        return "{} {} {}".format(self.process_date, self.process_type, self.container_label)
 
     class Meta:
         unique_together = ('process_type', 'container', 'container_label',
-            'process_date', 'process_time', 'reagent')
+                           'process_date', 'process_time', 'reagent')

ccdb/processing/tests/test_models.py

@@ -0,0 +1,58 @@
+from django.test import TestCase
+from django.db import IntegrityError, transaction
+
+from ..models import ProcessType, Reagent, Flaw, Processing
+from ..factories import ProcessTypeFactory, ReagentFactory, FlawFactory, ProcessingFactory
+
+
+class ProcessTypeTestCase(TestCase):
+    def test_creation(self):
+        p = ProcessTypeFactory()
+        self.assertTrue(isinstance(p, ProcessType))
+        self.assertEqual(p.__str__(), p.name)
+
+    def test_uniqueness(self):
+        p1 = ProcessTypeFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            ProcessTypeFactory(name=p1.name, code=p1.code)
+        p3 = ProcessTypeFactory()
+        self.assertTrue(isinstance(p3, ProcessType))
+
+
+class ReagentTestCase(TestCase):
+    def test_creation(self):
+        r = ReagentFactory()
+        self.assertTrue(isinstance(r, Reagent))
+        self.assertEqual(r.__str__(), r.name)
+
+    def test_uniqueness(self):
+        r1 = ReagentFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            ReagentFactory(name=r1.name, code=r1.code)
+        r3 = ReagentFactory()
+        self.assertTrue(isinstance(r3, Reagent))
+
+
+class FlawTestCase(TestCase):
+    def test_creation(self):
+        f = FlawFactory()
+        self.assertTrue(isinstance(f, Flaw))
+        self.assertEqual(f.__str__(), f.name)
+
+
+class ProcessingTestCase(TestCase):
+    def test_creation(self):
+        p = ProcessingFactory()
+        self.assertTrue(isinstance(p, Processing))
+        name = "{} {} {}".format(p.process_date, p.process_type, p.container_label)
+        self.assertEqual(p.__str__(), name)
+
+    def test_uniqueness(self):
+        p1 = ProcessingFactory()
+        with transaction.atomic(), self.assertRaises(IntegrityError):
+            ProcessingFactory(process_type=p1.process_type, container=p1.container,
+                              container_label=p1.container_label,
+                              process_date=p1.process_date,
+                              process_time=p1.process_time, reagent=p1.reagent)
+        p3 = ProcessingFactory()
+        self.assertTrue(isinstance(p3, Processing))