summaryrefslogtreecommitdiffstats
path: root/python/tests/abstractweb.py
diff options
context:
space:
mode:
Diffstat (limited to 'python/tests/abstractweb.py')
-rw-r--r--python/tests/abstractweb.py329
1 files changed, 0 insertions, 329 deletions
diff --git a/python/tests/abstractweb.py b/python/tests/abstractweb.py
deleted file mode 100644
index 94fb644d..00000000
--- a/python/tests/abstractweb.py
+++ /dev/null
@@ -1,329 +0,0 @@
-# -*- coding: UTF-8 -*-
-
-
-# Abstract web classes for HTTP clients and servers or simulators
-# By: Frederic Peters <fpeters@entrouvert.com>
-# Emmanuel Raviart <eraviart@entrouvert.com>
-#
-# Copyright (C) 2004 Entr'ouvert
-# http://www.entrouvert.org
-#
-# This program is free software; you can redistribute it and/or
-# modify it under the terms of the GNU General Public License
-# as published by the Free Software Foundation; either version 2
-# of the License, or (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
-
-
-"""Abstract web classes for HTTP clients and servers or simulators"""
-
-
-class HttpRequestMixin:
- body = None
- headers = None
- method = None # 'GET' or 'POST' or 'PUT' or...
- path = None
- pathAndQuery = None
- query = None
- scheme = None # 'http' or 'https'
- url = None
-
- def getFormField(self, name, default = None):
- raise NotImplementedError
-
- def getQueryBoolean(self, name, default = None):
- fieldValue = self.getQueryField(name, 'none')
- if fieldValue == 'none':
- return default
- else:
- return fieldValue.lower not in ('', '0', 'false')
-
- def getQueryField(self, name, default = None):
- if self.query:
- for field in self.query.split('&'):
- fieldName, fieldValue = field.split('=')
- if name == fieldName:
- return fieldValue
- return default
-
- def hasFormField(self, name):
- raise NotImplementedError
-
- def hasQueryField(self, name):
- if self.query:
- for field in self.query.split('&'):
- fieldName, fieldValue = field.split('=')
- if name == fieldName:
- return True
- return False
-
-
-class FunctionHttpRequest(HttpRequestMixin, object):
- method = 'GET'
- previousHttpRequest = None
- queryFields = None
-
- def __init__(self, previousHttpRequest, **queryFields):
- self.previousHttpRequest = previousHttpRequest
- self.queryFields = queryFields
-
- def getFormField(self, name, default = None):
- return default
-
- def getHeaders(self):
- return self.previousHttpRequest.headers
-
- def getQueryField(self, name, default = None):
- return self.queryFields.get(name, default)
-
- def hasFormField(self, name):
- return False
-
- def hasQueryField(self, name):
- return name in self.queryFields
-
- headers = property(getHeaders)
-
-
-class HttpResponseMixin:
- body = None
- defaultStatusMessages = {
- '100': 'Continue',
- '101': 'Switching Protocols',
- '200': 'OK',
- '201': 'Created',
- '202': 'Accepted',
- '203': 'Non-Authoritative Information',
- '204': 'No Content',
- '205': 'Reset Content',
- '206': 'Partial Content',
- '300': 'Multiple Choices',
- '301': 'Moved Permanently',
- '302': 'Found',
- '303': 'See Other',
- '304': 'Not Modified',
- '305': 'Use Proxy',
- '307': 'Temporary Redirect',
- '400': 'Bad Request',
- '401': 'Unauthorized',
- '402': 'Payment Required',
- '403': 'Forbidden',
- '404': 'Not Found',
- '405': 'Method Not Allowed',
- '406': 'Not Acceptable',
- '407': 'Proxy Authentication Required',
- '408': 'Request Time-out',
- '409': 'Conflict',
- '410': 'Gone',
- '411': 'Length Required',
- '412': 'Precondition Failed',
- '413': 'Request Entity Too Large',
- '414': 'Request-URI Too Large',
- '415': 'Unsupported Media Type',
- '416': 'Requested range not satisfiable',
- '417': 'Expectation Failed',
- '500': 'Internal Server Error',
- '501': 'Not Implemented',
- '502': 'Bad Gateway',
- '503': 'Service Unavailable',
- '504': 'Gateway Time-out',
- '505': 'HTTP Version not supported',
- }
- headers = None
- statusCode = None # 200 or...
- statusMessage = None
-
- def __init__(self, httpRequestHandler, statusCode, statusMessage = None, headers = None,
- body = None):
- self.statusCode = statusCode
- if statusMessage:
- self.statusMessage = statusMessage
- else:
- self.statusMessage = self.defaultStatusMessages.get(statusCode)
- httpResponseHeaders = httpRequestHandler.site.httpResponseHeaders
- if headers:
- httpResponseHeaders = httpResponseHeaders.copy()
- for name, value in headers.iteritems():
- httpResponseHeaders[name] = value
- if httpResponseHeaders:
- self.headers = httpResponseHeaders
- if body:
- self.body = body
-
- def send(self, httpRequestHandler):
- raise NotImplementedError
-
-
-class HttpRequestHandlerMixin:
- httpRequest = None
- HttpResponse = None # Class
- httpResponse = None
- session = None
- user = None
- site = None # The virtual host
-
- def createSession(self):
- session = self.site.newSession()
- self.session = session
- return session
-
- def createUser(self):
- user = self.site.newUser()
- self.user = user
- return user
-
- def respond(self, statusCode = 200, statusMessage = None, headers = None, body = None):
- self.httpResponse = self.HttpResponse(
- self, statusCode, statusMessage = statusMessage, headers = headers, body = body)
-
- # Session and user must be saved before responding. Otherwise, when the server is
- # multitasked or multithreaded, it may receive a new HTTP request before the session is
- # saved.
- if self.session is not None and self.session.isDirty:
- self.session.save()
- if self.user is not None and self.user.isDirty:
- self.user.save()
-
- return self.httpResponse.send(self)
-
- def respondRedirectTemporarily(self, url):
- raise NotImplementedError
-
-
-class WebClientMixin:
- httpRequestHeaders = {
- 'User-Agent': 'LassoSimulator/0.0.0',
- 'Accept': 'text/xml,application/xml,application/xhtml+xml,text/html',
- }
-
- def __init__(self):
- pass
-
-
-class WebSessionMixin(WebClientMixin):
- isDirty = True
- token = None
- userId = None # ID of logged user
-
- def __init__(self, token):
- WebClientMixin.__init__(self)
- self.token = token
-
- def getSimpleLabel(self):
- return self.token
-
- def save(self):
- pass
-
- simpleLabel = property(getSimpleLabel)
-
-
-class WebSiteMixin:
- FunctionHttpRequest = FunctionHttpRequest # Class
- httpResponseHeaders = {
- 'Server': 'Lasso Simulator Web Server',
- }
- lastSessionToken = 0
- lastUserId = 0
- users = None
- sessions = None
- WebSession = None # Class
- WebUser = None # Class
-
- def __init__(self):
- self.users = {}
- self.sessions = {}
-
- def authenticateX509User(self, clientCertificate):
- # We should check certificate (for example clientCertificate.get_serial_number()
- # and return the user if one matches, or None otherwise.
- return None
-
- def authenticateLoginPasswordUser(self, login, password):
- # We should check login & password and return the user if one matches or None otherwise.
- return None
-
- def callHttpFunction(self, function, httpRequestHandler, **queryFields):
- httpRequestHandler.httpRequest = self.FunctionHttpRequest(
- httpRequestHandler.httpRequest, **queryFields)
- try:
- result = function(httpRequestHandler)
- finally:
- httpRequestHandler.httpRequest = httpRequestHandler.httpRequest.previousHttpRequest
- return result
-
- def handleHttpRequestHandler(self, httpRequestHandler):
- methodName = httpRequestHandler.httpRequest.path.replace('/', '')
- try:
- method = getattr(self, methodName)
- except AttributeError:
- return httpRequestHandler.respond(
- 404, 'Path "%s" Not Found.' % httpRequestHandler.httpRequest.path)
- return method(httpRequestHandler)
-
- def login(self, handler):
- # On most site (except Liberty service providers), authentication is done locally.
- return self.callHttpFunction(self.login_local, handler)
-
- def login_done(self, handler, userAuthenticated, authenticationMethod):
- if not userAuthenticated:
- return self.login_failed(handler)
- return handler.respond(
- 200, headers = {'Content-Type': 'text/plain'},
- body = 'Login terminated:\n userAuthenticated = %s\n authenticationMethod = %s' % (
- userAuthenticated, authenticationMethod))
-
- def login_failed(self, handler):
- return handler.respond(401, 'Access Unauthorized: User has no account.')
-
- def login_local(self, handler):
- # Note: Once local login is done, the HTTP function login_done must be called.
- raise NotImplementedError
-
- def newSession(self):
- self.lastSessionToken += 1
- sessionToken = str(self.lastSessionToken)
- session = self.WebSession(sessionToken)
- self.sessions[sessionToken] = session
- return session
-
- def newUser(self, name = None):
- if name is None:
- self.lastUserId += 1
- userId = str(self.lastUserId)
- else:
- userId = name
- user = self.WebUser(userId, name = name)
- self.users[userId] = user
- return user
-
-
-class WebUserMixin:
- isDirty = True
- name = None
- sessionToken = None
- uniqueId = None
-
- def __init__(self, uniqueId, name = None):
- self.uniqueId = uniqueId
- if name:
- self.name = name
-
- def getSimpleLabel(self):
- if self.name:
- return self.name
- else:
- return 'Anonymous User #%s' % self.uniqueId
-
- def save(self):
- pass
-
- simpleLabel = property(getSimpleLabel)
generated by cgit (git 2.34.1) at 2026-01-20 18:29:39 +0000