samples/test_samples.py

   1 from __future__ import absolute_import, print_function
   2
   3 import datetime
   4 import json
   5 from unittest import skipUnless
   6
   7 from django.core.exceptions import ValidationError
   8 from django.test import TestCase, RequestFactory
   9 from django.utils.encoding import smart_text, smart_str, smart_bytes
  10
  11 from rdflib import ConjunctiveGraph, Graph
  12
  13 from .models import Affiliation, ExperimentType, Species, Library
  14 from .views import library_dict
  15 from .samples_factory import (
  16     AffiliationFactory,
  17     LibraryAccessionFactory,
  18     LibraryFactory,
  19     SpeciesFactory,
  20 )
  21 from htsworkflow.auth import apidata
  22 from htsworkflow.util.conversion import str_or_none
  23 from htsworkflow.util.ethelp import validate_xhtml
  24
  25 from htsworkflow.util.rdfhelp import add_default_schemas
  26 from htsworkflow.util.rdfinfer import Infer
  27 from htsworkflow.util.rdfns import libraryOntology
  28
  29 class LibraryAccessionTestCase(TestCase):
  30     def test_validator(self):
  31         library = LibraryFactory()
  32         acc = LibraryAccessionFactory(library_id=library.id)
  33         acc.clean_fields()
  34         accession = acc.accession
  35         # test a variety of escape characters one at a time
  36         for c in "<>'\"&;":
  37             acc.accession = accession + c
  38             self.assertRaises(ValidationError, acc.clean_fields)
  39
  40     def test_library_save_hook(self):
  41         library = LibraryFactory()
  42         acc = LibraryAccessionFactory(library_id=library.id)
  43
  44         self.assertEquals(acc.url[:len(acc.agency.homepage)],
  45                           acc.agency.homepage)
  46         self.assertEquals(acc.url[len(acc.agency.homepage):],
  47                           '/library/'+acc.accession)
  48
  49     def test_have_accession(self):
  50         library = LibraryFactory()
  51         acc = LibraryAccessionFactory(library_id=library.id)
  52         lib_response = self.client.get(library.get_absolute_url())
  53         lib_content = smart_text(lib_response.content)
  54
  55         model = Graph()
  56         model.parse(data=lib_content, format='rdfa')
  57
  58         body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
  59         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
  60
  61         select ?library ?accession
  62         where {
  63            ?library libns:accession ?accession
  64         }"""
  65         accessions = []
  66         for row in model.query(body):
  67             accessions.append(str(row['accession']))
  68         self.assertEqual(len(accessions), 1)
  69         self.assertEqual(accessions[0], acc.url)
  70
  71
  72 class LibraryTestCase(TestCase):
  73     def testOrganism(self):
  74         human = SpeciesFactory(common_name='human')
  75         self.assertEquals(human.common_name, 'human')
  76         library = LibraryFactory(library_species=human)
  77         self.assertEquals(library.organism(), 'human')
  78
  79     def testAddingOneAffiliation(self):
  80         affiliation = AffiliationFactory.create(name='Alice')
  81         library = LibraryFactory()
  82         library.affiliations.add(affiliation)
  83
  84         self.assertEqual(len(library.affiliations.all()), 1)
  85         self.assertEqual(library.affiliation(), 'Alice (contact name)')
  86
  87     def testMultipleAffiliations(self):
  88         alice = AffiliationFactory.create(name='Alice')
  89         bob = AffiliationFactory.create(name='Bob')
  90
  91         library = LibraryFactory()
  92         library.affiliations.add(alice, bob)
  93
  94         self.assertEqual(len(library.affiliations.all()), 2)
  95         self.assertEqual(library.affiliation(),
  96                          'Alice (contact name), Bob (contact name)')
  97
  98
  99 class SampleWebTestCase(TestCase):
 100     """
 101     Test returning data from our database in rest like ways.
 102     (like returning json objects)
 103     """
 104     def test_library_dict(self):
 105         library = LibraryFactory.create()
 106         lib_dict = library_dict(library.id)
 107         url = '/samples/library/%s/json' % (library.id,)
 108         lib_response = self.client.get(url, apidata)
 109         lib_json = json.loads(smart_text(lib_response.content))['result']
 110
 111         for d in [lib_dict, lib_json]:
 112             # amplified_from_sample is a link to the library table,
 113             # I want to use the "id" for the data lookups not
 114             # the embedded primary key.
 115             # It gets slightly confusing on how to implement sending the right id
 116             # since amplified_from_sample can be null
 117             #self.failUnlessEqual(d['amplified_from_sample'], lib.amplified_from_sample)
 118             self.failUnlessEqual(d['antibody_id'], library.antibody_id)
 119             self.failUnlessEqual(d['cell_line_id'], library.cell_line_id)
 120             self.failUnlessEqual(d['cell_line'], str_or_none(library.cell_line))
 121             self.failUnlessEqual(d['experiment_type'], library.experiment_type.name)
 122             self.failUnlessEqual(d['experiment_type_id'], library.experiment_type_id)
 123             self.failUnlessEqual(d['gel_cut_size'], library.gel_cut_size)
 124             self.failUnlessEqual(d['hidden'], library.hidden)
 125             self.failUnlessEqual(d['id'], library.id)
 126             self.failUnlessEqual(d['insert_size'], library.insert_size)
 127             self.failUnlessEqual(d['library_name'], library.library_name)
 128             self.failUnlessEqual(d['library_species'], library.library_species.scientific_name)
 129             self.failUnlessEqual(d['library_species_id'], library.library_species_id)
 130             self.failUnlessEqual(d['library_type_id'], library.library_type_id)
 131             self.assertTrue(d['library_type'].startswith('library type'))
 132             self.failUnlessEqual(d['made_for'], library.made_for)
 133             self.failUnlessEqual(d['made_by'], library.made_by)
 134             self.failUnlessEqual(d['notes'], library.notes)
 135             self.failUnlessEqual(d['replicate'], library.replicate)
 136             self.failUnlessEqual(d['stopping_point'], library.stopping_point)
 137             self.failUnlessEqual(d['successful_pM'], library.successful_pM)
 138             self.failUnlessEqual(d['undiluted_concentration'],
 139                                  str(library.undiluted_concentration))
 140
 141
 142         def junk(self):
 143                 # some specific tests
 144                 if library.id == '10981':
 145                     # test a case where there is no known status
 146                     lane_set = {u'status': u'Unknown',
 147                                 u'paired_end': True,
 148                                 u'read_length': 75,
 149                                 u'lane_number': 1,
 150                                 u'lane_id': 1193,
 151                                 u'flowcell': u'303TUAAXX',
 152                                 u'status_code': None}
 153                     self.failUnlessEqual(len(d['lane_set']), 1)
 154                     self.failUnlessEqual(d['lane_set'][0], lane_set)
 155                 elif library.id == '11016':
 156                     # test a case where there is a status
 157                     lane_set = {u'status': 'Good',
 158                                 u'paired_end': True,
 159                                 u'read_length': 75,
 160                                 u'lane_number': 5,
 161                                 u'lane_id': 1197,
 162                                 u'flowcell': u'303TUAAXX',
 163                                 u'status_code': 2}
 164                     self.failUnlessEqual(len(d['lane_set']), 1)
 165                     self.failUnlessEqual(d['lane_set'][0], lane_set)
 166
 167
 168     def test_invalid_library_json(self):
 169         """
 170         Make sure we get a 404 if we request an invalid library id
 171         """
 172         response = self.client.get('/samples/library/nottheone/json', apidata)
 173         self.failUnlessEqual(response.status_code, 404)
 174
 175
 176     def test_invalid_library(self):
 177         response = self.client.get('/library/nottheone/')
 178         self.failUnlessEqual(response.status_code, 404)
 179
 180
 181     def test_library_no_key(self):
 182         """
 183         Make sure we get a 403 if we're not logged in
 184         """
 185         library = LibraryFactory.create()
 186
 187         url = '/samples/library/{}/json'.format(library.id)
 188         response = self.client.get(url, apidata)
 189         self.failUnlessEqual(response.status_code, 200)
 190         response = self.client.get(url)
 191         self.failUnlessEqual(response.status_code, 403)
 192
 193     def test_library_rdf(self):
 194         library = LibraryFactory.create()
 195
 196         model = ConjunctiveGraph()
 197
 198         response = self.client.get(library.get_absolute_url())
 199         self.assertEqual(response.status_code, 200)
 200         model.parse(data=smart_text(response.content), format='rdfa')
 201
 202         body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
 203         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
 204
 205         select ?library ?name ?library_id ?gel_cut ?made_by
 206         where {
 207            ?library a libns:library ;
 208                     libns:name ?name ;
 209                     libns:library_id ?library_id ;
 210                     libns:gel_cut ?gel_cut ;
 211                     libns:made_by ?made_by
 212         }"""
 213         for r in model.query(body):
 214             self.assertEqual(fromTypedNode(r['library_id']),
 215                              library.id)
 216             self.assertEqual(fromTypedNode(r['name']),
 217                              library.name)
 218             self.assertEqual(fromTypedNode(r['gel_cut']),
 219                              library.gel_cut)
 220             self.assertEqual(fromTypedNode(r['made_by']),
 221                              library.made_by)
 222
 223         state = validate_xhtml(smart_bytes(response.content))
 224         if state is not None:
 225             self.assertTrue(state)
 226
 227         # validate a library page.
 228         add_default_schemas(model)
 229         inference = Infer(model)
 230         errmsgs = list(inference.run_validation())
 231         self.assertEqual(len(errmsgs), 0)
 232
 233     def test_library_index_rdfa(self):
 234         model = ConjunctiveGraph()
 235         add_default_schemas(model)
 236         inference = Infer(model)
 237
 238         response = self.client.get('/library/')
 239         self.assertEqual(response.status_code, 200)
 240         model.parse(data=smart_text(response.content), format='rdfa')
 241
 242         errmsgs = list(inference.run_validation())
 243         self.assertEqual(len(errmsgs), 0)
 244
 245         body =  """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
 246         prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
 247
 248         select ?library ?library_id ?name ?species ?species_name
 249         where {
 250            ?library a libns:Library .
 251            OPTIONAL { ?library libns:library_id ?library_id . }
 252            OPTIONAL { ?library libns:species ?species .
 253                       ?species libns:species_name ?species_name . }
 254            OPTIONAL { ?library libns:name ?name . }
 255         }"""
 256         bindings = set(['library', 'library_id', 'name', 'species', 'species_name'])
 257         count = 0
 258         for r in model.query(body):
 259             count += 1
 260             for name, value in r.items():
 261                 self.assertTrue(name in bindings)
 262                 self.assertTrue(value is not None)
 263
 264         self.assertEqual(count, len(Library.objects.filter(hidden=False)))
 265
 266         state = validate_xhtml(response.content)
 267         if state is not None:
 268             self.assertTrue(state)
 269
 270
 271 # The django test runner flushes the database between test suites not cases,
 272 # so to be more compatible with running via nose we flush the database tables
 273 # of interest before creating our sample data.
 274 def create_db(obj):
 275     obj.species_human = Species.objects.get(pk=8)
 276     obj.experiment_rna_seq = ExperimentType.objects.get(pk=4)
 277     obj.affiliation_alice = Affiliation.objects.get(pk=1)
 278     obj.affiliation_bob = Affiliation.objects.get(pk=2)
 279
 280     Library.objects.all().delete()
 281     obj.library_10001 = Library(
 282         id = "10001",
 283         library_name = 'C2C12 named poorly',
 284         library_species = obj.species_human,
 285         experiment_type = obj.experiment_rna_seq,
 286         creation_date = datetime.datetime.now(),
 287         made_for = 'scientist unit 2007',
 288         made_by = 'microfludics system 7321',
 289         stopping_point = '2A',
 290         undiluted_concentration = '5.01',
 291         hidden = False,
 292     )
 293     obj.library_10001.save()
 294     obj.library_10002 = Library(
 295         id = "10002",
 296         library_name = 'Worm named poorly',
 297         library_species = obj.species_human,
 298         experiment_type = obj.experiment_rna_seq,
 299         creation_date = datetime.datetime.now(),
 300         made_for = 'scientist unit 2007',
 301         made_by = 'microfludics system 7321',
 302         stopping_point = '2A',
 303         undiluted_concentration = '5.01',
 304         hidden = False,
 305     )
 306     obj.library_10002.save()
 307
 308 class TestRDFaLibrary(TestCase):
 309     def setUp(self):
 310         self.request = RequestFactory()
 311
 312     def test_parse_rdfa(self):
 313         model = Graph()
 314
 315         bob = AffiliationFactory.create(name='Bob')
 316
 317         lib_object = LibraryFactory()
 318         lib_object.affiliations.add(bob)
 319         url = '/library/{}/'.format(lib_object.id)
 320         ## request = self.request.get(url)
 321         ## lib_response = library(request)
 322         lib_response = self.client.get(url)
 323         lib_body = smart_str(lib_response.content)
 324         self.failIfEqual(len(lib_body), 0)
 325         with open('/tmp/body.html', 'wt') as outstream:
 326             outstream.write(lib_body)
 327
 328         model.parse(data=lib_body, format='rdfa', publicID='http://localhost'+url)
 329
 330         # help debugging rdf errrors
 331         #with open('/tmp/test.ttl', 'w') as outstream:
 332         #    dump_model(model, outstream)
 333         # http://jumpgate.caltech.edu/wiki/LibraryOntology#affiliation>
 334         self.check_literal_object(model, ['Bob'], p=libraryOntology['affiliation'])
 335         self.check_literal_object(model,
 336                                   ['experiment type name'],
 337                                   p=libraryOntology['experiment_type'])
 338         self.check_literal_object(model, [400], p=libraryOntology['gel_cut'])
 339         self.check_literal_object(model,
 340                                   ['microfluidics bot 7321'],
 341                                   p=libraryOntology['made_by'])
 342         self.check_literal_object(model,
 343                                   [lib_object.library_name],
 344                                   p=libraryOntology['name'])
 345         self.check_literal_object(model,
 346                                   [lib_object.library_species.scientific_name],
 347                                   p=libraryOntology['species_name'])
 348
 349
 350     def check_literal_object(self, model, values, s=None, p=None, o=None):
 351         statements = list(model.triples((s,p,o)))
 352         self.failUnlessEqual(len(statements), len(values),
 353                         "Couln't find %s %s %s" % (s,p,o))
 354         for stmt in statements:
 355             obj = stmt[2]
 356             self.assertIn(obj.value, values)
 357
 358     def check_uri_object(self, model, values, s=None, p=None, o=None):
 359         statements = list(model.triples((s,p,o)))
 360         self.failUnlessEqual(len(statements), len(values),
 361                         "Couln't find %s %s %s" % (s,p,o))
 362         for stmt in statements:
 363             subject = stmt[0]
 364             self.assertIn(str(subject.value), values)
 365
 366 def suite():
 367     from unittest import TestSuite, defaultTestLoader
 368     suite = TestSuite()
 369     suite.addTests(defaultTestLoader.loadTestsFromTestCase(LibraryAccessionTestCase))
 370     suite.addTests(defaultTestLoader.loadTestsFromTestCase(LibraryTestCase))
 371     suite.addTests(defaultTestLoader.loadTestsFromTestCase(SampleWebTestCase))
 372     suite.addTests(defaultTestLoader.loadTestsFromTestCase(TestRDFaLibrary))
 373     return suite
 374
 375 if __name__ == "__main__":
 376     from unittest import main
 377     main(defaultTest="suite")