htsworkflow/frontend/inventory/tests.py

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