Commit d4aa6aac authored by ale's avatar ale

added Python tests

parent f1058773
#!/usr/bin/python
#
# FastCGI receiver component of the offlinescan package.
# (Meant to be run behind authentication).
#
import json
import time
import redis
from flask import Flask, request, abort, g
app = Flask(__name__)
def get_redis_connection():
return redis.Redis(
host=app.config['REDIS_HOST'],
password=app.config['REDIS_PASSWORD'],
socket_timeout=10)
@app.before_request
def before_request():
g.redis = get_redis_connection()
@app.route('/submit', methods=('GET', 'POST'))
def receive():
if request.method == 'POST':
p = g.redis.pipeline(transaction=False)
#per_user = {}
for user, results in request.json.iteritems():
key = 'virus:%s' % user
#p.ltrim(key, 0, 0)
for r in results:
p.lpush(key, r)
#per_user[user] = len(results)
# Save some incremental stats with per-user infection counts.
#new_ptr = 'virus_stats:%f' % time.time()
#for user, count in per_user.iteritems():
# p.hset(new_ptr, user, count)
#p.set('virus_stats:head', new_ptr)
p.execute()
return 'ok'
def create_app():
# The env variable name is compatible with wsgi-app-installer.
app.config.from_envvar('APP_CONFIG', silent=True)
return app
def main():
from flup.server.fcgi import WSGIServer
WSGIServer(create_app()).run()
if __name__ == '__main__':
main()
import os
import unittest
from offlinescan.incremental_scan import *
class IncrementalScanTest(unittest.TestCase):
def setUp(self):
self.testroot = os.path.join(os.path.dirname(__file__), 'testdata')
def tearDown(self):
try:
os.remove(os.path.join(
self.testroot, 'rdiff-backup-data/.offlinescan_stamp'))
except:
pass
def _fq(self, inset):
return set(os.path.join(self.testroot, f) for f in inset)
def test_scan_dirs(self):
result = scan_dirs([self.testroot], [], [])
self.assertEquals(
self._fq({'testfile', 'lot of whitespaces'}),
result)
def test_excludes(self):
result = scan_dirs([self.testroot], [], ['test*'])
self.assertEquals(
self._fq({'lot of whitespaces'}),
result)
def test_includes(self):
result = scan_dirs([self.testroot], ['test*'], [])
self.assertEquals(
self._fq({'testfile'}),
result)
def test_scan_dirs_fails_on_bad_dir(self):
self.assertRaises(Exception,
scan_dirs,
[self.testroot + '/..'], [], [])
#!/bin/bash
TESTS="simple-xargs"
run_test() {
local t="$1"
local run_dir=$(mktemp -d)
local output="${run_dir}/output"
local result=0
${SHELL} ../incremental-virus-scan.sh -r ${run_dir} ${t} | xargs -0 stat -c "%n" > ${output}
if ! cmp ${output} ${t}.expected; then
echo "${t}... FAILED"
diff -a ${t}.expected ${output}
result=1
else
echo "${t}... OK"
fi
rm -fr ${run_dir} 2>/dev/null
return ${result}
}
status=0
for t in ${TESTS}; do
run_test ${t} || status=1
done
exit ${status}
#!/bin/bash
TESTS="simple"
run_test() {
local t="$1"
local run_dir=$(mktemp -d)
local output="${run_dir}/output"
local result=0
${SHELL} ../incremental-virus-scan.sh -r ${run_dir} ${t} > ${output}
if ! cmp ${output} ${t}.expected ; then
echo "${t}... FAILED"
diff -a ${t}.expected ${output}
result=1
else
echo "${t}... OK"
fi
rm -fr ${run_dir} 2>/dev/null
return ${result}
}
status=0
for t in ${TESTS}; do
run_test ${t} || status=1
done
exit ${status}
simple-xargs/testfile
simple-xargs/lot of whitespaces
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment