autoland: run unit tests with pytest (bug 1368516) r?glob draft
authorMāris Fogels <mars@mozilla.com>
Tue, 30 May 2017 13:05:42 -0400
changeset 11133 f509827fc8d0bbe9a67d56871a319cf19c834a45
parent 11114 ce4a8132a4d7ee4445b3a1fe7f86b5f24eb1694f
push id1693
push usermfogels@mozilla.com
push dateThu, 01 Jun 2017 00:42:21 +0000
reviewersglob
bugs1368516
autoland: run unit tests with pytest (bug 1368516) r?glob Configure the pytest unit test runner so it can execute tests against the autoland REST API. Add some simple unit tests for the 'autoland' API endpoint to prove that our test suite works. MozReview-Commit-ID: Jg4iccJ1K4o
autoland/dev-requirements.txt
autoland/pytest.ini
autoland/tests/unit/test_autoland_api.py
new file mode 100644
--- /dev/null
+++ b/autoland/dev-requirements.txt
@@ -0,0 +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
new file mode 100644
--- /dev/null
+++ b/autoland/pytest.ini
@@ -0,0 +1,2 @@
+[pytest]
+python_paths = autoland
new file mode 100644
--- /dev/null
+++ b/autoland/tests/unit/test_autoland_api.py
@@ -0,0 +1,153 @@
+# 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():
+    """Required by pytest-flask."""
+    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 success_headers(accept_json, auth_basic, content_json):
+    """Return headers that will result in a successful API request."""
+    return accept_json + auth_basic + content_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, success_headers, mock_queued_job
+):
+    response = client.post(
+        url_for('autoland'),
+        headers=success_headers,
+        data=json.dumps(dummy_request)
+    )
+    assert response.json == {'request_id': mock_queued_job}
+
+
+def test_fetch_job_status_returns_db_values(client, accept_json, dbcursor):
+    original_request = dummy_request.copy()
+    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, original_request, landed, result
+    )
+
+    # Status requests are unauthenticated.  We only need an Accept header.
+    response = client.get(
+        url_for('autoland_status', request_id=12345), headers=accept_json
+    )
+
+    expected_json = original_request.copy()
+    expected_json.update(
+        {
+            'destination': destination,
+            'landed': landed,
+            'result': result,
+            'error_msg': ''
+        }
+    )
+    # pingback_url is not returned in job status results
+    del expected_json['pingback_url']
+
+    assert response.json == expected_json
+
+
+def test_jobs_are_rejected_with_invalid_pingback_url(
+    client, success_headers, autoland_config
+):
+    autoland_config['pingback_allow'] = ['my_valid_hostname']
+    request = dummy_request.copy()
+    request['pingback_url'] = "http://sp00fer/"
+    response = client.post(
+        url_for('autoland'), headers=success_headers, data=json.dumps(request)
+    )
+    assert response.status_code == 400