autoland: add REST API unit tests (
bug 1368516) r?glob
Add some simple unit tests for the 'autoland' API endpoint to prove that our
test suite works.
MozReview-Commit-ID: Jg4iccJ1K4o
--- a/autoland/dev-requirements.txt
+++ b/autoland/dev-requirements.txt
@@ -1,8 +1,11 @@
pytest==3.0.7 \
--hash=sha256:66f332ae62593b874a648b10a8cb106bfdacd2c6288ed7dec3713c3a808a6017 \
--hash=sha256:b70696ebd1a5e6b627e7e3ac1365a4bc60aaf3495e843c1e70448966c5224cab
pytest-flask==0.10.0 \
--hash=sha256:2c5a36f9033ef8b6f85ddbefaebdd4f89197fc283f94b20dfe1a1beba4b77f03 \
--hash=sha256:657c7de386215ab0230bee4d76ace0339ae82fcbb34e134e17a29f65032eef03
pytest-pythonpath==0.7.1 \
--hash=sha256:2d506b8d7dbc2535a16c888211b7319ad32b3e73444bd9dbb1dd19427a6c7414
+mock==2.0.0 \
+ --hash=sha256:5ce3c71c5545b472da17b72268978914d0252980348636840bd34a00b5cc96c1 \
+ --hash=sha256:b158b6df76edd239b8208d481dc46b6afd45a846b7812ff0ce58971cf5bc8bba
--- a/autoland/tests/unit/test_autoland_api.py
+++ b/autoland/tests/unit/test_autoland_api.py
@@ -1,17 +1,137 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+import base64
+import json
+from flask import url_for
+from mock import MagicMock, sentinel
import pytest
import autoland_rest
+dummy_request = {
+ "ldap_username": "cthulhu@mozilla.org",
+ "tree": "mozilla-central",
+ "rev": "9cc25f7ac50a",
+ "destination": "try",
+ "trysyntax": "try: -b o -p linux -u mochitest-1 -t none",
+ "push_bookmark": "@",
+ "commit_descriptions": {
+ "9cc25f7ac50a": "bug 1 - did stuff r=gps"
+ },
+ "pingback_url": "http://localhost/"
+}
+
@pytest.fixture
def app():
return autoland_rest.app
+@pytest.fixture
+def autoland_config(monkeypatch):
+ """Swap out the .config module for a dict.
+
+ We need this because the .config module is hard-coded to read data from
+ disk.
+ """
+ fake_config = dict()
+ monkeypatch.setattr('autoland_rest.config', fake_config)
+ return fake_config
+
+
+@pytest.fixture
+def dbcursor(monkeypatch):
+ """Fake a database connection and return a Mock cursor object."""
+ cursor = MagicMock()
+ get_dbconn = MagicMock()
+ get_dbconn.return_value.cursor.return_value = cursor
+ monkeypatch.setattr('autoland_rest.get_dbconn', get_dbconn)
+ return cursor
+
+
+@pytest.fixture
+def auth_basic(autoland_config):
+ """Return a pre-configured HTTP Basic auth header.
+
+ Also monkeypatches the app config so that the API user and password are
+ valid.
+ """
+ autoland_config['auth'] = {'foo': 'password'}
+ return [('Authorization', 'Basic ' + base64.b64encode('foo:password'))]
+
+
+@pytest.fixture
+def content_json():
+ """Return a ready-to-use JSON Content-Type header."""
+ return [('Content-Type', 'application/json')]
+
+
+@pytest.fixture
+def mock_queued_job(dbcursor):
+ """Mock the database cursor return values for a successful job submission.
+ """
+ # During successful processing of an autoland request the database "queue"
+ # will be called 2 times: once to check that the job is already being
+ # processed, and once to insert the new job. We need to set the database
+ # cursor mock's return value to handle both calls.
+
+ # First return an empty result to indicate that the job is not already
+ # queued, then return a second result that is the new job's ID.
+ new_job_id = 123
+ query_results = [None, [new_job_id]]
+ dbcursor.fetchone.side_effect = query_results
+ return new_job_id
+
+
def test_hello(client):
response = client.get('/')
assert response.status_code == 200
+
+
+def test_request_transplant_returns_job_id(
+ client, accept_json, content_json, auth_basic, mock_queued_job
+):
+ headers = accept_json + auth_basic + content_json
+ response = client.post(
+ url_for('autoland'), headers=headers, data=json.dumps(dummy_request)
+ )
+ assert response.json == {'request_id': mock_queued_job}
+
+
+def test_fetch_existing_job_status(client, accept_json, dbcursor):
+ job_request = dict(pingback_url='')
+ destination = str(sentinel.destination)
+ landed = str(sentinel.landed)
+ result = str(sentinel.result)
+
+ # Mock the database to return the job status. The cursor should return the
+ # result of a single SQL "SELECT".
+ # Each DB row is a tuple of "destination, request_data, landed, result".
+ dbcursor.fetchone.return_value = (destination, job_request, landed, result)
+
+ # Status requests are unauthenticated.
+ response = client.get(
+ url_for('autoland_status', request_id=12345), headers=accept_json
+ )
+
+ expected_json = dict(
+ destination=destination, landed=landed, result=result, error_msg=''
+ )
+ assert response.json == expected_json
+
+
+def test_jobs_are_rejected_with_invalid_pingback_url(
+ client, accept_json, content_json, auth_basic, autoland_config
+):
+ autoland_config['pingback_allow'] = ['my_valid_hostname']
+ request = dummy_request.copy()
+ request['pingback_url'] = "http://sp00fer/"
+
+ headers = accept_json + auth_basic + content_json
+ response = client.post(
+ url_for('autoland'), headers=headers, data=json.dumps(request)
+ )
+
+ assert response.status_code == 400