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