htsworkflow/frontend/inventory/test_inventory.py

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