diff --git a/openxcap b/openxcap old mode 100644 new mode 100755 index f514fa7..0e66e95 --- a/openxcap +++ b/openxcap @@ -1,83 +1,83 @@ -#!/usr/bin/env python +#!/usr/bin/python2 if __name__ == '__main__': import sys import xcap from application import log from application.process import process, ProcessError from argparse import ArgumentParser name = 'openxcap' fullname = 'OpenXCAP' description = 'An open source XCAP Server' process.configuration.user_directory = None process.configuration.subdirectory = 'openxcap' parser = ArgumentParser(usage='%(prog)s [options]') parser.add_argument('--version', action='version', version='%(prog)s {}'.format(xcap.__version__)) parser.add_argument('--systemd', action='store_true', help='run as a systemd simple service and log to journal') parser.add_argument('--no-fork', action='store_false', dest='fork', help='run in the foreground and log to the terminal') parser.add_argument('--config-dir', dest='config_directory', default=None, help='the configuration directory ({})'.format(process.configuration.system_directory), metavar='PATH') parser.add_argument('--runtime-dir', dest='runtime_directory', default=None, help='the runtime directory ({})'.format(process.runtime.directory), metavar='PATH') parser.add_argument('--debug', action='store_true', help='enable verbose logging') parser.add_argument('--debug-memory', action='store_true', help='enable memory debugging') options = parser.parse_args() try: from xcap.logutil import web_logger except Exception as e: log.critical('Cannot start %s: %s' % (fullname, e)) if not isinstance(e, (RuntimeError, OSError, IOError)): log.exception() sys.exit(1) if web_logger.filename is None: # access log is reported along with the rest of the applications's logging log.Formatter.prefix_format = '{record.levelname:<8s} [{record.name}] ' else: log.Formatter.prefix_format = '{record.levelname:<8s} ' if options.config_directory is not None: process.configuration.local_directory = options.config_directory if options.runtime_directory is not None: process.runtime.directory = options.runtime_directory if options.systemd: from systemd.journal import JournalHandler log.set_handler(JournalHandler(SYSLOG_IDENTIFIER=name)) log.capture_output() elif options.fork: try: process.daemonize(pidfile='{}.pid'.format(name)) except ProcessError as e: log.critical('Cannot start %s: %s' % (fullname, e)) sys.exit(1) log.use_syslog(name) log.info('Starting %s %s' % (fullname, xcap.__version__)) try: process.wait_for_network(wait_time=10, wait_message='Waiting for network to become available...') except KeyboardInterrupt: sys.exit(0) except RuntimeError as e: log.critical('Cannot start %s: %s' % (fullname, e)) sys.exit(1) try: from xcap.server import XCAPServer if options.debug: log.level.current = log.level.DEBUG if options.debug_memory: from application.debug.memory import memory_dump server = XCAPServer() server.start() except Exception as e: log.critical('Failed to create %s: %s' % (fullname, e)) if type(e) is not RuntimeError: log.exception() sys.exit(1) if options.debug_memory: memory_dump() diff --git a/scripts/add-openxcap-user.py b/scripts/add-openxcap-user.py old mode 100644 new mode 100755 index a232520..ef53eb3 --- a/scripts/add-openxcap-user.py +++ b/scripts/add-openxcap-user.py @@ -1,17 +1,19 @@ -#!/usr/bin/env python +#!/usr/bin/python2 + """This is an example that shows how a new user could be added to `subscriber' table. It does NOT actually create a new record in the database. """ + import sys from hashlib import md5 print __doc__ try: username, domain, password = sys.argv[1:] except ValueError: sys.exit('USAGE: %s username domain password' % sys.argv[0]) hash = md5(":".join([username, domain, password])).hexdigest() query = """INSERT INTO subscriber (username, domain, password, ha1) VALUES ("%(username)s", "%(domain)s", "%(password)s", "%(hash)s");""" % locals() print query diff --git a/setup.py b/setup.py old mode 100644 new mode 100755 index e7a061a..1425d33 --- a/setup.py +++ b/setup.py @@ -1,48 +1,48 @@ -#!/usr/bin/env python +#!/usr/bin/python2 import os import xcap from distutils.core import setup long_description = """XCAP protocol allows a client to read, write, and modify application configuration data stored in XML format on a server. XCAP maps XML document sub-trees and element attributes to HTTP URIs, so that these components can be directly accessed by HTTP. An XCAP server is used by the XCAP clients to store data like Presence policy in combination with a SIP Presence server that supports PUBLISH/SUBSCRIBE/NOTIFY methods to provide a complete [http://www.tech-invite.com/Ti-sip-WGs.html#wg-simple SIP SIMPLE] server solution.""" def find_packages(toplevel): return [directory.replace(os.path.sep, '.') for directory, subdirs, files in os.walk(toplevel) if '__init__.py' in files] setup( name='openxcap', version=xcap.__version__, description='XCAP server', long_description=long_description, url='http://openxcap.org/', author='AG Projects', author_email='support@ag-projects.com', license='GPL', platforms=['Platform Independent'], classifiers=[ 'Development Status :: 5 - Production/Stable', 'Intended Audience :: Service Providers', 'License :: OSI Approved :: GNU General Public License (GPL)', 'Operating System :: POSIX :: Linux', 'Programming Language :: Python', ], packages=find_packages('xcap'), package_data={'xcap.appusage': ['xml-schemas/*']}, data_files=[('/etc/openxcap', ['config.ini.sample']), ('/etc/openxcap/tls', ['tls/README'])], scripts=['openxcap'] ) diff --git a/test/disabled_test_subscribediff.py b/test/disabled_test_subscribediff.py old mode 100644 new mode 100755 index 13cad05..60448c0 --- a/test/disabled_test_subscribediff.py +++ b/test/disabled_test_subscribediff.py @@ -1,201 +1,201 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 AG-Projects. # """This test will 1) SUBSCRIBE to xcap-diff event 2) read NOTIFY if any 3) delete the document, just in case 3) put a new document, remember ETag 4) read NOTIFY with that document url and new_etag=ETag 5) update the document, remember ETag 6) read NOTIFY with that document url, new_etag=ETag, old_etag=previous ETag 7) delete the document 8) read NOTIFY with that document url, old_etag=ETag, new_etag=Empty """ import re import time from Queue import Queue, Empty from optparse import OptionValueError from common import * from pypjua import * from xcap.xcapdiff import xml_document, xml_xcapdiff expires=20 event='xcap-diff' content_type='application/xcap-diff+xml' resource = 'resource-lists' body = """ 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.failUnless(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 old mode 100644 new mode 100755 index b5c753b..9c2bb79 --- a/test/test.py +++ b/test/test.py @@ -1,64 +1,64 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 AG-Projects. # import sys import os import traceback 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 old mode 100644 new mode 100755 index 88ba835..a6d29ee --- a/test/test_attribute.py +++ b/test/test_attribute.py @@ -1,69 +1,69 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 9aee0d4..40fefae --- a/test/test_auth.py +++ b/test/test_auth.py @@ -1,44 +1,44 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 88f126b..fc6c9a0 --- a/test/test_element.py +++ b/test/test_element.py @@ -1,112 +1,112 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 1682aa8..0f690a8 --- a/test/test_element_put.py +++ b/test/test_element_put.py @@ -1,177 +1,177 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 7598bdb..5025401 --- a/test/test_errors.py +++ b/test/test_errors.py @@ -1,73 +1,73 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 AG-Projects. # import common as c from urlparse 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, `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 old mode 100644 new mode 100755 index f678218..45be536 --- a/test/test_etags.py +++ b/test/test_etags.py @@ -1,113 +1,113 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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, `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 old mode 100644 new mode 100755 index e29668e..135006a --- a/test/test_etags2.py +++ b/test/test_etags2.py @@ -1,53 +1,53 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 610e7ef..91c9dc2 --- a/test/test_fragment.py +++ b/test/test_fragment.py @@ -1,50 +1,50 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 AG-Projects. # 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 old mode 100644 new mode 100755 index 43dce57..a3f9348 --- a/test/test_global.py +++ b/test/test_global.py @@ -1,32 +1,32 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index ade4aa6..84adfd8 --- a/test/test_nsbindings.py +++ b/test/test_nsbindings.py @@ -1,33 +1,33 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index e4194f6..99e7996 --- a/test/test_pidf.py +++ b/test/test_pidf.py @@ -1,31 +1,31 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index a28079b..9cef542 --- a/test/test_presrules.py +++ b/test/test_presrules.py @@ -1,42 +1,42 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 00f0a8c..8f42f0f --- a/test/test_resourcelists.py +++ b/test/test_resourcelists.py @@ -1,131 +1,131 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index c6041f5..a69682f --- a/test/test_rlsservices.py +++ b/test/test_rlsservices.py @@ -1,124 +1,124 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 2cb572f..a85fe3b --- a/test/test_watchers.py +++ b/test/test_watchers.py @@ -1,33 +1,33 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index b725320..f53f6f0 --- a/test/test_xcap_caps.py +++ b/test/test_xcap_caps.py @@ -1,24 +1,24 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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 old mode 100644 new mode 100755 index 83f29ee..4df39b7 --- a/test/test_xpath.py +++ b/test/test_xpath.py @@ -1,85 +1,85 @@ -#!/usr/bin/env python +#!/usr/bin/python2 # Copyright (C) 2007-2010 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()