backend.tracim_backend.lib.webdav.middlewares.TracimUserSession.__call__() - Code Metrics - tracim/tracim_v2 - Measure and Improve Code Quality continuously with Scrutinizer

TracimUserSession.call() A
last analyzed 2018-11-14 14:00 UTC

↳ Parent: backend.tracim_backend.lib.webdav.middlewares

Complexity

Conditions

Size

Total Lines	7
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	7
rs	10
c	0
b	0
f	0
cc	1
nop	3

import os
import sys
import threading
import time
from datetime import datetime
from xml.etree import ElementTree

import transaction
import yaml
from pyramid.paster import get_appsettings
from wsgidav import util, compat
from wsgidav.middleware import BaseMiddleware

from tracim_backend import CFG
from tracim_backend.lib.core.user import UserApi
from tracim_backend.models import get_engine, get_session_factory, get_tm_session


class TracimWsgiDavDebugFilter(BaseMiddleware):
    """
    COPY PASTE OF wsgidav.debug_filter.WsgiDavDebugFilter
    WITH ADD OF DUMP RESPONSE & REQUEST
    """
    def __init__(self, application, config):
        self._application = application
        self._config = config
        #        self.out = sys.stderr
        self.out = sys.stdout
        self.passedLitmus = {}
        # These methods boost verbose=2 to verbose=3
        self.debug_methods = config.get("debug_methods", [])
        # Litmus tests containing these string boost verbose=2 to verbose=3
        self.debug_litmus = config.get("debug_litmus", [])
        # Exit server, as soon as this litmus test has finished
        self.break_after_litmus = [
            #                                   "locks: 15",
        ]

        self.last_request_time = '__NOT_SET__'

        # We disable request content dump for moment
        # if self._config.get('dump_requests'):
        #     # Monkey patching
        #     old_parseXmlBody = util.parseXmlBody
        #     def new_parseXmlBody(environ, allowEmpty=False):
        #         xml = old_parseXmlBody(environ, allowEmpty)
        #         self._dump_request(environ, xml)
        #         return xml
        #     util.parseXmlBody = new_parseXmlBody

    def __call__(self, environ, start_response):
        """"""
        #        srvcfg = environ["wsgidav.config"]
        verbose = self._config.get("verbose", 2)
        self.last_request_time = '{0}_{1}'.format(
            datetime.utcnow().strftime('%Y-%m-%d_%H-%M-%S'),
            int(round(time.time() * 1000)),
        )

        method = environ["REQUEST_METHOD"]

        debugBreak = False
        dumpRequest = False
        dumpResponse = False

        if verbose >= 3 or self._config.get("dump_requests"):
            dumpRequest = dumpResponse = True

        # Process URL commands
        if "dump_storage" in environ.get("QUERY_STRING"):
            dav = environ.get("wsgidav.provider")
            if dav.lockManager:
                dav.lockManager._dump()
            if dav.propManager:
                dav.propManager._dump()

        # Turn on max. debugging for selected litmus tests
        litmusTag = environ.get("HTTP_X_LITMUS",
                                environ.get("HTTP_X_LITMUS_SECOND"))
        if litmusTag and verbose >= 2:
            print("----\nRunning litmus test '%s'..." % litmusTag,
                  file=self.out)
            for litmusSubstring in self.debug_litmus:
                if litmusSubstring in litmusTag:
                    verbose = 3
                    debugBreak = True
                    dumpRequest = True
                    dumpResponse = True
                    break
            for litmusSubstring in self.break_after_litmus:
                if litmusSubstring in self.passedLitmus and litmusSubstring not in litmusTag:
                    print(" *** break after litmus %s" % litmusTag,
                          file=self.out)
                    sys.exit(-1)
                if litmusSubstring in litmusTag:
                    self.passedLitmus[litmusSubstring] = True

        # Turn on max. debugging for selected request methods
        if verbose >= 2 and method in self.debug_methods:
            verbose = 3
            debugBreak = True
            dumpRequest = True
            dumpResponse = True

        # Set debug options to environment
        environ["wsgidav.verbose"] = verbose
        #        environ["wsgidav.debug_methods"] = self.debug_methods
        environ["wsgidav.debug_break"] = debugBreak
        environ["wsgidav.dump_request_body"] = dumpRequest
        environ["wsgidav.dump_response_body"] = dumpResponse

        # Dump request headers
        if dumpRequest:
            print("<%s> --- %s Request ---" % (
            threading.currentThread().ident, method), file=self.out)
            for k, v in environ.items():
                if k == k.upper():
                    print("%20s: '%s'" % (k, v), file=self.out)
            print("\n", file=self.out)
            self._dump_request(environ, xml=None)

        # Intercept start_response
        #
        sub_app_start_response = util.SubAppStartResponse()

        nbytes = 0
        first_yield = True
        app_iter = self._application(environ, sub_app_start_response)

        for v in app_iter:
            # Start response (the first time)
            if first_yield:
                # Success!
                start_response(sub_app_start_response.status,
                               sub_app_start_response.response_headers,
                               sub_app_start_response.exc_info)

            # Dump response headers
            if first_yield and dumpResponse:
                print("<%s> --- %s Response(%s): ---" % (
                threading.currentThread().ident,
                method,
                sub_app_start_response.status),
                      file=self.out)
                headersdict = dict(sub_app_start_response.response_headers)
                for envitem in headersdict.keys():
                    print("%s: %s" % (envitem, repr(headersdict[envitem])),
                          file=self.out)
                print("", file=self.out)

            # Check, if response is a binary string, otherwise we probably have
            # calculated a wrong content-length
            assert compat.is_bytes(v), v

            # Dump response body
            drb = environ.get("wsgidav.dump_response_body")
            if compat.is_basestring(drb):
                # Middleware provided a formatted body representation
                print(drb, file=self.out)
            elif drb is True:
                # Else dump what we get, (except for long GET responses)
                if method == "GET":
                    if first_yield:
                        print(v[:50], "...", file=self.out)
                elif len(v) > 0:
                    print(v, file=self.out)

            if dumpResponse:
                self._dump_response(sub_app_start_response, drb)

            drb = environ["wsgidav.dump_response_body"] = None

            nbytes += len(v)
            first_yield = False
            yield v
        if hasattr(app_iter, "close"):
            app_iter.close()

        # Start response (if it hasn't been done yet)
        if first_yield:
            # Success!
            start_response(sub_app_start_response.status,
                           sub_app_start_response.response_headers,
                           sub_app_start_response.exc_info)

        if dumpResponse:
            print("\n<%s> --- End of %s Response (%i bytes) ---" % (
            threading.currentThread().ident, method, nbytes), file=self.out)
        return

    def _dump_response(self, sub_app_start_response, drb):
        dump_to_path = self._config.get(
            'dump_requests_path',
            '/tmp/wsgidav_dumps',
        )
        os.makedirs(dump_to_path, exist_ok=True)
        dump_file = '{0}/{1}_RESPONSE_{2}.yml'.format(
            dump_to_path,
            self.last_request_time,
            sub_app_start_response.status[0:3],
        )
        with open(dump_file, 'w+') as f:
            dump_content = dict()
            headers = {}
            for header_tuple in sub_app_start_response.response_headers:
                headers[header_tuple[0]] = header_tuple[1]
            dump_content['headers'] = headers
            if isinstance(drb, str):
                dump_content['content'] = drb.replace('PROPFIND XML response body:\n', '')

            f.write(yaml.dump(dump_content, default_flow_style=False))

    def _dump_request(self, environ, xml):
        dump_to_path = self._config.get(
            'dump_requests_path',
            '/tmp/wsgidav_dumps',
        )
        os.makedirs(dump_to_path, exist_ok=True)
        dump_file = '{0}/{1}_REQUEST_{2}.yml'.format(
            dump_to_path,
            self.last_request_time,
            environ['REQUEST_METHOD'],
        )
        with open(dump_file, 'w+') as f:
            dump_content = dict()
            dump_content['path'] = environ.get('PATH_INFO', '')
            dump_content['Authorization'] = environ.get('HTTP_AUTHORIZATION', '')
            if xml:
                dump_content['content'] = ElementTree.tostring(xml, 'utf-8')

            f.write(yaml.dump(dump_content, default_flow_style=False))


class TracimEnforceHTTPS(BaseMiddleware):

    def __init__(self, application, config):
        super().__init__(application, config)
        self._application = application
        self._config = config

    def __call__(self, environ, start_response):
        # TODO - G.M - 06-03-2018 - Check protocol from http header first
        # see http://www.bortzmeyer.org/7239.html
        # if this params doesn't exist, rely on tracim config
        # from tracim.config.app_cfg import CFG
        # cfg = CFG.get_instance()
        #
        # if cfg.WEBSITE_BASE_URL.startswith('https'):
        #     environ['wsgi.url_scheme'] = 'https'
        return self._application(environ, start_response)


class TracimEnv(BaseMiddleware):

    def __init__(self, application, config):
        super().__init__(application, config)
        self._application = application
        self._config = config
        global_conf = get_appsettings(config['tracim_config']).global_conf
        local_conf = get_appsettings(config['tracim_config'], 'tracim_web')
        self.settings = global_conf
        self.settings.update(local_conf)
        self.engine = get_engine(self.settings)
        self.session_factory = get_session_factory(self.engine)
        self.app_config = CFG(self.settings)
        self.app_config.configure_filedepot()

    def __call__(self, environ, start_response):
        # TODO - G.M - 18-05-2018 - This code should not create trouble
        # with thread and database, this should be verify.
        # see https://github.com/tracim/tracim_backend/issues/62
        tm = transaction.manager
        dbsession = get_tm_session(self.session_factory, tm)
        environ['tracim_tm'] = tm
        environ['tracim_dbsession'] = dbsession
        environ['tracim_cfg'] = self.app_config
        app = self._application(environ, start_response)
        dbsession.close()
        return app


class TracimUserSession(BaseMiddleware):

    def __init__(self, application, config):
        super().__init__(application, config)
        self._application = application
        self._config = config

    def __call__(self, environ, start_response):
        environ['tracim_user'] = UserApi(
            None,
            session=environ['tracim_dbsession'],
            config=environ['tracim_cfg'],
        ).get_one_by_email(environ['http_authenticator.username'])
        return self._application(environ, start_response)


1			import os
2			import sys
3			import threading
4			import time
5			from datetime import datetime
6			from xml.etree import ElementTree
7
8			import transaction
9			import yaml
10			from pyramid.paster import get_appsettings
11			from wsgidav import util, compat
12			from wsgidav.middleware import BaseMiddleware
13
14			from tracim_backend import CFG
15			from tracim_backend.lib.core.user import UserApi
16			from tracim_backend.models import get_engine, get_session_factory, get_tm_session
17
18
19			class TracimWsgiDavDebugFilter(BaseMiddleware):
20			"""
21			COPY PASTE OF wsgidav.debug_filter.WsgiDavDebugFilter
22			WITH ADD OF DUMP RESPONSE & REQUEST
23			"""
24			def __init__(self, application, config):
25			self._application = application
26			self._config = config
27			# self.out = sys.stderr
28			self.out = sys.stdout
29			self.passedLitmus = {}
30			# These methods boost verbose=2 to verbose=3
31			self.debug_methods = config.get("debug_methods", [])
32			# Litmus tests containing these string boost verbose=2 to verbose=3
33			self.debug_litmus = config.get("debug_litmus", [])
34			# Exit server, as soon as this litmus test has finished
35			self.break_after_litmus = [
36			# "locks: 15",
37			]
38
39			self.last_request_time = '__NOT_SET__'
40
41			# We disable request content dump for moment
42			# if self._config.get('dump_requests'):
43			# # Monkey patching
44			# old_parseXmlBody = util.parseXmlBody
45			# def new_parseXmlBody(environ, allowEmpty=False):
46			# xml = old_parseXmlBody(environ, allowEmpty)
47			# self._dump_request(environ, xml)
48			# return xml
49			# util.parseXmlBody = new_parseXmlBody
50
51			def __call__(self, environ, start_response):
52			""""""
53			# srvcfg = environ["wsgidav.config"]
54			verbose = self._config.get("verbose", 2)
55			self.last_request_time = '{0}_{1}'.format(
56			datetime.utcnow().strftime('%Y-%m-%d_%H-%M-%S'),
57			int(round(time.time() * 1000)),
58			)
59
60			method = environ["REQUEST_METHOD"]
61
62			debugBreak = False
63			dumpRequest = False
64			dumpResponse = False
65
66			if verbose >= 3 or self._config.get("dump_requests"):
67			dumpRequest = dumpResponse = True
68
69			# Process URL commands
70			if "dump_storage" in environ.get("QUERY_STRING"):
71			dav = environ.get("wsgidav.provider")
72			if dav.lockManager:
73			dav.lockManager._dump()
74			if dav.propManager:
75			dav.propManager._dump()
76
77			# Turn on max. debugging for selected litmus tests
78			litmusTag = environ.get("HTTP_X_LITMUS",
79			environ.get("HTTP_X_LITMUS_SECOND"))
80			if litmusTag and verbose >= 2:
81			print("----\nRunning litmus test '%s'..." % litmusTag,
82			file=self.out)
83			for litmusSubstring in self.debug_litmus:
84			if litmusSubstring in litmusTag:
85			verbose = 3
86			debugBreak = True
87			dumpRequest = True
88			dumpResponse = True
89			break
90			for litmusSubstring in self.break_after_litmus:
91			if litmusSubstring in self.passedLitmus and litmusSubstring not in litmusTag:
92			print(" *** break after litmus %s" % litmusTag,
93			file=self.out)
94			sys.exit(-1)
95			if litmusSubstring in litmusTag:
96			self.passedLitmus[litmusSubstring] = True
97
98			# Turn on max. debugging for selected request methods
99			if verbose >= 2 and method in self.debug_methods:
100			verbose = 3
101			debugBreak = True
102			dumpRequest = True
103			dumpResponse = True
104
105			# Set debug options to environment
106			environ["wsgidav.verbose"] = verbose
107			# environ["wsgidav.debug_methods"] = self.debug_methods
108			environ["wsgidav.debug_break"] = debugBreak
109			environ["wsgidav.dump_request_body"] = dumpRequest
110			environ["wsgidav.dump_response_body"] = dumpResponse
111
112			# Dump request headers
113			if dumpRequest:
114			print("<%s> --- %s Request ---" % (
115			threading.currentThread().ident, method), file=self.out)
116			for k, v in environ.items():
117			if k == k.upper():
118			print("%20s: '%s'" % (k, v), file=self.out)
119			print("\n", file=self.out)
120			self._dump_request(environ, xml=None)
121
122			# Intercept start_response
123			#
124			sub_app_start_response = util.SubAppStartResponse()
125
126			nbytes = 0
127			first_yield = True
128			app_iter = self._application(environ, sub_app_start_response)
129
130			for v in app_iter:
131			# Start response (the first time)
132			if first_yield:
133			# Success!
134			start_response(sub_app_start_response.status,
135			sub_app_start_response.response_headers,
136			sub_app_start_response.exc_info)
137
138			# Dump response headers
139			if first_yield and dumpResponse:
140			print("<%s> --- %s Response(%s): ---" % (
141			threading.currentThread().ident,
142			method,
143			sub_app_start_response.status),
144			file=self.out)
145			headersdict = dict(sub_app_start_response.response_headers)
146			for envitem in headersdict.keys():
147			print("%s: %s" % (envitem, repr(headersdict[envitem])),
148			file=self.out)
149			print("", file=self.out)
150
151			# Check, if response is a binary string, otherwise we probably have
152			# calculated a wrong content-length
153			assert compat.is_bytes(v), v
154
155			# Dump response body
156			drb = environ.get("wsgidav.dump_response_body")
157			if compat.is_basestring(drb):
158			# Middleware provided a formatted body representation
159			print(drb, file=self.out)
160			elif drb is True:
161			# Else dump what we get, (except for long GET responses)
162			if method == "GET":
163			if first_yield:
164			print(v[:50], "...", file=self.out)
165			elif len(v) > 0:
166			print(v, file=self.out)
167
168			if dumpResponse:
169			self._dump_response(sub_app_start_response, drb)
170
171			drb = environ["wsgidav.dump_response_body"] = None
172
173			nbytes += len(v)
174			first_yield = False
175			yield v
176			if hasattr(app_iter, "close"):
177			app_iter.close()
178
179			# Start response (if it hasn't been done yet)
180			if first_yield:
181			# Success!
182			start_response(sub_app_start_response.status,
183			sub_app_start_response.response_headers,
184			sub_app_start_response.exc_info)
185
186			if dumpResponse:
187			print("\n<%s> --- End of %s Response (%i bytes) ---" % (
188			threading.currentThread().ident, method, nbytes), file=self.out)
189			return
190
191			def _dump_response(self, sub_app_start_response, drb):
192			dump_to_path = self._config.get(
193			'dump_requests_path',
194			'/tmp/wsgidav_dumps',
195			)
196			os.makedirs(dump_to_path, exist_ok=True)
197			dump_file = '{0}/{1}_RESPONSE_{2}.yml'.format(
198			dump_to_path,
199			self.last_request_time,
200			sub_app_start_response.status[0:3],
201			)
202			with open(dump_file, 'w+') as f:
203			dump_content = dict()
204			headers = {}
205			for header_tuple in sub_app_start_response.response_headers:
206			headers[header_tuple[0]] = header_tuple[1]
207			dump_content['headers'] = headers
208			if isinstance(drb, str):
209			dump_content['content'] = drb.replace('PROPFIND XML response body:\n', '')
210
211			f.write(yaml.dump(dump_content, default_flow_style=False))
212
213			def _dump_request(self, environ, xml):
214			dump_to_path = self._config.get(
215			'dump_requests_path',
216			'/tmp/wsgidav_dumps',
217			)
218			os.makedirs(dump_to_path, exist_ok=True)
219			dump_file = '{0}/{1}_REQUEST_{2}.yml'.format(
220			dump_to_path,
221			self.last_request_time,
222			environ['REQUEST_METHOD'],
223			)
224			with open(dump_file, 'w+') as f:
225			dump_content = dict()
226			dump_content['path'] = environ.get('PATH_INFO', '')
227			dump_content['Authorization'] = environ.get('HTTP_AUTHORIZATION', '')
228			if xml:
229			dump_content['content'] = ElementTree.tostring(xml, 'utf-8')
230
231			f.write(yaml.dump(dump_content, default_flow_style=False))
232
233
234			class TracimEnforceHTTPS(BaseMiddleware):
235
236			def __init__(self, application, config):
237			super().__init__(application, config)
238			self._application = application
239			self._config = config
240
241			def __call__(self, environ, start_response):
242			# TODO - G.M - 06-03-2018 - Check protocol from http header first
243			# see http://www.bortzmeyer.org/7239.html
244			# if this params doesn't exist, rely on tracim config
245			# from tracim.config.app_cfg import CFG
246			# cfg = CFG.get_instance()
247			#
248			# if cfg.WEBSITE_BASE_URL.startswith('https'):
249			# environ['wsgi.url_scheme'] = 'https'
250			return self._application(environ, start_response)
251
252
253			class TracimEnv(BaseMiddleware):
254
255			def __init__(self, application, config):
256			super().__init__(application, config)
257			self._application = application
258			self._config = config
259			global_conf = get_appsettings(config['tracim_config']).global_conf
260			local_conf = get_appsettings(config['tracim_config'], 'tracim_web')
261			self.settings = global_conf
262			self.settings.update(local_conf)
263			self.engine = get_engine(self.settings)
264			self.session_factory = get_session_factory(self.engine)
265			self.app_config = CFG(self.settings)
266			self.app_config.configure_filedepot()
267
268			def __call__(self, environ, start_response):
269			# TODO - G.M - 18-05-2018 - This code should not create trouble
270			# with thread and database, this should be verify.
271			# see https://github.com/tracim/tracim_backend/issues/62
272			tm = transaction.manager
273			dbsession = get_tm_session(self.session_factory, tm)
274			environ['tracim_tm'] = tm
275			environ['tracim_dbsession'] = dbsession
276			environ['tracim_cfg'] = self.app_config
277			app = self._application(environ, start_response)
278			dbsession.close()
279			return app
280
281
282			class TracimUserSession(BaseMiddleware):
283
284			def __init__(self, application, config):
285			super().__init__(application, config)
286			self._application = application
287			self._config = config
288
289			def __call__(self, environ, start_response):
290			environ['tracim_user'] = UserApi(
291			None,
292			session=environ['tracim_dbsession'],
293			config=environ['tracim_cfg'],
294			).get_one_by_email(environ['http_authenticator.username'])
295			return self._application(environ, start_response)
296

tracim / tracim_v2

TracimUserSession.__call__() A last analyzed 2018-11-14 14:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TracimUserSession.call() A
last analyzed 2018-11-14 14:00 UTC