- made the C version also pass the test

3 files changed, 113 insertions, 11 deletions

diff --git a/tests/all.py b/tests/all.py
index 3477c2e..f19ee15 100755
--- a/tests/all.py
+++ b/tests/all.py
@@ -4,7 +4,7 @@ import unittest
 import sys
 import getopt
 
-all_modules=["vcard","jid","disco","imports","cache","stream_reader"]
+all_modules=["vcard","jid","disco","imports","cache","stream_reader", "ns_operations"]
 
 def suite(modules=None):
      if not modules:
diff --git a/tests/ns_operations.py b/tests/ns_operations.py
new file mode 100755
index 0000000..2c45eda
--- /dev/null
+++ b/tests/ns_operations.py
@@ -0,0 +1,69 @@
+#!/usr/bin/python
+# -*- coding: UTF-8 -*-
+
+import unittest
+import libxml2
+from pyxmpp import xmlextra
+from stream_reader import xml_elements_equal
+
+input_xml = """<?xml version="1.0" ?>
+<root xmlns="http://pyxmpp.jabberstudio.org/xmlns/test" xmlns:prefix="http://pyxmpp.jabberstudio.org/xmlns/test1">
+   <a> <a1/> <a2/> </a>
+   <b xmlns="http://pyxmpp.jabberstudio.org/xmlns/test2">
+     <c/>
+     <prefix:d/>
+   </b>
+   <prefix:e/>
+   <f/>
+</root>
+"""
+
+output_xml = """<?xml version="1.0" ?>
+<root xmlns="http://pyxmpp.jabberstudio.org/xmlns/common" xmlns:prefix="http://pyxmpp.jabberstudio.org/xmlns/test1">
+   <a> <a1/> <a2/> </a>
+   <b xmlns="http://pyxmpp.jabberstudio.org/xmlns/test2">
+     <c/>
+     <prefix:d/>
+   </b>
+   <prefix:e/>
+   <f/>
+</root>
+"""
+
+input_doc = libxml2.parseDoc(input_xml)
+input_root = input_doc.getRootElement()
+output_doc = libxml2.parseDoc(output_xml)
+output_root = output_doc.getRootElement()
+
+class TestReplaceNs(unittest.TestCase):
+    def test_replace_ns(self):
+        doc = libxml2.newDoc("1.0")
+        
+        root = doc.newChild(None, "root", None)
+        common_ns = root.newNs(xmlextra.COMMON_NS, None)
+        root.setNs(common_ns)
+        doc.setRootElement(root)
+        
+        n = input_doc.getRootElement()
+        input_ns = n.ns()
+        n = n.children
+        while n:
+            n1 = n.docCopyNode(doc, 1)
+            root.addChild(n1)
+            if n1.type == 'element':
+                n1_ns = n1.ns()
+                if n1_ns.content == input_ns.content:
+                    xmlextra.replace_ns(n1, n1_ns, common_ns)
+            n = n.next
+        #print doc.serialize()
+        self.failUnless(xml_elements_equal(root, output_root))
+
+def suite():
+     suite = unittest.TestSuite()
+     suite.addTest(unittest.makeSuite(TestReplaceNs))
+     return suite
+
+if __name__ == '__main__':
+    unittest.TextTestRunner(verbosity=2).run(suite())
+
+# vi: sts=4 et sw=4 encoding=utf-8
diff --git a/tests/stream_reader.py b/tests/stream_reader.py
index 31386c1..047edb2 100755
--- a/tests/stream_reader.py
+++ b/tests/stream_reader.py
@@ -7,8 +7,9 @@ from pyxmpp import xmlextra
 from pyxmpp.jid import JID,JIDError
 from pyxmpp import xmppstringprep
 
-def xml_elements_equal(a,b):
+def xml_elements_equal(a, b, ignore_level1_cdata = False):
     if a.name!=b.name:
+        print "Name mismatch: %r, %r" % (a.name, b.name)
         return False
     try:
         ns1 = a.ns()
@@ -20,8 +21,10 @@ def xml_elements_equal(a,b):
         ns2 = None
     if ns1 or ns2:
         if None in (ns1,ns2):
+            print "Ns mismatch: %r, %r" % (ns1, ns2)
             return False
         if ns1.content != ns2.content:
+            print "Ns mismatch: %r, %r on %r, %r" % (ns1.content, ns2.content, a, b)
             return False
  
     ap = a.properties
@@ -40,9 +43,14 @@ def xml_elements_equal(a,b):
    
     ac = a.children
     bc = b.children
-    while 1:
-        if (ac, bc) == (None, None):
-            return True
+    while ac != None or bc != None:
+        if ignore_level1_cdata:
+            if ac and ac.type!='element':
+                ac=ac.next
+                continue
+            if bc and bc.type!='element':
+                bc=bc.next
+                continue
         if None in (ac, bc):
             return False
         if ac.type != bc.type:
@@ -90,6 +98,7 @@ class StreamHandler(xmlextra.StreamHandler):
         self.test_case.event("node", node)
 
 expected_events = []
+whole_stream = None
 
 def load_expected_events():
     for l in file("data/stream_info.txt"):
@@ -98,6 +107,10 @@ def load_expected_events():
         l=l.strip()
         expected_events.append(EventTemplate(l))
 
+def load_whole_stream():
+    global whole_stream
+    whole_stream = libxml2.parseFile("data/stream.xml")
+
 class TestStreamReader(unittest.TestCase):
     def setUp(self):
         self.expected_events = list(expected_events)
@@ -106,21 +119,27 @@ class TestStreamReader(unittest.TestCase):
         self.file = file("data/stream.xml")
         self.chunk_start = 0
         self.chunk_end = 0
+        self.whole_stream = libxml2.newDoc("1.0")
+
+    def tearDown(self):
+        del self.handler
+        del self.reader
+        self.whole_stream.freeDoc()
         
     def test_1(self):
-        return self.do_test(1)
+        self.do_test(1)
         
     def test_2(self):
-        return self.do_test(2)
+        self.do_test(2)
         
     def test_10(self):
-        return self.do_test(10)
+        self.do_test(10)
         
     def test_100(self):
-        return self.do_test(100)
+        self.do_test(100)
         
     def test_1000(self):
-        return self.do_test(1000)
+        self.do_test(1000)
         
     def do_test(self, chunk_length):
         while 1:
@@ -135,6 +154,11 @@ class TestStreamReader(unittest.TestCase):
                 self.event("end", None)
                 break
             self.chunk_start = self.chunk_end
+        r1 = self.whole_stream.getRootElement()
+        r2 = whole_stream.getRootElement()
+        if not xml_elements_equal(r1, r2, True):
+            self.fail("Whole stream invalid. Got: %r, Expected: %r"
+                    % (self.whole_stream.serialize(), whole_stream.serialize()))
 
     def event(self, event, node):
         expected = self.expected_events.pop(0)
@@ -148,9 +172,18 @@ class TestStreamReader(unittest.TestCase):
         if not expected.match(event,node):
             self.fail("Unmatched event. Expected: %r, got: %r;%r" 
                     % (expected, event, node.serialize()))
-        
+        if event == "start":
+            n = node.docCopyNode(self.whole_stream,1)
+            self.whole_stream.addChild(n)
+            self.whole_stream.setRootElement(n)
+        elif event == "node":
+            n = node.docCopyNode(self.whole_stream,1)
+            r = self.whole_stream.getRootElement()
+            r.addChild(n)
+            
 def suite():
     load_expected_events()
+    load_whole_stream()
     suite = unittest.TestSuite()
     suite.addTest(unittest.makeSuite(TestStreamReader))
     return suite