diff --git a/test/common.py b/test/common.py index 3237211..dca7120 100644 --- a/test/common.py +++ b/test/common.py @@ -1,401 +1,407 @@ # 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 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): + self.options.timeout = None 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') + try: + if status is not None: + if isinstance(status, int): + if 200 <= status <= 299: + self.assertHeader(r, 'ETag') + except TypeError: + print("Error with status:", status) 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) + r_put = client._put(application, resource.encode(), 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') + if r.status==409 or r.headers.get_content_type()=='application/xcap-error+xml': + self.assertEqual(r.headers.get_content_type(), 'application/xcap-error+xml') xml = validate_xcaps_error(r.body) if ' 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') + r = self.client._put('resource-lists', 'coworkers'.encode(), '/resource-lists/list[@name="friends"]/@name') self.assertStatus(r, 200) if __name__ == '__main__': runSuiteFromModule() diff --git a/test/test_element.py b/test/test_element.py index 9b80881..012456c 100755 --- a/test/test_element.py +++ b/test/test_element.py @@ -1,112 +1,112 @@ #!/usr/bin/env python3 # 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) + headers={'Content-Type': ''},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_errors.py b/test/test_errors.py index 3816a69..40f80f4 100755 --- a/test/test_errors.py +++ b/test/test_errors.py @@ -1,73 +1,73 @@ #!/usr/bin/env python3 # Copyright (C) 2007-2025 AG-Projects. # 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) + s.send(data.encode()) 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) + assert '400 Bad Request' in response.decode(), 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) + r = self.client.client.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', '') + r = self.client.client.request('POST', '') self.assertEqual(r.status, 405) - r = self.client.con.request('XXX', '') + r = self.client.client.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_resourcelists.py b/test/test_resourcelists.py index 28a231a..7c671f4 100755 --- a/test/test_resourcelists.py +++ b/test/test_resourcelists.py @@ -1,131 +1,131 @@ #!/usr/bin/env python3 # 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()