Skip to content
Permalink
Browse files

[IMPROVEMENT] Add and Improve Unit-Tests (#302)

* Move helping functions to base.py; they can be used in other tests as well and hence should be available to all.
* Add and improve unit-tests for mod_deploy
* Add and Improve unittests for mod_auth
- new tests and improved old ones
- no calling of side-effects
- skipped decorator for template rendering
* Add more deploy tests; simplify deploy logic
* Remove unused role_id_is_valid method
- We don't have a Model for role and hence there is no way we can query it.
- This is a stale code and no usage of the same is there anywhere in the sample-platform.
  • Loading branch information...
thealphadollar authored and canihavesomecoffee committed Jun 21, 2019
1 parent 2062fba commit ad12e30ed39799f5e8026b76358d702d897c146c
@@ -62,20 +62,6 @@ def _email_not_in_use(form, field):
return _email_not_in_use


def role_id_is_valid(form, field):
"""
Check for validity of User's Role.
:param form: The form which is being passed in
:type form: Form
:param field : The data value for the 'role' inserted by User
:type field : SelectField
"""
role = Role.query.filter(Role.id == field.data).first()
if role is None:
raise ValidationError('Role id is invalid')


class LoginForm(FlaskForm):
"""Render form for User to enter Log in credentials."""

@@ -78,67 +78,64 @@ def is_valid_signature(x_hub_signature, data, private_key):
def deploy():
"""Deploy the GitHub request to the test platform."""
from run import app
if request.method != 'POST':
return 'OK'
else:
abort_code = 418

event = request.headers.get('X-GitHub-Event')
if event == "ping":
return json.dumps({'msg': 'Hi!'})
if event != "push":
return json.dumps({'msg': "Wrong event type"})

x_hub_signature = request.headers.get('X-Hub-Signature')
# webhook content type should be application/json for request.data to have the payload
# request.data is empty in case of x-www-form-urlencoded
if not is_valid_signature(x_hub_signature, request.data, g.github['deploy_key']):
g.log.warning('Deploy signature failed: {sig}'.format(sig=x_hub_signature))
abort(abort_code)

payload = request.get_json()
if payload is None:
g.log.warning('Deploy payload is empty: {payload}'.format(payload=payload))
abort(abort_code)

if payload['ref'] != 'refs/heads/master':
return json.dumps({'msg': 'Not master; ignoring'})

# Update code
try:
repo = Repo(app.config['INSTALL_FOLDER'])
except InvalidGitRepositoryError:
return json.dumps({'msg': 'Folder is not a valid git directory'})

try:
origin = repo.remote('origin')
except ValueError:
return json.dumps({'msg': 'Remote origin does not exist'})

fetch_info = origin.fetch()
if len(fetch_info) == 0:
return json.dumps({'msg': "Didn't fetch any information from remote!"})

# Pull code (finally)
pull_info = origin.pull()

if len(pull_info) == 0:
return json.dumps({'msg': "Didn't pull any information from remote!"})

if pull_info[0].flags > 128:
return json.dumps({'msg': "Didn't pull any information from remote!"})

commit_hash = pull_info[0].commit.hexsha
build_commit = 'build_commit = "{commit}"'.format(commit=commit_hash)
with open('build_commit.py', 'w') as f:
f.write(build_commit)

# Update runCI
run_ci_repo = path.join(app.config['INSTALL_FOLDER'], 'install', 'ci-vm', 'ci-linux', 'ci', 'runCI')
run_ci_nfs = path.join(app.config['SAMPLE_REPOSITORY'], 'vm_data', app.config['KVM_LINUX_NAME'], 'runCI')
copyfile(run_ci_repo, run_ci_nfs)

# Reload platform service
g.log.info('Platform upgraded to commit {commit}'.format(commit=commit_hash))
subprocess.Popen(["sudo", "service", "platform", "reload"])
return json.dumps({'msg': 'Platform upgraded to commit {commit}'.format(commit=commit_hash)})
abort_code = 418

event = request.headers.get('X-GitHub-Event')
if event == "ping":
return json.dumps({'msg': 'Hi!'})
if event != "push":
return json.dumps({'msg': "Wrong event type"})

x_hub_signature = request.headers.get('X-Hub-Signature')
# webhook content type should be application/json for request.data to have the payload
# request.data is empty in case of x-www-form-urlencoded
if not is_valid_signature(x_hub_signature, request.data, g.github['deploy_key']):
g.log.warning('Deploy signature failed: {sig}'.format(sig=x_hub_signature))
abort(abort_code)

payload = request.get_json()
if payload is None:
g.log.warning('Deploy payload is empty: {payload}'.format(payload=payload))
abort(abort_code)

if payload['ref'] != 'refs/heads/master':
return json.dumps({'msg': 'Not master; ignoring'})

# Update code
try:
repo = Repo(app.config['INSTALL_FOLDER'])
except InvalidGitRepositoryError:
return json.dumps({'msg': 'Folder is not a valid git directory'})

try:
origin = repo.remote('origin')
except ValueError:
return json.dumps({'msg': 'Remote origin does not exist'})

fetch_info = origin.fetch()
if len(fetch_info) == 0:
return json.dumps({'msg': "Didn't fetch any information from remote!"})

# Pull code (finally)
pull_info = origin.pull()

if len(pull_info) == 0:
return json.dumps({'msg': "Didn't pull any information from remote!"})

if pull_info[0].flags > 128:
return json.dumps({'msg': "Didn't pull any information from remote!"})

commit_hash = pull_info[0].commit.hexsha
build_commit = 'build_commit = "{commit}"'.format(commit=commit_hash)
with open('build_commit.py', 'w') as f:
f.write(build_commit)

# Update runCI
run_ci_repo = path.join(app.config['INSTALL_FOLDER'], 'install', 'ci-vm', 'ci-linux', 'ci', 'runCI')
run_ci_nfs = path.join(app.config['SAMPLE_REPOSITORY'], 'vm_data', app.config['KVM_LINUX_NAME'], 'runCI')
copyfile(run_ci_repo, run_ci_nfs)

# Reload platform service
g.log.info('Platform upgraded to commit {commit}'.format(commit=commit_hash))
subprocess.Popen(["sudo", "service", "platform", "reload"])
return json.dumps({'msg': 'Platform upgraded to commit {commit}'.format(commit=commit_hash)})
@@ -5,6 +5,7 @@

from flask import g
from flask_testing import TestCase
from werkzeug.datastructures import Headers

from database import create_session
from mod_auth.models import Role, User
@@ -111,6 +112,39 @@ def mock_api_request_github(url, data=None, timeout=None):
}


def generate_signature(data, private_key):
"""
Generate signature token of hook request
:param data: Signature's data
:param private_key: Signature's token
"""
import hashlib
import hmac
algorithm = hashlib.__dict__.get('sha1')
encoded_key = bytes(private_key, 'latin-1')
mac = hmac.new(encoded_key, msg=data, digestmod=algorithm)
return mac.hexdigest()


def generate_git_api_header(event, sig):
"""
Create header for Github API Request, based on header information from https://developer.github.com/webhooks/.
:param event: Name of the event type that triggered the delivery.
:param sig: The HMAC hex digest of the response body. The HMAC hex digest is generated
using the sha1 hash function and the secret as the HMAC key.
"""
return Headers([
('X-GitHub-Event', event),
('X-Github-Delivery', '72d3162e-cc78-11e3-81ab-4c9367dc0958'),
('X-Hub-Signature', 'sha1={0}'.format(sig)),
('User-Agent', 'GitHub-Hookshot/044aadd'),
('Content-Type', 'application/json'),
('Content-Length', 6615)
])


class BaseTestCase(TestCase):
@mock.patch('config_parser.parse_config', side_effect=load_config)
def create_app(self, mock_config):

0 comments on commit ad12e30

Please sign in to comment.
You can’t perform that action at this time.