Package lxml :: Package tests :: Module test_threading
[frames] | no frames]

Source Code for Module lxml.tests.test_threading

  1  # -*- coding: utf-8 -*- 
  2   
  3  """ 
  4  Tests for thread usage in lxml.etree. 
  5  """ 
  6   
  7  import unittest, threading 
  8   
  9  from common_imports import etree, HelperTestCase, StringIO 
 10   
11 -class ThreadingTestCase(HelperTestCase):
12 """Threading tests""" 13 etree = etree 14
15 - def _run_thread(self, func):
16 thread = threading.Thread(target=func) 17 thread.start() 18 thread.join()
19
20 - def test_subtree_copy_thread(self):
21 tostring = self.etree.tostring 22 XML = self.etree.XML 23 xml = "<root><threadtag/></root>" 24 main_root = XML("<root/>") 25 26 def run_thread(): 27 thread_root = XML(xml) 28 main_root.append(thread_root[0]) 29 del thread_root
30 31 self._run_thread(run_thread) 32 self.assertEquals(xml, tostring(main_root))
33
34 - def test_main_xslt_in_thread(self):
35 XML = self.etree.XML 36 style = XML('''\ 37 <xsl:stylesheet version="1.0" 38 xmlns:xsl="http://www.w3.org/1999/XSL/Transform"> 39 <xsl:template match="*"> 40 <foo><xsl:copy><xsl:value-of select="/a/b/text()" /></xsl:copy></foo> 41 </xsl:template> 42 </xsl:stylesheet>''') 43 st = etree.XSLT(style) 44 45 result = [] 46 47 def run_thread(): 48 root = XML('<a><b>B</b><c>C</c></a>') 49 result.append( st(root) )
50 51 self._run_thread(run_thread) 52 self.assertEquals('''\ 53 <?xml version="1.0"?> 54 <foo><a>B</a></foo> 55 ''', 56 str(result[0])) 57
58 - def test_thread_xslt(self):
59 XML = self.etree.XML 60 tostring = self.etree.tostring 61 root = XML('<a><b>B</b><c>C</c></a>') 62 63 def run_thread(): 64 style = XML('''\ 65 <xsl:stylesheet version="1.0" 66 xmlns:xsl="http://www.w3.org/1999/XSL/Transform"> 67 <xsl:template match="*"> 68 <foo><xsl:copy><xsl:value-of select="/a/b/text()" /></xsl:copy></foo> 69 </xsl:template> 70 </xsl:stylesheet>''') 71 st = etree.XSLT(style) 72 root.append( st(root).getroot() )
73 74 self._run_thread(run_thread) 75 self.assertEquals('<a><b>B</b><c>C</c><foo><a>B</a></foo></a>', 76 tostring(root)) 77
78 - def test_thread_create_xslt(self):
79 XML = self.etree.XML 80 tostring = self.etree.tostring 81 root = XML('<a><b>B</b><c>C</c></a>') 82 83 stylesheets = [] 84 85 def run_thread(): 86 style = XML('''\ 87 <xsl:stylesheet 88 xmlns:xsl="http://www.w3.org/1999/XSL/Transform" 89 version="1.0"> 90 <xsl:output method="xml" /> 91 <xsl:template match="/"> 92 <div id="test"> 93 <xsl:apply-templates/> 94 </div> 95 </xsl:template> 96 </xsl:stylesheet>''') 97 stylesheets.append( etree.XSLT(style) )
98 99 self._run_thread(run_thread) 100 101 st = stylesheets[0] 102 result = tostring( st(root) ) 103 104 self.assertEquals('<div id="test">BC</div>', 105 result) 106
107 - def test_thread_mix(self):
108 XML = self.etree.XML 109 Element = self.etree.Element 110 SubElement = self.etree.SubElement 111 tostring = self.etree.tostring 112 xml = '<a><b>B</b><c xmlns="test">C</c></a>' 113 root = XML(xml) 114 fragment = XML("<other><tags/></other>") 115 116 result = self.etree.Element("{myns}root", att = "someval") 117 118 def run_XML(): 119 thread_root = XML(xml) 120 result.append(thread_root[0]) 121 result.append(thread_root[-1])
122 123 def run_parse(): 124 thread_root = self.etree.parse(StringIO(xml)).getroot() 125 result.append(thread_root[0]) 126 result.append(thread_root[-1]) 127 128 def run_move_main(): 129 result.append(fragment[0]) 130 131 def run_build(): 132 result.append( 133 Element("{myns}foo", attrib={'{test}attr':'val'})) 134 SubElement(result, "{otherns}tasty") 135 136 def run_xslt(): 137 style = XML('''\ 138 <xsl:stylesheet version="1.0" 139 xmlns:xsl="http://www.w3.org/1999/XSL/Transform"> 140 <xsl:template match="*"> 141 <foo><xsl:copy><xsl:value-of select="/a/b/text()" /></xsl:copy></foo> 142 </xsl:template> 143 </xsl:stylesheet>''') 144 st = etree.XSLT(style) 145 result.append( st(root).getroot()[0] ) 146 147 for test in (run_XML, run_parse, run_move_main, run_xslt): 148 tostring(result) 149 self._run_thread(test) 150 151 self.assertEquals( 152 '<ns0:root xmlns:ns0="myns" att="someval"><b>B</b><c xmlns="test">C</c><b>B</b><c xmlns="test">C</c><tags/><a>B</a></ns0:root>', 153 tostring(result)) 154 155 def strip_first(): 156 root = Element("newroot") 157 root.append(result[0]) 158 159 while len(result): 160 self._run_thread(strip_first) 161 162 self.assertEquals( 163 '<ns0:root xmlns:ns0="myns" att="someval"/>', 164 tostring(result)) 165 166
167 -def test_suite():
168 suite = unittest.TestSuite() 169 suite.addTests([unittest.makeSuite(ThreadingTestCase)]) 170 return suite
171 172 if __name__ == '__main__': 173 print 'to test use test.py %s' % __file__ 174