htsworkflow/frontend/inventory/test_inventory.py

   1 import RDF
   2
   3 from django.test import TestCase
   4 from django.contrib.auth.models import User
   5 from django.core import urlresolvers
   6
   7 from htsworkflow.frontend.inventory.models import Item, Vendor
   8 from htsworkflow.util.rdfhelp import get_model, load_string_into_model, get_serializer, inventoryOntology, libraryOntology, fromTypedNode
   9
  10 def localhostNode(url):
  11     return RDF.Node(RDF.Uri('http://localhost%s' % (url,)))
  12
  13 class InventoryTestCase(TestCase):
  14     fixtures = ['test_user', 'test_harddisks']
  15
  16     def test_fixture(self):
  17         # make sure that some of our test data is was loaded
  18         # since there was no error message when I typoed the test fixture
  19         hd1 = Item.objects.get(pk=1)
  20         self.failUnlessEqual(hd1.uuid, '8a90b6ce522311de99b00015172ce556')
  21
  22         user = User.objects.get(pk=5)
  23         self.failUnlessEqual(user.username, 'test')
  24
  25     def test_item(self):
  26         url = '/inventory/8a90b6ce522311de99b00015172ce556/'
  27         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
  28         response = self.client.get(url)
  29         self.failUnlessEqual(response.status_code, 200)
  30
  31         model = get_model()
  32         load_string_into_model(model, 'rdfa', response.content, url)
  33
  34         itemNode = RDF.Node(RDF.Uri(url))
  35         item_type = fromTypedNode(model.get_target(itemNode, inventoryOntology['item_type']))
  36         self.failUnlessEqual(item_type, u'Hard Drive')
  37
  38     def test_itemindex(self):
  39         url = '/inventory/it/Hard Drive/'
  40         indexNode = localhostNode(url)
  41         diskNode = localhostNode('/inventory/8a90b6ce522311de99b00015172ce556/')
  42         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
  43
  44         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  45         self.failUnlessEqual(len(flowcells), 2)
  46         self.failUnless('http://localhost/flowcell/11ONEAAXX/' in flowcells)
  47         self.failUnless('http://localhost/flowcell/22TWOAAXX/' in flowcells)
  48
  49     def test_add_disk(self):
  50         url = '/inventory/it/Hard Drive/'
  51         #url_disk = '/inventory/8a90b6ce522311de99b00015172ce556/'
  52         url_disk = '/inventory/b0792d425aa411de99b00015172ce556/'
  53         indexNode = localhostNode(url)
  54         diskNode = localhostNode(url_disk)
  55         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
  56
  57         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  58         self.failUnlessEqual(len(flowcells), 0)
  59
  60         # step two link the flowcell
  61         flowcell = '22TWOAAXX'
  62         serial = 'WCAU49042470'
  63         link_url = urlresolvers.reverse(
  64                 'htsworkflow.frontend.inventory.views.link_flowcell_and_device',
  65                 args=(flowcell, serial))
  66         link_response = self.client.get(link_url)
  67         self.failUnlessEqual(link_response.status_code, 200)
  68
  69         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  70         self.failUnlessEqual(len(flowcells), 1)
  71         self.failUnlessEqual('http://localhost/flowcell/%s/' % (flowcell),
  72                              flowcells[0])
  73
  74     def test_add_disk_failed_flowcell(self):
  75         url = '/inventory/it/Hard Drive/'
  76         #url_disk = '/inventory/8a90b6ce522311de99b00015172ce556/'
  77         url_disk = '/inventory/b0792d425aa411de99b00015172ce556/'
  78         indexNode = localhostNode(url)
  79         diskNode = localhostNode(url_disk)
  80         self.client.login(username='test', password='BJOKL5kAj6aFZ6A5')
  81
  82         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  83         self.failUnlessEqual(len(flowcells), 0)
  84
  85         # step two link the flowcell
  86         flowcell = '33THRAAXX'
  87         serial = 'WCAU49042470'
  88         link_url = urlresolvers.reverse(
  89                 'htsworkflow.frontend.inventory.views.link_flowcell_and_device',
  90                 args=(flowcell, serial))
  91         link_response = self.client.get(link_url)
  92         self.failUnlessEqual(link_response.status_code, 200)
  93
  94         flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
  95         self.failUnlessEqual(len(flowcells), 1)
  96         self.failUnlessEqual('http://localhost/flowcell/%s/' % (flowcell),
  97                              flowcells[0])
  98
  99
 100     def get_flowcells_from_content(self, url, rootNode, diskNode):
 101         model = get_model()
 102
 103         response = self.client.get(url)
 104         self.failUnlessEqual(response.status_code, 200)
 105
 106         load_string_into_model(model, 'rdfa', response.content, rootNode.uri)
 107         targets = model.get_targets(diskNode, libraryOntology['flowcell_id'])
 108         flowcells = [ str(x.uri) for x in targets]
 109         return flowcells
 110
 111 def suite():
 112     from unittest2 import TestSuite, defaultTestLoader
 113     suite = TestSuite()
 114     suite.addTests(defaultTestLoader.loadTestsFromTestCase(InventoryTestCase))
 115     return suite
 116
 117 if __name__ == "__main__":
 118     from unittest2 import main
 119     main(defaultTest="suite")