diff --git a/test/common.py b/test/common.py index eda155d..78c6b9c 100644 --- a/test/common.py +++ b/test/common.py @@ -1,401 +1,401 @@ -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # import os import re import socket import sys import time import traceback import types import unittest from copy import copy from configparser import SafeConfigParser as ConfigParser from lxml import etree from optparse import OptionParser, SUPPRESS_HELP from xcaplib import xcapclient apps = ['pres-rules', 'org.openmobilealliance.pres-rules', 'resource-lists', 'pidf-manipulation', 'watchers', 'rls-services', 'test-app', 'xcap-caps'] def succeed(r): return 200 <= r.status <= 299 class XCAPTest(unittest.TestCase): # if true, each PUT or DELETE will be followed by GET to ensure that it has indeed succeeded invariant_check = True @classmethod def setupOptionParser(cls, parser): xcapclient.setup_parser_client(parser) def initialize(self, options, args = []): if not hasattr(self, '_options'): self._options = copy(options) if not hasattr(self, '_args'): self._args = copy(args) def new_client(self): return xcapclient.make_xcapclient(self.options) def update_client_options(self): self.client = self.new_client() def setUp(self): self.options = copy(self._options) self.args = copy(self._args) self.update_client_options() def assertStatus(self, r, status, msg=None): if status is None: return elif isinstance(status, int): if r.status != status: if msg is None: msg = 'Status (%s) != %s' % (r.status, status) raise self.failureException(msg) else: ## status is a tuple or a list if r.status not in status: if msg is None: msg = 'Status (%s) not in %s' % (r.status, str(status)) raise self.failureException(msg) def assertHeader(self, r, key, value=None, msg=None): """Fail if (key, [value]) not in r.headers.""" lowkey = key.lower() for k, v in list(r.headers.items()): if k.lower() == lowkey: if value is None or str(value) == v: return v if msg is None: if value is None: msg = '%s not in headers' % key else: msg = '%s:%s not in headers' % (key, value) raise self.failureException(msg) def assertETag(self, r): v = self.assertHeader(r, 'ETag') return r.etag def assertNoHeader(self, r, key, msg=None): """Fail if key in r.headers.""" lowkey = key.lower() matches = [k for k, v in r.headers if k.lower() == lowkey] if matches: if msg is None: msg = '%s in headers' % key raise self.failureException(msg) def assertBody(self, r, value, msg=None): """Fail if value != r.body.""" if value != r.body: if msg is None: msg = 'expected body:\n"%s"\n\nactual body:\n"%s"' % (value, r.body) raise self.failureException(msg) def assertInBody(self, r, value, msg=None): """Fail if value not in r.body.""" if value not in r.body: if msg is None: msg = '%r not in body\nbody: %r' % (value, r.body) raise self.failureException(msg) def assertNotInBody(self, r, value, msg=None): """Fail if value in r.body.""" if value in r.body: if msg is None: msg = '%s found in body' % value raise self.failureException(msg) def assertMatchesBody(self, r, pattern, msg=None, flags=0): """Fail if value (a regex pattern) is not in r.body.""" if re.search(pattern, r.body, flags) is None: if msg is None: msg = 'No match for %s in body' % pattern raise self.failureException(msg) def assertDocument(self, application, body, client=None): r = self.get(application, client=client) self.assertBody(r, body) def get(self, application, node=None, status=200, **kwargs): client = kwargs.pop('client', None) or self.client r = client._get(application, node, **kwargs) self.validate_error(r, application) self.assertStatus(r, status) if 200<=status<=299: self.assertHeader(r, 'ETag') return r def get_global(self, *args, **kwargs): kwargs['globaltree'] = True return self.get(*args, **kwargs) def put(self, application, resource, node=None, status=[200,201], content_type_in_GET=None, client=None, **kwargs): client = client or self.client r_put = client._put(application, resource, node, **kwargs) self.validate_error(r_put, application) self.assertStatus(r_put, status) # if PUTting succeed, check that document is there and equals to resource if self.invariant_check and succeed(r_put): r_get = self.get(application, node, status=None, client=client) self.assertStatus(r_get, 200, 'although PUT succeed, following GET on the same URI did not: %s %s' % \ (r_get.status, r_get.reason)) self.assertEqual(resource.strip(), r_get.body) # is body put equals to body got? if content_type_in_GET is not None: self.assertHeader(r_get, 'content-type', content_type_in_GET) return r_put def put_new(self, application, resource, node=None, status=201, content_type_in_GET=None, client=None): # QQQ use If-None-Match or some other header to do that without get self.get(application, node=node, status=404, client=client) return self.put(application, resource, node, status, content_type_in_GET, client) def delete(self, application, node=None, status=200, client=None, **kwargs): client = client or self.client r = client._delete(application, node, **kwargs) self.validate_error(r, application) self.assertStatus(r, status) # if deleting succeed, GET should return 404 if self.invariant_check and succeed(r) or r.status == 404: r_get = self.get(application, node, status=None) self.assertStatus(r_get, 404, 'although DELETE succeed, following GET on the same URI did not return 404: %s %s' % \ (r_get.status, r_get.reason)) return r def put_rejected(self, application, resource, status=409, client=None): """DELETE the document, then PUT it and expect 409 error. Return PUT result. If PUT has indeed failed, also check that GET returns 404 """ self.delete(application, status=[200,404], client=client) put_result = self.put(application, resource, status=status, client=client) self.get(application, status=404, client=client) return put_result def getputdelete(self, application, document, content_type, client=None): self.delete(application, status=[200,404], client=client) self.get(application, status=404, client=client) self.put(application, document, status=201, content_type_in_GET=content_type, client=client) self.put(application, document, status=200, content_type_in_GET=content_type, client=client) self.put(application, document, status=200, content_type_in_GET=content_type, client=client) self.delete(application, status=200, client=client) self.delete(application, status=404, client=client) def validate_error(self, r, application): if r.status==409 or r.headers.gettype()=='application/xcap-error+xml': self.assertEqual(r.headers.gettype(), 'application/xcap-error+xml') xml = validate_xcaps_error(r.body) if ' Bill Doe Close Friends Joe Smith Nancy Gross Marketing """ def get_xcapdiff(xcap_root, resource, username, old_etag, new_etag): uri = xcap_root + '/' + resource + '/users/' + username + '/index.xml' return xml_xcapdiff(xcap_root, xml_document(uri, old_etag, new_etag)) queue = Queue() packet_count = 0 start_time = None is_subscribed = False def event_handler(event_name, **kwargs): global start_time, packet_count, is_subscribed if event_name == "Subscription_state": if kwargs["state"] == "ACTIVE": is_subscribed = True #elif kwargs["state"] == "TERMINATED": # if kwargs.has_key("code"): # print "Unsubscribed: %(code)d %(reason)s" % kwargs # else: # print "Unsubscribed" elif event_name == "Subscription_notify": queue.put(("NOTIFY", kwargs)) elif event_name == "siptrace": if start_time is None: start_time = kwargs["timestamp"] packet_count += 1 if kwargs["received"]: direction = "RECEIVED" else: direction = "SENDING" print("%s: Packet %d, +%s" % (direction, packet_count, (kwargs["timestamp"] - start_time))) print("%(timestamp)s: %(source_ip)s:%(source_port)d --> %(destination_ip)s:%(destination_port)d" % kwargs) print(kwargs["data"]) elif event_name=='log': pass else: print('UNHANDLED EVENT', event_name, kwargs) def get(queue, blocking=True, timeout=1): try: return queue.get(blocking, timeout) except Empty: return None class Test(XCAPTest): def assertContains(self, element, list): if element not in list: raise self.failureException("%s not in %s" % (element, list)) @classmethod def setupOptionParser(_cls, parser): parser.add_option("-p", "--outbound-proxy", type="string", action="callback", callback=parse_proxy_cb, help="Outbound SIP proxy to use. By default a lookup is performed based on SRV and A records.", metavar="IP[:PORT]") parser.add_option("-t", "--siptrace", default=False, action='store_true') setup_parser_client(parser) def test(self): opts = self.options self.delete(resource, status=[200,404]) initial_events = Engine.init_options_defaults["initial_events"] if content_type is not None: initial_events[event] = [content_type] e = Engine(event_handler, do_siptrace=opts.siptrace, auto_sound=False, initial_events=initial_events) e.start() try: if opts.outbound_proxy is None: route = None else: route = Route(opts.proxy_ip, opts.proxy_port) sub = Subscription(Credentials(SIPURI(user=opts.username, host=opts.domain), opts.password), SIPURI(user=opts.username, host=opts.domain), event, route=route, expires=expires) sub.subscribe() try: # wait for SUBSCRIBE to succeed AND absorb out-of-date NOTIFYs end = time.time() + 1.5 while time.time() < end: get(queue, timeout=0.1) self.assertTrue(is_subscribed, 'SUBSCRIBE failed') # try: # X = queue.get(True, timeout = 1) # except Empty: # pass # else: # self.assertEqual(X[0], 'NOTIFY') def get_notify(comment = ''): try: X = queue.get(True, timeout = 1) except Empty: self.fail("Didn't get a NOTIFY %s" % comment) self.assertEqual(X[0], 'NOTIFY') return X[1] r = self.put(resource, body) etag = r.headers['ETag'].strip('"') X = get_notify('after put') xcap_root = opts.xcap_root.replace(':8000', '') self.assertEqual(X['body'], get_xcapdiff(xcap_root, resource, opts.username, None, etag)) #print etag r = self.put(resource, body.replace('Close', 'Intimate')) new_etag = r.headers['ETag'].strip('"') X = get_notify() self.assertEqual(X['body'], get_xcapdiff(xcap_root, resource, opts.username, etag, new_etag)) #print etag, new_etag r = self.delete(resource) X = get_notify() self.assertEqual(X['body'], get_xcapdiff(xcap_root, resource, opts.username, new_etag, None)) #print new_etag, None finally: sub.unsubscribe() time.sleep(2) finally: e.stop() re_ip_port = re.compile("^(?P\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})(:(?P\d+))?$") def parse_proxy(value, parser): match = re_ip_port.match(value) if match is None: raise OptionValueError("Could not parse supplied outbound proxy address") parser.values.proxy_ip = match.group('proxy_ip') parser.values.proxy_port = int(match.group('proxy_port') or '5060') def parse_proxy_cb(_option, _opt_str, value, parser): return parse_proxy(value, parser) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test.py b/test/test.py index ffc6956..c636511 100755 --- a/test/test.py +++ b/test/test.py @@ -1,64 +1,64 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # import sys import os import traceback from . import common as c class TestHarness(object): """A test harness for OpenXCAP.""" def __init__(self, tests, option_parser): """Constructor to populate the TestHarness instance. tests should be a list of module names (strings). """ self.tests = tests self.option_parser = option_parser self.test_suites = [] self.import_errors = 0 for testmod in self.tests: try: self.import_errors += 1 m = __import__(testmod, globals(), locals()) suite = c.loadSuiteFromModule(m, option_parser) suite.modname = testmod self.test_suites.append(suite) self.import_errors -= 1 except Exception: traceback.print_exc() def run(self, options, args): c.run_suite(c.TestSuite(self.test_suites), options, args) def all_tests(): my_dir = os.path.dirname(os.path.abspath(__file__)) lst = [x.strip('.py') for x in os.listdir(my_dir) if x.startswith('test_') and x.endswith('.py')] return lst def run(): parser = c.prepare_optparser() parser.add_option("-l", "--list", action="store_true", help="Print list of all tests") t = TestHarness(all_tests(), parser) options, args = parser.parse_args() if options.list: for x in t.test_suites: print(x.modname) for i in x: print(' - ', i) print() return c.process_options(options) c.run_command(lambda : t.run(options, args), options) if t.import_errors: sys.exit('there were import errors!\n') if __name__ == '__main__': run() diff --git a/test/test_attribute.py b/test/test_attribute.py index 5f2c35b..86b8773 100755 --- a/test/test_attribute.py +++ b/test/test_attribute.py @@ -1,69 +1,69 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * resource_list_xml = """ Joe Smith Nancy Gross Petri Aukia """ class AttributeTest(XCAPTest): def test_get(self): self.put('resource-lists', resource_list_xml) self.get('resource-lists', '/resource-lists/list[@name="other"]/@some-attribute', status=404) r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/@name') self.assertBody(r, "friends") self.assertHeader(r, 'ETag') self.assertHeader(r, 'Content-type', 'application/xcap-att+xml') r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/external/@anchor') uri = 'http://xcap.example.org/resource-lists/users/sip:a@example.org/index/~~/resource-lists/list%5b@name=%22mkting%22%5d' self.assertBody(r, uri) print('WARNING: test with URI in att_value is disabled') # r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/external[@anchor="%s"]/@anchor' % uri) # self.assertBody(r, uri) r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/external[]/@anchor', status=400) def test_delete(self): self.put('resource-lists', resource_list_xml) self.delete('resource-lists', '/resource-lists/list[@name="other"]/@some-attribute', status=404) # XXX is it legal for parent selector (/resource-lists/list[@name="friends"]) to become invalid? # I don't think it is, check with RFC self.delete('resource-lists', '/resource-lists/list[@name="friends"]/@name', status=200) self.delete('resource-lists', '/resource-lists/list[@name="friends"]/@name', status=404) def test_put(self): self.put('resource-lists', resource_list_xml) self.put('resource-lists', 'coworkers', '/resource-lists/list[@name="friends"]/@some-attribute', status=409) # fails GET(PUT(x))==x test. must be rejected in the server #self.put('resource-lists', 'coworkers', '/resource-lists/list[@name="friends"]/@name', status=409) # XXX parent's selector becomes invalid r = self.client._put('resource-lists', 'coworkers', '/resource-lists/list[@name="friends"]/@name') self.assertStatus(r, 200) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_auth.py b/test/test_auth.py index fdd1b9d..2ede9d2 100755 --- a/test/test_auth.py +++ b/test/test_auth.py @@ -1,44 +1,44 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * class AuthTest(XCAPTest): def test_users_auth(self): self.get(self.app, status=[200,404]) self.options.password += 'x' self.update_client_options() self.get(self.app, status=[401]) def test_global_auth(self): self.get_global(self.app, status=[200,404]) #self.options.password += 'x' #self.update_client_options() #for app in apps: # self.get_global(app, status=401) # XXX test PUT/DELETE auth as well? # XXX test digest authentication # XXX test authorization #def test_authorization(self): ### the request cannot be authorized (we're trying to access someone else' resource) #account = self.account #self.account = "dummy" + self.account #r = self.get('resource-lists', status=401) #self.client.account = account for app in apps: exec("""class AuthTest_%s(AuthTest): app = %r """ % (app.replace('-', '_').replace('.', '_'), app)) del AuthTest if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_element.py b/test/test_element.py index 64215e8..5455de8 100755 --- a/test/test_element.py +++ b/test/test_element.py @@ -1,112 +1,112 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * xml = """ Joe Smith Nancy Gross Petri Aukia """ def index(s, sub, skip=0, start=0): while skip >= 0: found = s.index(sub, start) skip -= 1 start = found + 1 return found def eindex(s, sub, skip=0): return index(s, sub, skip)+len(sub) lst = xml[xml.index('')] nancy = xml[xml.index(' Alice """ external = xml[xml.index('')] class ElementTest(XCAPTest): def test_get(self): self.delete('resource-lists', status=[200,404]) self.put('resource-lists', xml) self.get('resource-lists', '/resource-lists/list[@name="other"]', status=404) self.get('resource-lists', '/resource-lists/list/entry[4]', status=404) r = self.get('resource-lists', '/resource-lists/list[@name="friends"]') self.assertBody(r, lst) self.assertHeader(r, 'ETag') self.assertHeader(r, 'Content-type', 'application/xcap-el+xml') r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/entry[2]') self.assertBody(r, nancy) self.assertHeader(r, 'ETag') self.assertHeader(r, 'Content-type', 'application/xcap-el+xml') r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/*[2]') self.assertBody(r, nancy) self.assertHeader(r, 'ETag') self.assertHeader(r, 'Content-type', 'application/xcap-el+xml') print('WARNING: test with URI in att_value is disabled') # r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/external[@anchor="http://xcap.example.org/resource-lists/users/sip:a@example.org/index/~~/resource-lists/list%5b@name="mkting"5d"]') # self.assertBody(r, external) # self.assertHeader(r, 'ETag') # self.assertHeader(r, 'Content-type', 'application/xcap-el+xml') def test_delete(self): self.put('resource-lists', xml) # cannot delete something in the middle self.delete('resource-lists', '/resource-lists/list[@name="friends"]/entry[2]', status=409) self.delete('resource-lists', '/resource-lists/list[@name="friends"]/*[3]', status=409) # it's ok to delete the last one though r = self.delete('resource-lists', '/resource-lists/list[@name="friends"]/*[4]') self.assertHeader(r, 'ETag') r = self.delete('resource-lists', '/resource-lists/list[@name="friends"]/*[3]') self.assertHeader(r, 'ETag') r = self.delete('resource-lists', '/resource-lists/list[@name="friends"]/*[2]') self.assertHeader(r, 'ETag') r = self.delete('resource-lists', '/resource-lists/list[@name="friends"]/entry') self.assertHeader(r, 'ETag') r = self.get('resource-lists', '/resource-lists/list') self.assertMatchesBody(r, '^\\s*$') self.delete('resource-lists', '/resource-lists/list[@name="friends"]/entry[@uri="sip:joe@example.com"]', status=404) def test_put_error(self): self.put('resource-lists', xml) # 415 content type not set self.put('resource-lists', nancy, '/resource-lists/list[@name="friends"]', headers={'Content-Type' : None},status=415) # 409 r = self.put('resource-lists', broken, '/resource-lists/list[@name="friends"]', status=409) # 409 r = self.put('resource-lists', nancy, '/resource-lists/list[@name="others"]/entry[2]', status=409) # 409 r = self.put('resource-lists', nancy, '/resource-lists/list[@name="friends"]/entry[1]', status=409) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_element_put.py b/test/test_element_put.py index 72fa671..f64ccd3 100755 --- a/test/test_element_put.py +++ b/test/test_element_put.py @@ -1,177 +1,177 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * app = 'test-app' start = ''' ''' # when changing to the document could be put, but # element GET respons with 404. # either GET should return what expected or a document without namespaces declaration # should be rejected class PutElementTest(XCAPTest): def reverse(self, node_selector): self.delete(app, node_selector) self.assertDocument(app, start) def test_creation(self): """Testing different ways of inserting an element as described in examples from Section 8.2.3 (http://tools.ietf.org/html/rfc4825#section-8.2.3) After each PUT, DELETE is executed on the same URI and the resulting document must be the same as before the insertion. """ self.put(app, start) for node_selector in ['/root/el1[@att="third"]', '/root/el1[3][@att="third"]', '/root/*[3][@att="third"]']: self.put_new(app, '', node_selector) self.assertDocument(app, ''' ''') self.reverse(node_selector) # out-of-bound positional index in node selector results in 409 (XXX or 404?) for node_selector in ['root/el1[4][@att="third"]', 'root/*[0][@att="third"]']: self.put_new(app, '', node_selector, status=409) self.assertDocument(app, start) # replace 500 with something more appropriate #for node_selector in ['root/*[-1][@att="third"]']: # self.put_new(app, '', node_selector, status=500) # self.assertDocument(app, start) # following request would fail idempotency requirement (GET(PUT(x))=>x) if succeeded for node_selector in ['root/el1[@att="third"]', 'root/el1[3][@att="third"]', 'root/*[3][@att="third"]']: r = self.put_new(app, '', node_selector, status=409) self.assertInBody(r, 'cannot-insert') self.assertDocument(app, start) self.put_new(app, '', 'root/el3') self.assertDocument(app, ''' ''') self.reverse('root/el3') for node_selector in ['root/el2[@att="2"]', 'root/el2[2][@att="2"]']: self.put_new(app, '', node_selector) self.assertDocument(app, ''' ''') self.reverse(node_selector) self.put_new(app, '', 'root/*[2][@att="2"]') self.assertDocument(app, ''' ''') self.reverse('root/*[2][@att="2"]') self.put_new(app, '', 'root/el2[1][@att="2"]') self.assertDocument(app, ''' ''') self.reverse('root/el2[1][@att="2"]') def test_creation_starattr(self): """Testing PUT requests of form '*[@att="some"]' which require looking into body of PUT""" self.put(app, start) for selector in ['root/*[@att="2"]', 'root/el1[@att="2"]']: self.put_new(app, '', selector) self.assertDocument(app, ''' ''') self.reverse(selector) # the same request - different body for selector in ['root/*[@att="2"]', 'root/el2[@att="2"]']: self.put_new(app, '', selector) self.assertDocument(app, ''' ''') self.reverse(selector) # the same request - different body for selector in ['root/*[@att="2"]', 'root/el3[@att="2"]']: self.put_new(app, '', selector) self.assertDocument(app, ''' ''') self.reverse(selector) def test_replacement(self): self.put(app, start) for node_selector in ['root/el1[@att="first"]', 'root/el1[1][@att="first"]', 'root/*[1][@att="first"]']: self.put(app, '', node_selector, status=409) self.assertDocument(app, start) for node_selector in ['root/el1[1]', 'root/*[1]']: self.put(app, start) self.put(app, '', node_selector, status=200) self.assertDocument(app, ''' ''') if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_errors.py b/test/test_errors.py index 54dd6fa..0f9bfac 100755 --- a/test/test_errors.py +++ b/test/test_errors.py @@ -1,73 +1,73 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from . import common as c from urllib.parse import urlparse class ErrorsTest(c.XCAPTest): def communicate(self, data): s = c.socket.socket() x = urlparse(self.options.xcap_root) if x.port is None: port = {'http': 80, 'https': 443}.get(x.scheme) s.connect((x.hostname, x.port or port)) if x.scheme == 'https': s = c.socket.ssl(s) s.write(data) return s.read(1024*8) s.send(data) return s.recv(1024*8) def test_gibberish(self): response = self.communicate('\r\r\r\n\r\n') assert '400 Bad Request' in response, repr(response) def test409(self): self.put('resource-lists', 'xxx', status=409) def check(self, code, message, *uris): for uri in uris: r = self.client.con.request('GET', uri) self.assertEqual(r.status, code) self.assertInBody(r, message) def test400_1(self): self.get('resource-lists', '/resource-lists/list[@name="friends"]/external[]/@anchor', status=400) def test400_2(self): self.check(400, "to parse node", 'resource-lists/users/alice@example.com/index.xml~~') def test404(self): self.check(404, 'XCAP Root', '') self.check(404, 'context', 'xxx') self.check(404, "context", 'resource-lists/user/alice@example.com/index.xml') self.check(404, 'user id', 'resource-lists/users') self.check(404, "not contain ", 'resource-lists/users/alice@example.com', 'resource-lists/users/alice@example.com/') # XXX test for multiple matches def test405(self): r = self.client.con.request('POST', '') self.assertEqual(r.status, 405) r = self.client.con.request('XXX', '') self.assertEqual(r.status, 405) # but apache responds with 501 # 412: tested in test_etags.py if __name__ == '__main__': c.runSuiteFromModule() diff --git a/test/test_etags.py b/test/test_etags.py index a478d6c..d0bf860 100755 --- a/test/test_etags.py +++ b/test/test_etags.py @@ -1,113 +1,113 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * resource_list_xml = """ """ class ETagTest(XCAPTest): def test_conditional_GET(self): r = self.put('resource-lists', resource_list_xml) etag = self.assertHeader(r, 'ETag') # Test If-Match (both valid and invalid) self.get('resource-lists', headers={'If-Match': etag}) self.get('resource-lists', headers={'If-Match': '*'}) self.get('resource-lists', headers={'if-Match': "another-etag"}, status=412) # Test If-None-Match (both valid and invalid) self.get('resource-lists', headers={'If-None-Match': etag}, status=304) self.get('resource-lists', headers={'If-None-Match': '*'}, status=304) self.get('resource-lists', headers={'If-None-Match': "another-etag"}, status=200) def test_conditional_PUT(self): self.delete('resource-lists', status=[200,404]) self.get('resource-lists', status=404) # Test conditional PUT when document doesn't exist self.put('resource-lists', resource_list_xml, headers={'If-Match': '12345asdf'}, status=412) r = self.put('resource-lists', resource_list_xml) etag = self.assertHeader(r, 'ETag') # Test conditional PUT logic ## Alice and Bob initially share the same etag alice_etag = bob_etag = etag ## Bob modifies the resource r = self.put('resource-lists', resource_list_xml, headers={'If-Match': bob_etag}) bob_etag = self.assertHeader(r, 'ETag') ## now Alice tries to modify the resource self.put('resource-lists', resource_list_xml, headers={'If-Match': alice_etag}, status=412) ## the etag has changed so now she updates her in-memory document r = self.get('resource-lists') new_alice_etag = self.assertHeader(r, 'ETag') self.assertEqual(bob_etag, new_alice_etag) self.put('resource-lists', resource_list_xml, headers={'If-Match': new_alice_etag}) def test_conditional_PUT_2(self): self.delete('resource-lists', status=[200,404]) self.get('resource-lists', status=404) self.put('resource-lists', resource_list_xml, headers={'If-None-Match': '*'}, status=201) self.put('resource-lists', resource_list_xml, headers={'If-None-Match': '*'}, status=412) class ETagTest2(XCAPTest): # the same as prev, but using 'etag' param def test_conditional_GET(self): r = self.put('resource-lists', resource_list_xml) etag = self.assertHeader(r, 'ETag') # Test If-Match (both valid and invalid) self.get('resource-lists', etag=etag) self.get('resource-lists') self.get('resource-lists', etag="another-etag", status=412) def test_conditional_PUT(self): r = self.put('resource-lists', resource_list_xml) etag = self.assertETag(r) assert etag is not None, repr(etag) # Test conditional PUT logic ## Alice and Bob initially share the same etag alice_etag = bob_etag = etag ## Bob modifies the resource r = self.put('resource-lists', resource_list_xml, etag=bob_etag) bob_etag = self.assertETag(r) ## now Alice tries to modify the resource self.put('resource-lists', resource_list_xml, etag=alice_etag, status=412) ## the etag has changed so now she updates her in-memory document r = self.get('resource-lists') new_alice_etag = self.assertETag(r) self.assertEqual(bob_etag, new_alice_etag) self.put('resource-lists', resource_list_xml, etag=new_alice_etag) def test_etag_parsing(self): r = self.put('resource-lists', resource_list_xml) etag = self.assertETag(r) # no quotes r = self.put('resource-lists', resource_list_xml, headers = {'if-match' : 'xxx' }, status=412) r = self.put('resource-lists', resource_list_xml, headers = {'if-match' : etag }, status=200) etag = self.assertETag(r) r = self.put('resource-lists', resource_list_xml, headers = {'if-match' : '"' + etag + '"' }, status=200) etag = self.assertETag(r) self.put('resource-lists', resource_list_xml, headers = {'if-match' : '"' + etag + 'xx"' }, status=412) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_etags2.py b/test/test_etags2.py index 1decb87..aea4c74 100755 --- a/test/test_etags2.py +++ b/test/test_etags2.py @@ -1,53 +1,53 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * resource_list_xml = """ """ class ETagTest(XCAPTest): def test_conditional_PUT(self): self.delete('resource-lists', status=[200,404]) self.get('resource-lists', status=404) # Test conditional PUT when document doesn't exist self.put('resource-lists', resource_list_xml, headers={'If-Match': '12345asdf'}, status=412) # r = self.put('resource-lists', resource_list_xml) # etag = self.assertHeader(r, 'ETag') # # # Test conditional PUT logic # ## Alice and Bob initially share the same etag # alice_etag = bob_etag = etag # # ## Bob modifies the resource # r = self.put('resource-lists', resource_list_xml, headers={'If-Match': bob_etag}) # bob_etag = self.assertHeader(r, 'ETag') # # ## now Alice tries to modify the resource # self.put('resource-lists', resource_list_xml, headers={'If-Match': alice_etag}, status=412) # # ## the etag has changed so now she updates her in-memory document # r = self.get('resource-lists') # new_alice_etag = self.assertHeader(r, 'ETag') # self.assertEqual(bob_etag, new_alice_etag) # # self.put('resource-lists', resource_list_xml, headers={'If-Match': new_alice_etag}) # def test_conditional_PUT_2(self): self.delete('resource-lists', status=[200,404]) self.get('resource-lists', status=404) self.put('resource-lists', resource_list_xml, headers={'If-None-Match': '*'}, status=201) self.put('resource-lists', resource_list_xml, headers={'If-None-Match': '*'}, status=412) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_fragment.py b/test/test_fragment.py index 3dc3c72..72268fa 100755 --- a/test/test_fragment.py +++ b/test/test_fragment.py @@ -1,50 +1,50 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from . import common document = """ Foo """ # well-formed fragment that would've been rejected by XML parser because of # unbound namespace prefix fragment = """ Test """ node = '/resource-lists/list/entry[@uri="sip:xxx@yyyyy.net"]' class FragmentTest(common.XCAPTest): def test_success(self): self.put('resource-lists', document) self.put('resource-lists', fragment, node) def test_errors(self): self.put('resource-lists', document) r = self.put('resource-lists', "", node, status=409) self.assertInBody(r, 'mismatched tag') r = self.put('resource-lists', "", node, status=409) self.assertInBody(r, 'not well-formed (invalid token)') r = self.put('resource-lists', "", node, status=409) self.assertInBody(r, 'not well-formed (invalid token)') r = self.put('resource-lists', "", node, status=409) self.assertInBody(r, 'junk after document element') r = self.put('resource-lists', "", node, status=409) self.assertInBody(r, 'not well-formed (invalid token)') if __name__ == '__main__': common.runSuiteFromModule() diff --git a/test/test_global.py b/test/test_global.py index 6bc03b6..d472c5f 100755 --- a/test/test_global.py +++ b/test/test_global.py @@ -1,32 +1,32 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * has_global = ['xcap-caps'] no_global = set(apps) - set(has_global) class TestGlobal(XCAPTest): def test_no_global(self): for app in no_global: self.get(app, status=404, globaltree=True) # at the moment, no one authorized to do that # NOTE, even though 404 would be also a valid response here, 401 should take priority # 404 or 401? # self.put(app, xml, status=401, globaltree=True) # self.delete(app, status=401, globaltree=True) def test_has_global(self): for app in has_global: self.get(app, status=200, globaltree=True) # # at the moment, no one authorized to do that # #self.put(app, xml, status=401, globaltree=True) # self.delete(app, status=401, globaltree=True) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_nsbindings.py b/test/test_nsbindings.py index 4c9f11b..e5a26c8 100755 --- a/test/test_nsbindings.py +++ b/test/test_nsbindings.py @@ -1,33 +1,33 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * resource_list_xml = """ Joe Smith Nancy Gross Petri Aukia """ class NSBindingsTest(XCAPTest): def test_ns_bindings(self): self.put('resource-lists', resource_list_xml) r = self.get('resource-lists', '/resource-lists/list[@name="friends"]/namespace::*') self.assertHeader(r, 'ETag') self.assertHeader(r, 'Content-type', 'application/xcap-ns+xml') # add expected content if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_pidf.py b/test/test_pidf.py index 0c01ee7..13d97a3 100755 --- a/test/test_pidf.py +++ b/test/test_pidf.py @@ -1,31 +1,31 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * pidf_xml = """ open """ class PIDFTest(XCAPTest): def test_pidf_manipulation(self): self.getputdelete('pidf-manipulation', pidf_xml, 'application/pidf+xml') if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_presrules.py b/test/test_presrules.py index 655becb..1f1040a 100755 --- a/test/test_presrules.py +++ b/test/test_presrules.py @@ -1,42 +1,42 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * pres_rules_xml = """ allow """ class PresenceRulesTest(XCAPTest): def test_pidf_manipulation(self): self.getputdelete('pres-rules', pres_rules_xml, 'application/auth-policy+xml') if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_resourcelists.py b/test/test_resourcelists.py index 6409efc..3f23856 100755 --- a/test/test_resourcelists.py +++ b/test/test_resourcelists.py @@ -1,131 +1,131 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * resource_lists_xml = """ Bill Doe Close Friends Joe Smith Nancy Gross Marketing """ resource_lists_xml_badformed = """ Bill Doe Close Friends Joe Smith Nancy Gross Marketing """ # well-formed, but fails to meet constraints resource_lists_xml_non_unique_list = """ Bill Doe Close Friends Joe Smith Nancy Gross Marketing """ resource_lists_xml_baduri = """ Bill Doe Close Friends Joe Smith Nancy Gross Marketing """ class DocumentTest(XCAPTest): def test_operations1(self): self.getputdelete('resource-lists', resource_lists_xml, 'application/resource-lists+xml') def test_operations2(self): self.getputdelete('resource-lists', resource_lists_xml.replace('UTF-8', 'utf-8'), 'application/resource-lists+xml') def test_operations3(self): r = self.put_rejected('resource-lists', resource_lists_xml_badformed) self.assertInBody(r, ' if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_rlsservices.py b/test/test_rlsservices.py index 4cdc452..b769df1 100755 --- a/test/test_rlsservices.py +++ b/test/test_rlsservices.py @@ -1,124 +1,124 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * rls_services_xml = """ http://xcap.example.com/resource-lists/users/sip:joe@example.com/index/~~/resource-lists/list%5b@name=%22l1%22%5d presence presence """ rls_services_xml_badformed = """ http://xcap.example.com/resource-lists/users/sip:joe@example.com/index/~~/resource-lists/list%5b@name=%22l1%22%5d presence presence """ # resource-lists constraints should be checked as well rls_services_xml_non_unique_list = """ http://xcap.example.com/resource-lists/users/sip:joe@example.com/index/~~/resource-lists/list%5b@name=%22l1%22%5d presence presence """ # this one is actually caught by schema validation, not by code rls_services_xml_non_unique_service = """ http://xcap.example.com/resource-lists/users/sip:joe@example.com/index/~~/resource-lists/list%5b@name=%22l1%22%5d presence presence """ # check for that service uniqueness is enforced across different users # check index class DocumentTest(XCAPTest): def test_operations1(self): self.getputdelete('rls-services', rls_services_xml, 'application/rls-services+xml') def test_operations2(self): self.put_rejected('rls-services', rls_services_xml_badformed) def test_operations3(self): self.put_rejected('rls-services', rls_services_xml_non_unique_list) def test_operations4(self): self.put_rejected('rls-services', rls_services_xml_non_unique_service) #self.account = 'test2@example.com' #self.delete_resource('rls-services') #self.assertStatus([200, 404]) ## we aint doing that ## rejected because the other user has the services with the same name ##self.put_rejected('rls-services', rls_services_xml) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_watchers.py b/test/test_watchers.py index 07f7409..8ca4dc5 100755 --- a/test/test_watchers.py +++ b/test/test_watchers.py @@ -1,33 +1,33 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import XCAPTest, runSuiteFromModule watchers = """ """ class Test(XCAPTest): def test_get(self): self.get('org.openxcap.watchers') self.get('org.openxcap.watchers', '/watchers') self.get('org.openxcap.watchers', globaltree=True, status=404) self.get('org.openxcap.watchers', '/watchers', globaltree=True, status=404) # def test_put_not_allowed(self): # self.put('watchers', watchers, status=405) # self.put('watchers', watchers, '/watchers', status=405) # self.put('watchers', watchers, globaltree=True, status=405) # self.put('watchers', watchers, '/watchers', globaltree=True, status=405) # def test_delete_not_allowed(self): # self.delete('watchers', status=405) # self.delete('watchers', '/watchers', status=405) # self.delete('watchers', globaltree=True, status=405) # self.delete('watchers', '/watchers', globaltree=True, status=405) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_xcap_caps.py b/test/test_xcap_caps.py index bb4d85e..ced3c0d 100755 --- a/test/test_xcap_caps.py +++ b/test/test_xcap_caps.py @@ -1,24 +1,24 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # from .common import * class XCAPCaps(XCAPTest): def test_schema(self): r = self.get_global('xcap-caps') validate_xcapcaps_schema(r.body) # TODO: auto check schema for every get schema = load_schema('xcap-caps.xsd') def validate_xcapcaps_schema(document): xml = validate(document, schema) assert xml.find('{urn:ietf:params:xml:ns:xcap-caps}auids') is not None assert xml.find('{urn:ietf:params:xml:ns:xcap-caps}extensions') is not None assert xml.find('{urn:ietf:params:xml:ns:xcap-caps}namespaces') is not None if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_xpath.py b/test/test_xpath.py index 4df39b7..d53ede5 100755 --- a/test/test_xpath.py +++ b/test/test_xpath.py @@ -1,85 +1,85 @@ -#!/usr/bin/python2 +#!/usr/bin/env python3 -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # # All tests where ported from a test suite created by # Inaki Baz Castillo import unittest import sys sys.path = ['../..'] + sys.path import xcap.xpath from xcap.uri import XCAPUri default_namespaces = {'org.openxcap.watchers': 'http://openxcap.org/ns/watchers', 'org.openmobilealliance.pres-rules': 'urn:ietf:params:xml:ns:pres-rules', 'rls-services': 'urn:ietf:params:xml:ns:rls-services', 'resource-lists': 'urn:ietf:params:xml:ns:resource-lists', 'xcap-caps': 'urn:ietf:params:xml:ns:xcap-caps', 'org.openxcap.dialog-rules': 'http://openxcap.org/ns/dialog-rules', 'test-app': 'test-app', 'org.openmobilealliance.pres-content': 'urn:oma:xml:prs:pres-content', 'pidf-manipulation': 'urn:ietf:params:xml:ns:pidf', 'pres-rules': 'urn:ietf:params:xml:ns:pres-rules', 'org.openmobilealliance.xcap-directory': 'urn:oma:xml:xdm:xcap-directory'} class XPathTest(unittest.TestCase): def test_xpath1_valid(self): selector = '/pres-rules/users/sip:%61lice@domain.org/Mis%20Documentos/index?xmlns(pr=urn:ietf:params:xml:ns:pres-rules)xmlns(cp=urn:ietf:params:xml:ns:common-policy)' u = XCAPUri('https://xcap.sipthor.net/xcap-root', selector, default_namespaces) self.assertEqual(str(u), 'https://xcap.sipthor.net/xcap-root/pres-rules/users/sip:alice@domain.org/Mis Documentos/index?xmlns(pr=urn:ietf:params:xml:ns:pres-rules)xmlns(cp=urn:ietf:params:xml:ns:common-policy)') def test_xpath2_invalid(self): selector = '' self.assertRaises(xcap.xpath.DocumentSelectorError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath3_invalid(self): selector = '/pres-rules/global/mydoc/~~/' self.assertRaises(xcap.xpath.NodeParsingError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath4_invalid(self): selector = 'pres-rules/global/mydoc' self.assertRaises(xcap.xpath.DocumentSelectorError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath5_invalid(self): selector = '/pres-rules/lalala/presrules' self.assertRaises(xcap.xpath.DocumentSelectorError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath6_invalid(self): selector = '/pres-rules/users/sip:alice@domain.org/' self.assertRaises(xcap.xpath.DocumentSelectorError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath7_invalid(self): selector = '/pres-rules/users/sip:alice@domain.org' self.assertRaises(xcap.xpath.DocumentSelectorError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath8_invalid(self): selector = '/pres-rules/users/sip:alice@domain.org/My%20presrules/~~/cp:ruleset/cp:rule%5b@id=%22pres_whitelist%22%5d/cp:conditions/cp:identity/cp:one%5b@id=%22sip:alice@example.org%22%5d' self.assertRaises(xcap.xpath.NodeParsingError, XCAPUri, 'https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath9_valid(self): selector = '/pres-rules/users/sip:alice@domain.org/My%20presrules/~~/cp:ruleset/cp:rule%5b@id=%22pres_whitelist%22%5d/cp:conditions/cp:identity/cp:one%5b@id=%22sip:alice@example.org%22%5d?xmlns(cp=urn:ietf:params:xml:ns:common-policy)' u = XCAPUri('https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath10_valid(self): selector = '/pres-rules/users/sip:alice@domain.org/My%20presrules/~~/cp:ruleset/cp:rule%5b@id=%22pres_whitelist%22%5d/cp:conditions/cp:identity/cp:one%5b@id=%22sip:alice@example.org%22%5d?xmlns(cp=urn:ietf:params:xml:ns:common-policy)' u = XCAPUri('https://xcap.sipthor.net/xcap-root', selector, default_namespaces) def test_xpath11_valid(self): selector = '/pres-rules/users/sip:alice@domain.org/presrules/~~/cp:ruleset/cp:rule%5b@id=%22pres_whitelist%22%5d/cp:conditions/cp:identity/@name?xmlns(cp=urn:ietf:params:xml:ns:common-policy)' u = XCAPUri('https://xcap.sipthor.net/xcap-root', selector, default_namespaces) self.assertEqual(xcap.xpath.AttributeSelector, type(u.node_selector.terminal_selector)) self.assertEqual('@name', str(u.node_selector.terminal_selector)) def test_xpath12_valid(self): selector = '/pres-rules/users/sip:alice@domain.org/presrules/~~/cp:ruleset/cp:rule%5b@id=%22pres_whitelist%22%5d/cp:conditions/cp:identity/namespace::*?xmlns(cp=urn:ietf:params:xml:ns:common-policy)' u = XCAPUri('https://xcap.sipthor.net/xcap-root', selector, default_namespaces) self.assertEqual(xcap.xpath.NamespaceSelector, type(u.node_selector.terminal_selector)) if __name__ == '__main__': unittest.main() diff --git a/test/xcapclientwrap.py b/test/xcapclientwrap.py index 44d485d..bee4b11 100644 --- a/test/xcapclientwrap.py +++ b/test/xcapclientwrap.py @@ -1,137 +1,137 @@ -# Copyright (C) 2007-2010 AG-Projects. +# Copyright (C) 2007-2025 AG-Projects. # import os import re from subprocess import Popen, PIPE from xcaplib.httpclient import HTTPResponse DEBUG = 0 def make_client(options): return XCAPClient(options.xcap_root, options.sip_address, options.password) class XCAPClient(object): """Wrapper of command-line utility xcapclient. Pointless, unless you want to test xcapclient itself. """ XCAPCLIENT = '/home/denis/work/python-xcaplib/xcapclient' def __init__(self, xcap_root, sip_address, password): self.params = ['--xcap-root', xcap_root, '--sip-address', sip_address, '--password', password] def get_params(self, etag=None, globaltree=False, filename=None, headers=None): params = self.params[:] if etag is not None: params += ['--etag', etag] if globaltree: params += ['-c', 'global'] else: params += ['-c', 'users'] if filename is not None: params += ['--filename', filename] for k, v in (headers or {}).items(): if v is None: params += ['--add-header', k] else: params += ['--add-header', '%s:%s' % (k, v)] return params def request(self, method, application, input=None, node=None, **params): params = ['--app', application] + self.get_params(**params) params.append(method) if node is not None: if node[:1]!='/': node = '/' + node params.append(node) return self._run(params, input) def _get(self, application, node=None, **params): return self.request('get', application, node=node, **params) def _put(self, application, resource, node=None, **params): return self.request('put', application, input=resource, node=node, **params) def _delete(self, application, node=None, **params): return self.request('delete', application, node=node, **params) def _run(self, params, input=None): params = [self.XCAPCLIENT] + params p = Popen(params, stdin=input and PIPE, stdout=PIPE, stderr=PIPE, env=os.environ) (stdout, stderr) = p.communicate(input=input) if DEBUG: print('\n______________') print(stdout) print('--------------') print(stderr) print('^^^^^^^^^^^^^^') code, comment, etag, content_type = parse_stderr(stderr) hdrs = headers() if p.wait() == 0: if code is None: code, comment = 200, 'OK' else: assert code is not None, repr(stderr) assert comment is not None, repr(stderr) if etag is not None: hdrs['ETag'] = etag if content_type is not None: hdrs['Content-Type'] = content_type return HTTPResponse(None, code, comment, hdrs, stdout) class headers(dict): def gettype(self): typ = self.get('Content-Type') if typ is None: return typ return typ.split(';', 1)[0] re_status_line = re.compile("^(\d\d\d) (.*?)$", re.M) re_etag = re.compile('^etag: (".*?")$', re.M | re.I) re_content_type = re.compile("^content-type: (.*?)$", re.M | re.I) def findone(re, str): m = re.findall(str) assert len(m)<=1, (m, str) if not m: return None elif len(m)==1: return m[0] def parse_stderr(stderr): """ >>> parse_stderr('''url: https://10.1.1.3/xcap-root/resource-lists/listxx ... 404 Not Found ... content-length: 121 ... ''') (404, 'Not Found', None, None) >>> parse_stderr('''url: https://10.1.1.3/xcap-root/resource-lists/users/alice@example ... etag: "5342d9c443c7fad5d76669c7253688f0" ... content-length: 1829 ... ''') (None, None, '"5342d9c443c7fad5d76669c7253688f0"', None) >>> parse_stderr('url: https://10.1.1.3/xcap-root/xcap-caps/global/index\\netag: "6fc08e7c18116bb145c7052fc9a2d6bf"\\ncontent-length: 826\\n\\n') (None, None, '"6fc08e7c18116bb145c7052fc9a2d6bf"', None) """ m = findone(re_status_line, stderr) if m is None: code, comment = None, None else: code, comment = m code = int(code) etag = findone(re_etag, stderr) content_type = findone(re_content_type, stderr) return code, comment, etag, content_type if __name__=='__main__': import doctest doctest.testmod()