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()