inventory/test_inventory.py

   1 from __future__ import absolute_import, print_function
   2
   3 import RDF
   4
   5 from django.test import TestCase
   6 from django.test.utils import setup_test_environment, \
   7      teardown_test_environment
   8 from django.db import connection
   9 from django.conf import settings
  10
  11 from django.contrib.auth.models import User
  12 from django.core.urlresolvers import reverse
  13 from django.utils.encoding import smart_text
  14
  15 from .models import Item, Vendor
  16 from .inventory_factory import ItemFactory, LongTermStorageFactory
  17 from samples.samples_factory import HTSUserFactory, LibraryFactory
  18 from experiments.experiments_factory import FlowCellFactory
  19 from htsworkflow.util.rdfhelp import get_model, load_string_into_model, get_serializer, inventoryOntology, libraryOntology, fromTypedNode
  20
  21 def localhostNode(url):
  22     return RDF.Node(RDF.Uri('http://localhost%s' % (url,)))
  23
  24 class InventoryTestCase(TestCase):
  25     def setUp(self):
  26         self.password = 'foo'
  27         self.user = HTSUserFactory.create()
  28         self.user.set_password(self.password)
  29         self.user.save()
  30
  31     def test_item(self):
  32         item = ItemFactory()
  33         self.assertTrue(len(item.uuid), 32)
  34         url = '/inventory/{}/'.format(item.uuid)
  35         self.assertTrue(self.client.login(username=self.user.username,
  36                                           password=self.password))
  37         response = self.client.get(url)
  38         self.failUnlessEqual(response.status_code, 200)
  39         content = smart_text(response.content)
  40
  41         model = get_model()
  42         load_string_into_model(model, 'rdfa', content, url)
  43
  44         itemNode = RDF.Node(RDF.Uri(url))
  45         item_type = fromTypedNode(
  46             model.get_target(itemNode, inventoryOntology['item_type']))
  47         self.failUnlessEqual(item_type, item.item_type.name)
  48
  49     def test_itemindex(self):
  50         item = ItemFactory()
  51         fc1 = FlowCellFactory()
  52         lib1 = LibraryFactory()
  53         lts = LongTermStorageFactory(flowcell=fc1,
  54                                      libraries=[lib1,],
  55                                      storage_devices=[item,],)
  56
  57         url = reverse('inventory.views.itemtype_index',
  58                       kwargs={'name': item.item_type.name})
  59         disk_url = reverse('inventory.views.item_summary_by_uuid',
  60                            kwargs={'uuid': item.uuid})
  61         indexNode = localhostNode(url)
  62         diskNode = localhostNode(disk_url)
  63         self.assertTrue(self.client.login(username=self.user.username,
  64                                           password=self.password))
  65
  66         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  67         self.failUnlessEqual(len(flowcells), 1)
  68         flowcell_url = reverse('experiments.views.flowcell_detail',
  69                                kwargs={'flowcell_id': fc1.flowcell_id})
  70         self.assertTrue(flowcells[0].endswith(flowcell_url))
  71
  72
  73     def test_add_disk(self):
  74         item = ItemFactory()
  75         url = reverse('inventory.views.itemtype_index',
  76                       kwargs={'name': item.item_type.name})
  77         disk_url = reverse('inventory.views.item_summary_by_uuid',
  78                            kwargs={'uuid': item.uuid})
  79         indexNode = localhostNode(url)
  80         diskNode = localhostNode(disk_url)
  81
  82         self.assertTrue(self.client.login(username=self.user.username,
  83                                           password=self.password))
  84
  85         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  86         self.failUnlessEqual(len(flowcells), 0)
  87
  88         # step two link the flowcell
  89         flowcell = FlowCellFactory(flowcell_id='22TWOAAXX')
  90         link_url = reverse('inventory.views.link_flowcell_and_device',
  91                            args=(flowcell.flowcell_id,
  92                                  item.barcode_id))
  93         link_response = self.client.get(link_url)
  94         self.assertEqual(link_response.status_code, 200)
  95
  96         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  97         flowcell_url = reverse('experiments.views.flowcell_detail',
  98                                kwargs={'flowcell_id': flowcell.flowcell_id})
  99         self.assertEqual(len(flowcells), 1)
 100         self.assertTrue(flowcells[0].endswith(flowcell_url))
 101
 102     def test_add_disk_failed_flowcell(self):
 103         item = ItemFactory()
 104         url = reverse('inventory.views.itemtype_index',
 105                       kwargs={'name': item.item_type.name})
 106         disk_url = reverse('inventory.views.item_summary_by_uuid',
 107                            kwargs={'uuid': item.uuid})
 108         indexNode = localhostNode(url)
 109         diskNode = localhostNode(disk_url)
 110
 111         self.assertTrue(self.client.login(username=self.user.username,
 112                                           password=self.password))
 113
 114         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
 115         self.failUnlessEqual(len(flowcells), 0)
 116
 117         # step two link the flowcell
 118         flowcell_id = '33THRAAXX'
 119         flowcell = FlowCellFactory(flowcell_id=flowcell_id +' (failed)')
 120         link_url = reverse('inventory.views.link_flowcell_and_device',
 121                            args=(flowcell.flowcell_id, item.barcode_id))
 122         link_response = self.client.get(link_url)
 123         self.failUnlessEqual(link_response.status_code, 200)
 124
 125         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
 126         self.assertEqual(len(flowcells), 1)
 127         flowcell_url = reverse('experiments.views.flowcell_detail',
 128                                kwargs={'flowcell_id': flowcell_id})
 129         self.assertTrue(flowcells[0].endswith(flowcell_url))
 130
 131
 132     def get_flowcells_from_content(self, url, rootNode, diskNode):
 133         model = get_model()
 134
 135         response = self.client.get(url)
 136         self.failUnlessEqual(response.status_code, 200)
 137
 138         content = smart_text(response.content)
 139         load_string_into_model(model, 'rdfa', content, rootNode.uri)
 140         targets = model.get_targets(diskNode, libraryOntology['flowcell_id'])
 141         flowcells = [ str(x.uri) for x in targets]
 142         return flowcells
 143
 144 def suite():
 145     from unittest import TestSuite, defaultTestLoader
 146     suite = TestSuite()
 147     suite.addTests(defaultTestLoader.loadTestsFromTestCase(InventoryTestCase))
 148     return suite
 149
 150 if __name__ == "__main__":
 151     from unittest import main
 152     main(defaultTest="suite")