Parent: [ddf08c] (diff)

Download this file

validation.py    322 lines (275 with data), 11.4 kB

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Functions to syntax-validate output content
"""
from os import path, environ, getcwd
import os
import sys
import logging
import tempfile
import subprocess
import json
import urllib2
import re
import tg
import mock
import beaker.session
from paste.deploy import loadapp
from paste.script.appinstall import SetupCommand
from pylons import tmpl_context as c, app_globals as g
from pylons import url, request, response, session
import webtest
from webtest import TestApp
from webob import Request, Response
from nose.tools import ok_, assert_true, assert_false
from poster.encode import multipart_encode
from poster.streaminghttp import register_openers
from ming.utils import LazyProperty
from allura.lib import utils
ENABLE_CONTENT_VALIDATION = False
# By default we want to run only validations which are fast,
# but on special test hosts - all.
COMPLETE_TESTS_HOST = 'sb-forge-4039'
log = logging.getLogger(__name__)
class Config(object):
"Config to encapsulate flexible/complex test enabled/disabled rules."
_instance = None
def __init__(self):
self.ini_config = None
pass
@classmethod
def instance(cls):
if not cls._instance:
cls._instance = cls()
return cls._instance
@LazyProperty
def test_ini(self):
if not self.ini_config:
from . import controller
import ConfigParser
conf = ConfigParser.ConfigParser({'validate_html5': 'false', 'validate_inlinejs': 'false'})
conf.read(controller.get_config_file())
self.ini_config = conf
return self.ini_config
@LazyProperty
def hostname(self):
if os.path.exists('/etc/soghost'):
with open('/etc/soghost') as fp:
return fp.read().strip()
def validation_enabled(self, val_type):
env_var = os.getenv('ALLURA_VALIDATION')
if env_var == 'all':
return True
elif env_var == 'none':
return False
elif env_var is not None:
return val_type in env_var.split(',')
if self.hostname == COMPLETE_TESTS_HOST:
return True
enabled = self.test_ini.getboolean('validation', 'validate_' + val_type)
return enabled
def fail_on_validation(self, val_type):
env_var = os.getenv('ALLURA_VALIDATION')
if env_var == 'all':
return True
if self.hostname == COMPLETE_TESTS_HOST:
return True
return ENABLE_CONTENT_VALIDATION
def report_validation_error(val_name, filename, message):
message = '%s Validation errors (%s):\n%s\n' % (val_name, filename, message)
if Config.instance().fail_on_validation(val_name):
ok_(False, message)
else:
sys.stderr.write('=' * 40 + '\n' + message)
def dump_to_file(prefix, html):
f = tempfile.NamedTemporaryFile(prefix=prefix, delete=False)
f.write(html)
f.close()
return f.name
def validate_html(html_or_response):
if hasattr(html_or_response, 'body'):
html = html_or_response.body
else:
html = html_or_response
html = html.lstrip()
if html.startswith('<!DOCTYPE html>'):
return validate_html5(html)
else:
assert False, 'Non-valid HTML: ' + html[:100] + '...'
def validate_json(json_or_response):
if hasattr(json_or_response, 'body'):
j = json_or_response.body
else:
j = json_or_response
try:
obj = json.loads(j)
except Exception, e:
ok_(False, "Couldn't validate JSON: " + str(e) + ':' + j[:100] + '...')
return obj
def validate_html5(html_or_response):
if hasattr(html_or_response, 'body'):
html = html_or_response.body
else:
html = html_or_response
register_openers()
params = [("out","text"),("content",html)]
datagen, headers = multipart_encode(params)
request = urllib2.Request("http://html5.validator.nu/", datagen, headers)
count = 3
while True:
try:
resp = urllib2.urlopen(request, timeout=3).read()
break
except:
resp = "Couldn't connect to validation service to check the HTML"
count -= 1
if count == 0:
sys.stderr.write('WARNING: ' + resp + '\n')
break
resp = resp.replace('���','"').replace('���','"').replace('���','-')
ignored_errors = [
'Required attributes missing on element "object"',
'Stray end tag "embed".',
'Stray end tag "param".',
r'Bad value .+? for attribute "onclick" on element "input": invalid return',
]
for ignore in ignored_errors:
resp = re.sub('Error: ' + ignore, 'Ignoring: ' + ignore, resp)
if 'Error:' in resp:
fname = dump_to_file('html5-', html)
message = resp.decode('ascii','ignore')
report_validation_error('html5', fname, message)
def validate_html5_chunk(html):
""" When you don't have a html & body tags - this adds it"""
# WebTest doesn't like HTML fragments without doctype,
# so we output them sometimes for fragments, which is hack.
# Unhack it here.
doctype = '<!DOCTYPE html>'
if html.startswith(doctype):
html = html[len(doctype):]
html = '''<!DOCTYPE html>
<html>
<head><title></title></head>
<body>
%s
</body></html>''' % html
return validate_html5(html)
def validate_js(html_or_response):
if hasattr(html_or_response, 'body'):
if html_or_response.status_int != 200:
return
html = html_or_response.body
else:
html = html_or_response
basedir = path.dirname(path.abspath(__file__))
jslint_dir = basedir + '/../jslint'
fname = dump_to_file('jslint-', html)
cmd = 'java -jar ' + jslint_dir + '/js.jar '+ jslint_dir +'/jslint.js ' + fname
p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, stderr = p.communicate(html)
if stdout.startswith('jslint: No problems found'):
os.unlink(fname)
return
stdout = stdout.decode('UTF-8', 'replace')
msg = '\n'.join(repr(s) for s in stdout.split('\n') if s)
report_validation_error('js', fname, msg)
def validate_page(html_or_response):
if Config.instance().validation_enabled('html5'):
validate_html(html_or_response)
if Config.instance().validation_enabled('inlinejs'):
validate_js(html_or_response)
class AntiSpamTestApp(TestApp):
def post(self, *args, **kwargs):
if kwargs.pop('antispam', False):
antispam = utils.AntiSpam()
params = {
'timestamp': antispam.timestamp_text,
'spinner': antispam.spinner_text,
antispam.enc('honey0'): '',
antispam.enc('honey1'): '',
}
for k, v in kwargs['params'].iteritems():
params[antispam.enc(k)] = v
kwargs['params'] = params
return super(AntiSpamTestApp, self).post(*args, **kwargs)
class PostParamCheckingTestApp(AntiSpamTestApp):
def _validate_params(self, params, method):
if not params:
return
# params can be a list or a dict
if hasattr(params, 'items'):
params = params.items()
for k, v in params:
if not isinstance(k, basestring):
raise TypeError('%s key %s is %s, not str' % (method, k, type(k)))
if not isinstance(v, (basestring, webtest.app.File)):
raise TypeError('%s key %s has value %s of type %s, not str. ' % (method, k, v, type(v)))
def get(self, *args, **kwargs):
self._validate_params(kwargs.get('params'), 'get')
return super(PostParamCheckingTestApp, self).get(*args, **kwargs)
def post(self, *args, **kwargs):
self._validate_params(kwargs.get('params'), 'post')
return super(PostParamCheckingTestApp, self).post(*args, **kwargs)
class ValidatingTestApp(PostParamCheckingTestApp):
# Subclasses may set this to True to skip validation altogether
validate_skip = False
def _validate(self, resp, method, val_params):
"""Perform validation on webapp response. This handles responses of
various types and forms."""
if resp.status_int != 200:
return
content = resp.body
content_type = resp.headers['Content-Type']
if content_type.startswith('text/html'):
if val_params['validate_chunk']:
validate_html5_chunk(content)
else:
validate_page(resp)
elif content_type.split(';', 1)[0] in ('text/plain', 'text/x-python', 'application/octet-stream'):
pass
elif content_type.startswith('application/json'):
validate_json(content)
elif content_type.startswith(('application/x-javascript','application/javascript', 'text/javascript')):
validate_js(content)
elif content_type.startswith('application/xml'):
import feedparser
d = feedparser.parse(content)
assert d.bozo == 0, 'Non-wellformed feed'
elif content_type.startswith('image/'):
pass
else:
assert False, 'Unexpected output content type: ' + content_type
def _get_validation_params(self, kw):
"Separate validation params from normal TestApp methods params."
params = {}
for k in ('validate_skip', 'validate_chunk'):
params[k] = kw.pop(k, False)
return params, kw
def get(self, *args, **kw):
val_params, kw = self._get_validation_params(kw)
resp = super(ValidatingTestApp, self).get(*args, **kw)
if not self.validate_skip and not val_params['validate_skip']:
self._validate(resp, 'get', val_params)
return resp
def post(self, *args, **kw):
val_params, kw = self._get_validation_params(kw)
resp = super(ValidatingTestApp, self).post(*args, **kw)
if not self.validate_skip and not val_params['validate_skip']:
self._validate(resp, 'post', val_params)
return resp