In database terminology primary key refers to the column in a table that's intended to be the primary way of identifying rows. Each table must have exactly one, and it needs to be unique. This is usually some kind of a unique identifier associated with objects presented by the table, or if such an identifier doesn't exist simply a running ID number (which is incremented automatically).
Testing Flask Applications part 2¶
This page is a sequel to part 1 where we tested Flask applications manually and did some simple unit tests for
model classes
. This time we're going to do more "proper" testing. The focus is on API testing - ensuring that our API actually does what it promises. Just like in API implementation, some kind of strategy is crucial to stay sane. Otherwise there will be just endless amounts of copypaste code that is difficult to manage.Resource Testing with Pytest¶
Preliminary Preparations¶
You can grab the single file version of the sensor management API from below. This first part of this material uses the app from the ex2-06-caching-manual branch. We haven't added authentication yet. Let's cover basics first.
New Fix(ture)¶
Our fixture should be changed to one that yields a Flask test client. The new fixture looks like this:
from app import Measurement, Sensor, app, cache, db
@pytest.fixture
def client():
ctx = app.app_context()
ctx.push()
db.create_all()
_populate_db()
yield app.test_client()
cache.clear()
db.session.rollback()
db.drop_all()
db.session.remove()
ctx.pop()
In order to properly test an API, there usually needs to be some data in the database for GET, PUT and DELETE requests to have something to work on. Our new fixture calls the yet-to-be-defined
_populate_db function in order to achieve this. This function gives us some sensors and measurements to test with. Let's make them:def _populate_db():
for i in range(1, 4):
s = Sensor(
name="test-sensor-{}".format(i),
model="testsensor"
)
now = datetime.datetime.now()
interval = datetime.timedelta(seconds=10)
for i in range(125):
meas = Measurement(
value=round(random.random() * 100, 2),
time=now
)
now += interval
s.measurements.append(meas)
db.session.add(s)
db.session.commit()
Basically you can use the code you used for creating instances of each
model class
in database testing for this purpose, at least as a basis (just remove the asserts...) We're prefixing this function and its ilk with a single underscore to softly hint that these are the test module's internal tools.Basic Testing¶
View and
resource
testing in Flask is generally done with the Flask test client which we introduced in the previous testing material. We already modified the fixture to provide this client, and its use is rather straightforward. We also recommend grouping tests into classes, one test class per resource class
, mostly to get get some organization into the code with the added bonus of defining some constants as class attributes (e.g. the resource URI
). Pytest's discovery automatically creates instances of these classes and calls any methods that start or end with test. For basic unit testing, we want our tests to cover all nooks and crannies in the resource class methods thus ensuring that all lines of code actually work. This means covering all error scenarios in addition to testing with valid
requests
. For valid requests it may also be good to test that we get the data we expected, and likewise that our modifications actually take hold. With these things in mind, let's consider the first test: sensor collection get method test. We're also introducing the TestSensorCollection class.class TestSensorCollection(object):
RESOURCE_URL = "/api/sensors/"
def test_get(self, client):
resp = client.get(self.RESOURCE_URL)
assert resp.status_code == 200
body = json.loads(resp.data)
assert len(body) == 3
for item in body:
assert "name" in item
assert "model" in item
Because we created 3 sensors in the database population step, we're now ensuring that the API sends us all three, and that they have both of the attributes we expect them to have. This goes a bit beyond ensuring that the get method works. However, with this we can be sure that our API returns the data it is supposed to. For sensor collection's post method we have two options: we can wrap all the scenarios in one method, or we can put each
response
into its own. The examples below shows this as separate methods. The first method tests with a valid request and also checks that we can find the resource we created using the response's location header
. def test_post_valid_request(self, client):
valid = _get_sensor_json()
resp = client.post(self.RESOURCE_URL, json=valid)
assert resp.status_code == 201
assert resp.headers["Location"].endswith(self.RESOURCE_URL + valid["name"] + "/")
resp = client.get(resp.headers["Location"])
assert resp.status_code == 200
body = json.loads(resp.data)
assert body["name"] == "extra-sensor-1"
assert body["model"] == "extrasensor"
The second method sends invalid
media type
. With Flask's test client this can be done by not using the json keyword argument, and instead dumping the dictionary as a string into the data argument. def test_post_wrong_mediatype(self, client):
valid = _get_sensor_json()
resp = client.post(self.RESOURCE_URL, data=json.dumps(valid))
assert resp.status_code == 415
Another relatively simple test is sending a JSON document that doesn't pass through validation. In our case there is only one thing to test because our fields don't have value restriction: missing fields. For testing whether the method's validation handling works, one test case is sufficient.
def test_post_missing_field(self, client):
valid = _get_sensor_json()
valid.pop("model")
resp = client.post(self.RESOURCE_URL, json=valid)
assert resp.status_code == 400
Finally we need to test for conflict. Sensor names are unique, so we should try sending a POST request with a name that's already taken, e.g. one of the test sensors we created in database population.
def test_post_valid_request(self, client):
valid = _get_sensor_json()
valid["name"] = "test-sensor-1"
resp = client.post(self.RESOURCE_URL, json=valid)
assert resp.status_code == 409
On the item side, GET and PUT tests will look very similar and are thus not shown here - you can find them in the full example. That leaves DELETE test in which we should check that the deletion actually took by trying to send a GET request to the resource we just deleted.
class TestSensorItem(object):
RESOURCE_URL = "/api/sensors/test-sensor-1/"
INVALID_URL = "/api/sensors/non-sensor-x/"
def test_delete_valid(self, client):
resp = client.delete(self.RESOURCE_URL)
assert resp.status_code == 204
resp = client.get(self.RESOURCE_URL)
assert resp.status_code == 404
And finally test that sending a DELETE request to a sensor that doesn't exist returns a 404 error:
def test_delete_missing(self, client):
resp = client.delete(self.INVALID_URL)
assert resp.status_code == 404
Using Coverage¶
One nice tool to use with pytest is its coverage plugin. This plugin will track which lines of source code in the application being tested are executed during the tests. It's helpful in determining what kinds of tests are needed to ensure that every line in the program executes correctly. We already asked you to install the plugin along with pytest. We didn't use it for the database tests because database
models
didn't contain any callable code - class definitions are always executed as soon as the module imported, thus they will be always covered. Now that we have some callable code in our
resource class
methods, we can also see the coverage plugin in action. Assuming a single file application, you can run pytest with the following command line arguments to get a coverage summary in the terminal:pytest --cov-report term-missing --cov=app
Where the
--cov-report term-missing defines the reporting method to use and --cov=app defines for which module (or package) coverage should be tracked for. In our case we want to track our single file application, and we want a summary with line numbers printed for each line that is not covered. Using the test files in the repository you should see something like this:Name Stmts Miss Cover Missing -------------------------------------- app.py 169 20 88% 51-59, 62-66, 133, 138, 164, 195, 238-239, 258, 264 -------------------------------------- TOTAL 169 20 88%
The larger missing parts are related to the location models which we did not test. The rest are mostly stub methods.
You can see all of the available reporting options in coverage plugin's documentation. The limitation of coverage is that it only shows you that lines have been executed, it doesn't say anything about whether they actually did what was expected. Even with a 100% coverage it is not safe to say that your program is fully working according to its specification - it is "merely" fully working (assuming all tests pass).
Testing with Authentication¶
Once authentication is added to the application and keys are required for some views, the current set of tests will start to fail. The ex2-07-authentication branch in the project's repository already has the necessary modifications to test with authentication. The changes are explained below.
Creating the API Key¶
The first change we need to make is creating an API key. For running the server we added creating a random key to the database initialize script. For tests, the test module needs to actually know what the key is, so we have to modify the code a little bit. First we need to decide what the key is. Since we're only running tests in a local environment, its security is not important, and we can use something like:
TEST_KEY = "verysafetestkey"
Then we just create the corresponding key when populating the database. For these tests we're only creating the admin key.
def _populate_db():
for i in range(1, 4):
s = Sensor(
name="test-sensor-{}".format(i),
model="testsensor"
)
now = datetime.datetime.now()
interval = datetime.timedelta(seconds=10)
for i in range(125):
meas = Measurement(
value=round(random.random() * 100, 2),
time=now
)
now += interval
s.measurements.append(meas)
db.session.add(s)
db_key = ApiKey(
key=ApiKey.key_hash(TEST_KEY),
admin=True
)
db.session.add(db_key)
db.session.commit()
With this all of our test methods have access to the key's real value, allowing them to include it in the headers.
Adding Authentication Header¶
Now, we could of course simply add the API key header to each request in the tests. However, this isn't particularly sustainable. A better approach is to create a new child class of Flask's test client, and have it add the header to all requests.
# https://stackoverflow.com/questions/16416001/set-http-headers-for-all-requests-in-a-flask-test
class AuthHeaderClient(FlaskClient):
def open(self, *args, **kwargs):
headers = Headers({
'sensorhub-api-key': TEST_KEY
})
extra_headers = kwargs.pop('headers', Headers())
headers.extend(extra_headers)
kwargs['headers'] = headers
return super().open(*args, **kwargs)
This class adds some preprocessing before calling the parent's open method to open the connection. In particular, it sets the API key header to use the key we created when populating the database. Then, if the test method using this client gave any headers with its request, those headers are added. This also allows a test method to override the API key by providing the same header. This is useful for testing that invalid keys are actually rejected.
In case you are wondering about the extend method used here instead of update that you'd normally see when combining dictionaries, the reason is that internally the client class uses werkzeug's Headers type instead of normal dictionaries.
In the fixture, we can override the Flask app's test_client_class with the newly created AuthHeaderClient. Now whenever test_client is called it will return an instance of our custom class instead of the default FlaskClient class. With this our fixture becomes:
@pytest.fixture
def client():
ctx = app.app_context()
ctx.push()
db.create_all()
_populate_db()
app.test_client_class = AuthHeaderClient
yield app.test_client()
cache.clear()
db.session.rollback()
db.drop_all()
db.session.remove()
ctx.pop()
No changes are needed in the test cases themselves. If we wanted to test multiple different keys, then we'd need to add some mechanism to choose which key to use, but for now this will be sufficient.
Testing Invalid Keys¶
Finally, we want to make sure all the endpoints that should be protected are actually protected. For each them, we can write a test that overrides the correct key set by the custom client class with an incorrect one. Below is an example for testing SensorCollection's POST method.
def test_unauthorized(self, client):
valid = _get_sensor_json()
resp = client.post(self.RESOURCE_URL, json=valid, headers={"sensorhub-api-key": "wrongkey"})
assert resp.status_code == 403
Give feedback on this content
Comments about this material