Coverage for runmacs/processor/frontend/tools.py : 43%

# -*- coding: utf-8 -*- 

# Filename: tools.py 

""" 

frontendServer - render macsServer database to web page 

======================================================= 

Copyright (C) 2013 Tobias Kölling 

""" 

import tornado.web 

import datetime 

from runmacs.processor.utils import dthandler, parse_dates_in_tree 

import json 

import urlparse 

import hashlib 

https_hostnames = ['macsserver.physik.uni-muenchen.de', 

'macsserver.physik.lmu.de'] 

class RequestHandler(tornado.web.RequestHandler): 

_current_rights = None 

conservativeEtagGeneration = False 

def staticFile(self, fn): 

return self.serverPrefix + "/static/" + fn 

def initialize(self, *args, **kwargs): 

self.serverPrefix = self.request.headers.get('X-Server-Prefix', '') 

self.host = self.request.headers.get('X-Forwarded-Host', self.request.host) 

28 ↛ 29line 28 didn't jump to line 29, because the condition on line 28 was never true if self.host in https_hostnames: 

self.protocol = 'https' 

self.dap_requires_key = True 

else: 

self.protocol = 'http' 

self.dap_requires_key = False 

self.urlprefix = '%s://%s'%(self.protocol, self.host) 

self.nav = [{'target': self.serverPrefix + '/calibrated.html', 'name': 'Calibrated'}, 

{'target': self.serverPrefix + '/aligned.html', 'name': 'Aligned',}, 

{'target': self.serverPrefix + '/polar.html', 'name': 'Polar',}, 

{'target': self.serverPrefix + '/equi.html', 'name': 'Equirectangular',}] 

def compute_etag(self): 

if self.conservativeEtagGeneration or self.request.method != 'GET': 

return super(RequestHandler, self).compute_etag() 

else: 

uri = self.request.uri 

uriparts = list(urlparse.urlparse(uri)) 

query = uriparts[4].split('&') 

query = [x for x in query if not x.startswith('key=')] 

uriparts[4] = '&'.join(query) 

uri = urlparse.urlunparse(uriparts) 

roles = '&'.join(sorted(self.current_roles)) 

gitrev = self.settings['git-rev'] 

return hashlib.sha256('!'.join((uri, roles, gitrev))).hexdigest() 

def chk_etag(self): 

self.set_etag_header() 

54 ↛ 55line 54 didn't jump to line 55, because the condition on line 54 was never true if self.check_etag_header(): 

self.set_status(304) 

return True 

else: 

return False 

def render(self, *args, **kwargs): 

kwargs['staticFile'] = self.staticFile 

kwargs['prefix'] = self.serverPrefix 

kwargs['nav'] = self.nav 

kwargs['protocol'] = self.protocol 

kwargs['host'] = self.host 

kwargs['urlprefix'] = self.urlprefix 

kwargs['current_user'] = self.current_user 

kwargs['special_rights'] = self.special_rights 

kwargs['dap_requires_key'] = self.dap_requires_key 

return super(RequestHandler, self).render(*args, **kwargs) 

@property 

def current_user(self): 

headers = self.request.headers 

if 'X-Varnish' in headers: 

return self.request.headers.get('X-Auth-User', None) 

else: 

return None 

@property 

def current_roles(self): 

headers = self.request.headers 

81 ↛ 82line 81 didn't jump to line 82, because the condition on line 81 was never true if 'X-Varnish' in headers: 

return self.request.headers.get('X-Auth-Roles', '').split() 

else: 

return ['local'] 

@property 

def current_rights(self): 

if self._current_rights is None: 

rolesCollection = self.settings['authDB'].roles 

if rolesCollection is None: 

return None 

rights = list(rolesCollection.find({'name': {'$in': self.current_roles}})) 

specialRights = [] 

for r in rights: 

specialRights += r.get('special', []) 

queryRestrictions = [] 

for r in rights: 

try: 

queryRestrictions.append(json.loads(r['queryRestriction'])) 

except KeyError: 

continue 

self._current_rights = {'special': specialRights, 'queryRestrictions': parse_dates_in_tree(queryRestrictions)} 

return self._current_rights 

@property 

def special_rights(self): 

current_rights = self.current_rights 

if current_rights is None: 

return [] 

return current_rights.get('special', []) 

def deny(self): 

self.set_status(403) 

def productType2PreviewInfo(pt): 

info = {} 

info['hasCoordinates'] = pt in ('aligned_image', 'projected_image') 

if pt in ('mount_position', 'window_transmission', 'aircraft_positions'): 

info['needPreviewCache'] = False 

info['previewExtension'] = 'png' 

else: 

info['needPreviewCache'] = True 

info['previewExtension'] = 'jpg' 

return info 

def addToDict(d, k, v): 

d[k] = v 

return d