1 from __future__ import absolute_import, print_function, unicode_literals
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
11 from django.contrib.auth.models import User
12 from django.core.urlresolvers import reverse
14 from .models import Item, Vendor
15 from .inventory_factory import ItemFactory, LongTermStorageFactory
16 from samples.samples_factory import HTSUserFactory, LibraryFactory
17 from experiments.experiments_factory import FlowCellFactory
18 from htsworkflow.util.rdfhelp import get_model, load_string_into_model, get_serializer, inventoryOntology, libraryOntology, fromTypedNode
20 def localhostNode(url):
21 return RDF.Node(RDF.Uri('http://localhost%s' % (url,)))
23 class InventoryTestCase(TestCase):
26 self.user = HTSUserFactory.create()
27 self.user.set_password(self.password)
32 self.assertTrue(len(item.uuid), 32)
33 url = '/inventory/{}/'.format(item.uuid)
34 self.assertTrue(self.client.login(username=self.user.username,
35 password=self.password))
36 response = self.client.get(url)
37 self.failUnlessEqual(response.status_code, 200)
40 load_string_into_model(model, 'rdfa', response.content, url)
42 itemNode = RDF.Node(RDF.Uri(url))
43 item_type = fromTypedNode(
44 model.get_target(itemNode, inventoryOntology[b'item_type']))
45 self.failUnlessEqual(item_type, item.item_type.name)
47 def test_itemindex(self):
49 fc1 = FlowCellFactory()
50 lib1 = LibraryFactory()
51 lts = LongTermStorageFactory(flowcell=fc1,
53 storage_devices=[item,],)
55 url = reverse('inventory.views.itemtype_index',
56 kwargs={'name': item.item_type.name})
57 disk_url = reverse('inventory.views.item_summary_by_uuid',
58 kwargs={'uuid': item.uuid})
59 indexNode = localhostNode(url)
60 diskNode = localhostNode(disk_url)
61 self.assertTrue(self.client.login(username=self.user.username,
62 password=self.password))
64 flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
65 self.failUnlessEqual(len(flowcells), 1)
66 flowcell_url = reverse('experiments.views.flowcell_detail',
67 kwargs={'flowcell_id': fc1.flowcell_id})
68 self.assertTrue(flowcells[0].endswith(flowcell_url))
71 def test_add_disk(self):
73 url = reverse('inventory.views.itemtype_index',
74 kwargs={'name': item.item_type.name})
75 disk_url = reverse('inventory.views.item_summary_by_uuid',
76 kwargs={'uuid': item.uuid})
77 indexNode = localhostNode(url)
78 diskNode = localhostNode(disk_url)
80 self.assertTrue(self.client.login(username=self.user.username,
81 password=self.password))
83 flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
84 self.failUnlessEqual(len(flowcells), 0)
86 # step two link the flowcell
87 flowcell = FlowCellFactory(flowcell_id='22TWOAAXX')
88 link_url = reverse('inventory.views.link_flowcell_and_device',
89 args=(flowcell.flowcell_id,
91 link_response = self.client.get(link_url)
92 self.assertEqual(link_response.status_code, 200)
94 flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
95 flowcell_url = reverse('experiments.views.flowcell_detail',
96 kwargs={'flowcell_id': flowcell.flowcell_id})
97 self.assertEqual(len(flowcells), 1)
98 self.assertTrue(flowcells[0].endswith(flowcell_url))
100 def test_add_disk_failed_flowcell(self):
102 url = reverse('inventory.views.itemtype_index',
103 kwargs={'name': item.item_type.name})
104 disk_url = reverse('inventory.views.item_summary_by_uuid',
105 kwargs={'uuid': item.uuid})
106 indexNode = localhostNode(url)
107 diskNode = localhostNode(disk_url)
109 self.assertTrue(self.client.login(username=self.user.username,
110 password=self.password))
112 flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
113 self.failUnlessEqual(len(flowcells), 0)
115 # step two link the flowcell
116 flowcell_id = '33THRAAXX'
117 flowcell = FlowCellFactory(flowcell_id=flowcell_id +' (failed)')
118 link_url = reverse('inventory.views.link_flowcell_and_device',
119 args=(flowcell.flowcell_id, item.barcode_id))
120 link_response = self.client.get(link_url)
121 self.failUnlessEqual(link_response.status_code, 200)
123 flowcells = self.get_flowcells_from_content(url, indexNode, diskNode)
124 self.assertEqual(len(flowcells), 1)
125 flowcell_url = reverse('experiments.views.flowcell_detail',
126 kwargs={'flowcell_id': flowcell_id})
127 self.assertTrue(flowcells[0].endswith(flowcell_url))
130 def get_flowcells_from_content(self, url, rootNode, diskNode):
133 response = self.client.get(url)
134 self.failUnlessEqual(response.status_code, 200)
136 load_string_into_model(model, 'rdfa', response.content, rootNode.uri)
137 targets = model.get_targets(diskNode, libraryOntology['flowcell_id'])
138 flowcells = [ str(x.uri) for x in targets]
142 from unittest import TestSuite, defaultTestLoader
144 suite.addTests(defaultTestLoader.loadTestsFromTestCase(InventoryTestCase))
147 if __name__ == "__main__":
148 from unittest import main
149 main(defaultTest="suite")