Fix RDF schema problems with lane_number and species.
[htsworkflow.git] / htsworkflow / frontend / experiments / tests.py
index 537e4b47c2ef871ea084a489ebf68eef6c2d43ef..0f241476e78ff7938145337bb64409accfa69a04 100644 (file)
@@ -8,6 +8,7 @@ import os
 import shutil
 import sys
 import tempfile
+from urlparse import urljoin
 
 from django.conf import settings
 from django.core import mail
@@ -16,6 +17,7 @@ from django.test import TestCase
 from htsworkflow.frontend.experiments import models
 from htsworkflow.frontend.experiments import experiments
 from htsworkflow.frontend.auth import apidata
+from htsworkflow.util.ethelp import validate_xhtml
 
 from htsworkflow.pipelines.test.simulate_runfolder import TESTDATA_DIR
 
@@ -255,6 +257,7 @@ class ExperimentsTestCases(TestCase):
                         u'11061',u'11062',u'11063',u'11064']
         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
         response = self.client.get('/admin/experiments/flowcell/153/')
+
         tree = fromstring(response.content)
         for i in range(0,8):
             xpath_expression = '//input[@id="id_lane_set-%d-library"]'
@@ -273,6 +276,10 @@ class ExperimentsTestCases(TestCase):
         """
         self.client.login(username='supertest', password='BJOKL5kAj6aFZ6A5')
         response = self.client.get('/library/11070/')
+        self.assertEqual(response.status_code, 200)
+        status = validate_xhtml(response.content)
+        if status is not None: self.assertTrue(status)
+
         tree = fromstring(response.content)
         flowcell_spans = tree.xpath('//span[@property="libns:flowcell_id"]',
                                     namespaces=NSMAP)
@@ -280,12 +287,18 @@ class ExperimentsTestCases(TestCase):
         failed_fc_span = flowcell_spans[0]
         failed_fc_a = failed_fc_span.getparent()
         # make sure some of our RDF made it.
-        self.assertEqual(failed_fc_a.get('rel'), 'libns:flowcell')
+        self.assertEqual(failed_fc_a.get('typeof'), 'libns:IlluminaFlowcell')
         self.assertEqual(failed_fc_a.get('href'), '/flowcell/30012AAXX/')
         fc_response = self.client.get(failed_fc_a.get('href'))
         self.assertEqual(fc_response.status_code, 200)
+        status = validate_xhtml(response.content)
+        if status is not None: self.assertTrue(status)
+
         fc_lane_response = self.client.get('/flowcell/30012AAXX/8/')
         self.assertEqual(fc_lane_response.status_code, 200)
+        status = validate_xhtml(response.content)
+        if status is not None: self.assertTrue(status)
+
 
     def test_pooled_multiplex_id(self):
         fc_dict = experiments.flowcell_information('42JU1AAXX')
@@ -411,6 +424,56 @@ class ExperimentsTestCases(TestCase):
 
             self.assertEqual(mimetype, response['content-type'])
 
+    def test_flowcell_rdf(self):
+        import RDF
+        from htsworkflow.util.rdfhelp import get_model, \
+             fromTypedNode, \
+             load_string_into_model, \
+             rdfNS, \
+             libraryOntology, \
+             dump_model
+
+        model = get_model()
+
+        expected = {'1': ['11034'],
+                    '2': ['11036'],
+                    '3': ['12044','11045'],
+                    '4': ['11047','13044'],
+                    '5': ['11055'],
+                    '6': ['11067'],
+                    '7': ['11069'],
+                    '8': ['11070']}
+        url = '/flowcell/42JU1AAXX/'
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        status = validate_xhtml(response.content)
+        if status is not None: self.assertTrue(status)
+
+        ns = urljoin('http://localhost', url)
+        load_string_into_model(model, 'rdfa', response.content, ns=ns)
+        body = """prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
+        prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
+
+        select ?flowcell ?flowcell_id ?lane_id ?library_id
+        where {
+          ?flowcell a libns:IlluminaFlowcell ;
+                    libns:flowcell_id ?flowcell_id ;
+                    libns:has_lane ?lane .
+          ?lane libns:lane_number ?lane_id ;
+                libns:library ?library .
+          ?library libns:library_id ?library_id .
+        }"""
+        query = RDF.SPARQLQuery(body)
+        count = 0
+        for r in query.execute(model):
+            count += 1
+            self.assertEqual(fromTypedNode(r['flowcell_id']), u'42JU1AAXX')
+            lane_id = fromTypedNode(r['lane_id'])
+            library_id = fromTypedNode(r['library_id'])
+            self.assertTrue(library_id in expected[lane_id])
+        self.assertEqual(count, 10)
+
+
 class TestFileType(TestCase):
     def test_file_type_unicode(self):
         file_type_objects = models.FileType.objects
@@ -498,8 +561,10 @@ class TestEmailNotify(TestCase):
         response = self.client.get('/experiments/started/153/', {'send':'1','bcc':'on'})
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(mail.outbox), 4)
+        bcc = set(settings.NOTIFICATION_BCC).intersect(set(settings.MANAGERS))
         for m in mail.outbox:
             self.assertTrue(len(m.body) > 0)
+            self.assertEqual(m.bcc, bcc)
 
     def test_email_navigation(self):
         """
@@ -535,24 +600,78 @@ class TestSequencer(TestCase):
                              "Illumina Genome Analyzer IIx")
         self.assertEqual(fc.sequencer.instrument_name,
                              "ILLUMINA-EC5D15")
+        # well actually we let the browser tack on the host name
+        url = fc.get_absolute_url()
+        self.assertEqual(url, '/flowcell/FC12150/')
 
     def test_rdf(self):
         response = self.client.get('/flowcell/FC12150/', apidata)
         tree = fromstring(response.content)
-        divs = tree.xpath('//div[@rel="libns:sequenced_by"]',
-                          namespaces=NSMAP)
-        self.assertEqual(len(divs), 1)
-        self.assertEqual(divs[0].attrib['rel'], 'libns:sequenced_by')
-        self.assertEqual(divs[0].attrib['resource'], '/sequencer/2')
-
-        name = divs[0].xpath('./span[@property="libns:sequencer_name"]')
+        seq_by = tree.xpath('//div[@rel="libns:sequenced_by"]',
+                            namespaces=NSMAP)
+        self.assertEqual(len(seq_by), 1)
+        self.assertEqual(seq_by[0].attrib['rel'], 'libns:sequenced_by')
+        seq = seq_by[0].getchildren()
+        self.assertEqual(len(seq), 1)
+        self.assertEqual(seq[0].attrib['about'], '/sequencer/2')
+        self.assertEqual(seq[0].attrib['typeof'], 'libns:Sequencer')
+
+        name = seq[0].xpath('./span[@property="libns:sequencer_name"]')
         self.assertEqual(len(name), 1)
         self.assertEqual(name[0].text, 'Tardigrade')
-        instrument = divs[0].xpath(
+        instrument = seq[0].xpath(
             './span[@property="libns:sequencer_instrument"]')
         self.assertEqual(len(instrument), 1)
         self.assertEqual(instrument[0].text, 'ILLUMINA-EC5D15')
-        model = divs[0].xpath(
+        model = seq[0].xpath(
             './span[@property="libns:sequencer_model"]')
         self.assertEqual(len(model), 1)
         self.assertEqual(model[0].text, 'Illumina Genome Analyzer IIx')
+
+    def test_flowcell_with_rdf_validation(self):
+        from htsworkflow.util.rdfhelp import add_default_schemas, \
+             dump_model, \
+             get_model, \
+             load_string_into_model
+        from htsworkflow.util.rdfinfer import Infer
+
+        model = get_model()
+        add_default_schemas(model)
+        inference = Infer(model)
+
+        url ='/flowcell/FC12150/'
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        status = validate_xhtml(response.content)
+        if status is not None: self.assertTrue(status)
+
+        load_string_into_model(model, 'rdfa', response.content)
+
+        errmsgs = list(inference.run_validation())
+        self.assertEqual(len(errmsgs), 2)
+        for errmsg in errmsgs:
+            self.assertEqual(errmsg, 'Missing type for: http://localhost/')
+
+    def test_lane_with_rdf_validation(self):
+        from htsworkflow.util.rdfhelp import add_default_schemas, \
+             dump_model, \
+             get_model, \
+             load_string_into_model
+        from htsworkflow.util.rdfinfer import Infer
+
+        model = get_model()
+        add_default_schemas(model)
+        inference = Infer(model)
+
+        url = '/lane/1193'
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        status = validate_xhtml(response.content)
+        if status is not None: self.assertTrue(status)
+
+        load_string_into_model(model, 'rdfa', response.content)
+
+        errmsgs = list(inference.run_validation())
+        self.assertEqual(len(errmsgs), 2)
+        for errmsg in errmsgs:
+            self.assertEqual(errmsg, 'Missing type for: http://localhost/')