encode_submission/test_encode_find.py

   1 #!/usr/bin/env python
   2 from datetime import datetime
   3 import os
   4 from unittest import TestCase
   5
   6 import RDF
   7
   8 import keyring.backend
   9
  10 class MockKeyring(keyring.backend.KeyringBackend):
  11     priority = 1
  12     def set_password(self, servicename, username, password):
  13         pass
  14
  15     def get_password(self, servicename, username):
  16         return "example"
  17
  18     def delete_password(self, servicename, username, password):
  19         pass
  20
  21 import keyring
  22 keyring.set_keyring(MockKeyring())
  23
  24 import encode_find
  25 from htsworkflow.submission.ucsc import submission_view_url
  26 from htsworkflow.util.rdfhelp import add_default_schemas, \
  27      dump_model, get_model, fromTypedNode
  28 from htsworkflow.util.rdfinfer import Infer
  29
  30 SOURCE_PATH = os.path.split(os.path.abspath(__file__))[0]
  31
  32 class TestEncodeFind(TestCase):
  33     def setUp(self):
  34         self.model = get_model()
  35         add_default_schemas(self.model)
  36         self.inference = Infer(self.model)
  37
  38     def test_create_status_node_with_uri(self):
  39         subURL = submission_view_url('5136')
  40         submissionUri = RDF.Uri(subURL)
  41         timestamp = '2011-12-19T12:42:53.048956'
  42         manualUri = subURL + '/' + timestamp
  43         nodeUri = encode_find.create_status_node(submissionUri, timestamp)
  44         self.assertEqual(str(nodeUri.uri), manualUri)
  45
  46     def test_create_status_node_with_str(self):
  47         subURL = submission_view_url('5136')
  48         timestamp = '2011-12-19T12:42:53.048956'
  49         manualUri = subURL + '/' + timestamp
  50         nodeUri = encode_find.create_status_node(subURL, timestamp)
  51         self.assertEqual(str(nodeUri.uri), manualUri)
  52
  53     def test_parse_submission_page(self):
  54         timestamp = '2011-12-19T12:42:53.048956'
  55         subURL = submission_view_url('5136')
  56         subNode = encode_find.create_status_node(subURL, timestamp)
  57         test_file = os.path.join(SOURCE_PATH, 'testdata', '5136SubDetail.html')
  58         from lxml.html import parse
  59         tree = parse(test_file)
  60         dates = encode_find.get_creation_dates(self.model, subNode)
  61         self.assertEqual(len(dates), 0)
  62         encode_find.parse_submission_page(self.model, tree, subNode)
  63         dates = encode_find.get_creation_dates(self.model, subNode)
  64         self.assertEqual(len(dates), 1)
  65         object_date = fromTypedNode(dates[0].object)
  66         self.assertEqual(object_date, datetime(2011,12,7,15,23,0))
  67
  68     def test_delete_simple_lane(self):
  69         model = get_model()
  70         parser = RDF.Parser(name='turtle')
  71         parser.parse_string_into_model(model, '''@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
  72 @prefix : <http://www.w3.org/1999/xhtml> .
  73 @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
  74 @prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
  75 @prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#> .
  76
  77 <http://jumpgate.caltech.edu/lane/1232>
  78     libns:flowcell <http://jumpgate.caltech.edu/flowcell/42JV5AAXX/> ;
  79     libns:total_unique_locations 5789938 .
  80
  81 ''', 'http://jumpgate.caltech.edu/library/')
  82         errmsgs = list(self.inference.run_validation())
  83         self.assertEqual(len(errmsgs), 0)
  84         urn = RDF.Node(RDF.Uri('http://jumpgate.caltech.edu/lane/1232'))
  85         encode_find.delete_lane(model, urn)
  86         self.failUnlessEqual(len(model), 0)
  87
  88     def test_delete_lane_with_mapping(self):
  89         ontology_size = len(self.model)
  90         parser = RDF.Parser(name='turtle')
  91         parser.parse_string_into_model(self.model, '''@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
  92 @prefix : <http://www.w3.org/1999/xhtml> .
  93 @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
  94 @prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
  95 @prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#> .
  96
  97 <http://jumpgate.caltech.edu/flowcell/42JV5AAXX/> a libns:IlluminaFlowcell .
  98 <http://jumpgate.caltech.edu/lane/1232>
  99     libns:flowcell <http://jumpgate.caltech.edu/flowcell/42JV5AAXX/> ;
 100     libns:has_mappings _:bnode110110 ;
 101     libns:total_unique_locations 5789938 ;
 102     a libns:IlluminaLane .
 103
 104 _:bnode110110
 105     a libns:MappedCount ;
 106     libns:mapped_to "newcontam_UK.fa"@en ;
 107     libns:reads 42473 .
 108 ''', 'http://jumpgate.caltech.edu/library/')
 109         errmsgs = list(self.inference.run_validation())
 110         self.assertEqual(len(errmsgs), 0)
 111         self.failUnlessEqual(len(self.model), 8 + ontology_size)
 112         urn = RDF.Node(RDF.Uri('http://jumpgate.caltech.edu/lane/1232'))
 113         encode_find.delete_lane(self.model, urn)
 114         self.failUnlessEqual(len(self.model), 1 + ontology_size)
 115         # the flowcell triple wasn't deleted.
 116
 117     def test_delete_library(self):
 118         parser = RDF.Parser(name='turtle')
 119         parser.parse_string_into_model(self.model, '''@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
 120 @prefix : <http://www.w3.org/1999/xhtml> .
 121 @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
 122 @prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
 123 @prefix libns: <http://jumpgate.caltech.edu/wiki/LibraryOntology#> .
 124
 125 <http://jumpgate.caltech.edu/flowcell/42JV5AAXX/> a libns:IlluminaFlowcell .
 126 <http://jumpgate.caltech.edu/flowcell/62WCKAAXX/> a libns:IlluminaFlowcell .
 127
 128 <http://jumpgate.caltech.edu/lane/1232>
 129     a libns:IlluminaLane ;
 130     libns:flowcell <http://jumpgate.caltech.edu/flowcell/42JV5AAXX/> ;
 131     libns:has_mappings _:bnode110110 ;
 132     libns:total_unique_locations 5789938 .
 133
 134 <http://jumpgate.caltech.edu/library/11011/>
 135     libns:affiliation "ENCODE"@en, "ENCODE_Tier1"@en, "Georgi Marinov"@en ;
 136     libns:has_lane <http://jumpgate.caltech.edu/lane/1232> ;
 137     libns:library_id "11011"@en ;
 138     libns:library_type "None"@en ;
 139     a libns:Library ;
 140     <http://www.w3.org/1999/xhtml/vocab#stylesheet> <http://jumpgate.caltech.edu/static/css/app.css>, <http://jumpgate.caltech.edu/static/css/data-browse-index.css> .
 141
 142 _:bnode110110
 143     a libns:MappedCount ;
 144     libns:mapped_to "newcontam_UK.fa"@en ;
 145     libns:reads 42473 .
 146
 147 <http://jumpgate.caltech.edu/lane/1903>
 148     a libns:IlluminaLane ;
 149     libns:flowcell <http://jumpgate.caltech.edu/flowcell/62WCKAAXX/> ;
 150     libns:has_mappings _:bnode120970 ;
 151     libns:total_unique_locations 39172114 .
 152
 153 <http://jumpgate.caltech.edu/library/12097/>
 154     libns:has_lane <http://jumpgate.caltech.edu/lane/1903> ;
 155     libns:library_id "12097"@en ;
 156     libns:library_type "Paired End (non-multiplexed)"@en ;
 157     a libns:Library .
 158
 159 _:bnode120970
 160     a libns:MappedCount ;
 161     libns:mapped_to "newcontam_UK.fa"@en ;
 162     libns:reads 64 .
 163 ''', 'http://jumpgate.caltech.edu/library')
 164         errmsgs = list(self.inference.run_validation())
 165         self.assertEqual(len(errmsgs), 0)
 166         urn = RDF.Node(RDF.Uri('http://jumpgate.caltech.edu/library/11011/'))
 167         encode_find.delete_library(self.model, urn)
 168         q = RDF.Statement(None, encode_find.libraryOntology['reads'], None)
 169         stmts = list(self.model.find_statements(q))
 170         self.failUnlessEqual(len(stmts), 1)
 171         self.failUnlessEqual(fromTypedNode(stmts[0].object),
 172                              64)
 173
 174         q = RDF.Statement(None, encode_find.libraryOntology['library_id'], None)
 175         stmts = list(self.model.find_statements(q))
 176         self.failUnlessEqual(len(stmts), 1)
 177         self.failUnlessEqual(fromTypedNode(stmts[0].object),
 178                              '12097')
 179
 180 def suite():
 181     from unittest import TestSuite, defaultTestLoader
 182     suite = TestSuite()
 183     suite.addTests(defaultTestLoader.loadTestsFromTestCase(TestEncodeFind))
 184     return suite
 185
 186 if __name__ == "__main__":
 187     from unittest import main
 188     main()