test_connection_endpoint.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

import os
from unittest import mock

import pytest

from airflow.api_connexion.exceptions import EXCEPTIONS_LINK_MAP
from airflow.models import Connection
from airflow.secrets.environment_variables import CONN_ENV_PREFIX
from airflow.security import permissions
from airflow.utils.session import provide_session
from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_user
from tests.test_utils.config import conf_vars
from tests.test_utils.db import clear_db_connections
from tests.test_utils.www import _check_last_log

pytestmark = [pytest.mark.db_test, pytest.mark.skip_if_database_isolation_mode]


@pytest.fixture(scope="module")
def configured_app(minimal_app_for_api):
    app = minimal_app_for_api
    create_user(
        app,  # type: ignore
        username="test",
        role_name="Test",
        permissions=[
            (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_CONNECTION),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_CONNECTION),
            (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_CONNECTION),
            (permissions.ACTION_CAN_DELETE, permissions.RESOURCE_CONNECTION),
        ],
    )
    create_user(app, username="test_no_permissions", role_name="TestNoPermissions")  # type: ignore

    yield app

    delete_user(app, username="test")  # type: ignore
    delete_user(app, username="test_no_permissions")  # type: ignore


class TestConnectionEndpoint:
    @pytest.fixture(autouse=True)
    def setup_attrs(self, configured_app) -> None:
        self.app = configured_app
        self.client = self.app.test_client()  # type:ignore
        # we want only the connection created here for this test
        clear_db_connections(False)

    def teardown_method(self) -> None:
        clear_db_connections()

    def _create_connection(self, session):
        connection_model = Connection(conn_id="test-connection-id", conn_type="test_type")
        session.add(connection_model)
        session.commit()


class TestDeleteConnection(TestConnectionEndpoint):
    def test_delete_should_respond_204(self, session):
        connection_model = Connection(conn_id="test-connection", conn_type="test_type")

        session.add(connection_model)
        session.commit()
        conn = session.query(Connection).all()
        assert len(conn) == 1
        response = self.client.delete(
            "/api/v1/connections/test-connection", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 204
        connection = session.query(Connection).all()
        assert len(connection) == 0
        _check_last_log(session, dag_id=None, event="api.connection.delete", execution_date=None)

    def test_delete_should_respond_404(self):
        response = self.client.delete(
            "/api/v1/connections/test-connection", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 404
        assert response.json == {
            "detail": "The Connection with connection_id: `test-connection` was not found",
            "status": 404,
            "title": "Connection not found",
            "type": EXCEPTIONS_LINK_MAP[404],
        }

    def test_should_raises_401_unauthenticated(self):
        response = self.client.delete("/api/v1/connections/test-connection")

        assert_401(response)

    def test_should_raise_403_forbidden(self):
        response = self.client.get(
            "/api/v1/connections/test-connection-id", environ_overrides={"REMOTE_USER": "test_no_permissions"}
        )
        assert response.status_code == 403

    @pytest.mark.parametrize(
        "set_auto_role_public, expected_status_code",
        (("Public", 403), ("Admin", 204)),
        indirect=["set_auto_role_public"],
    )
    def test_with_auth_role_public_set(self, set_auto_role_public, expected_status_code, session):
        connection_model = Connection(conn_id="test-connection", conn_type="test_type")
        session.add(connection_model)
        session.commit()
        conn = session.query(Connection).all()
        assert len(conn) == 1

        response = self.client.delete("/api/v1/connections/test-connection")

        assert response.status_code == expected_status_code


class TestGetConnection(TestConnectionEndpoint):
    def test_should_respond_200(self, session):
        connection_model = Connection(
            conn_id="test-connection-id",
            conn_type="mysql",
            description="test description",
            host="mysql",
            login="login",
            schema="testschema",
            port=80,
            extra='{"param": "value"}',
        )
        session.add(connection_model)
        session.commit()
        result = session.query(Connection).all()
        assert len(result) == 1
        response = self.client.get(
            "/api/v1/connections/test-connection-id", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        assert response.json == {
            "connection_id": "test-connection-id",
            "conn_type": "mysql",
            "description": "test description",
            "host": "mysql",
            "login": "login",
            "schema": "testschema",
            "port": 80,
            "extra": '{"param": "value"}',
        }

    @pytest.mark.enable_redact
    def test_should_mask_sensitive_values_in_extra(self, session):
        connection_model = Connection(
            conn_id="test-connection-id",
            conn_type="mysql",
            description="test description",
            extra={"nonsensitive": "just_a_value", "api_token": "secretvalue"},
        )
        session.add(connection_model)
        session.commit()

        response = self.client.get(
            "/api/v1/connections/test-connection-id", environ_overrides={"REMOTE_USER": "test"}
        )

        assert response.json["extra"] == '{"nonsensitive": "just_a_value", "api_token": "***"}'

    def test_should_respond_404(self):
        response = self.client.get(
            "/api/v1/connections/invalid-connection", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 404
        assert {
            "detail": "The Connection with connection_id: `invalid-connection` was not found",
            "status": 404,
            "title": "Connection not found",
            "type": EXCEPTIONS_LINK_MAP[404],
        } == response.json

    def test_should_raises_401_unauthenticated(self):
        response = self.client.get("/api/v1/connections/test-connection-id")

        assert_401(response)

    @pytest.mark.parametrize(
        "set_auto_role_public, expected_status_code",
        (("Public", 403), ("Admin", 200)),
        indirect=["set_auto_role_public"],
    )
    def test_with_auth_role_public_set(self, set_auto_role_public, expected_status_code, session):
        connection_model = Connection(
            conn_id="test-connection-id",
            conn_type="mysql",
            description="test description",
            host="mysql",
            login="login",
            schema="testschema",
            port=80,
            extra='{"param": "value"}',
        )
        session.add(connection_model)
        session.commit()
        result = session.query(Connection).all()
        assert len(result) == 1

        response = self.client.get("/api/v1/connections/test-connection-id")

        assert response.status_code == expected_status_code


class TestGetConnections(TestConnectionEndpoint):
    def test_should_respond_200(self, session):
        connection_model_1 = Connection(conn_id="test-connection-id-1", conn_type="test_type")
        connection_model_2 = Connection(conn_id="test-connection-id-2", conn_type="test_type")
        connections = [connection_model_1, connection_model_2]
        session.add_all(connections)
        session.commit()
        result = session.query(Connection).all()
        assert len(result) == 2
        response = self.client.get("/api/v1/connections", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200
        assert response.json == {
            "connections": [
                {
                    "connection_id": "test-connection-id-1",
                    "conn_type": "test_type",
                    "description": None,
                    "host": None,
                    "login": None,
                    "schema": None,
                    "port": None,
                },
                {
                    "connection_id": "test-connection-id-2",
                    "conn_type": "test_type",
                    "description": None,
                    "host": None,
                    "login": None,
                    "schema": None,
                    "port": None,
                },
            ],
            "total_entries": 2,
        }

    def test_should_respond_200_with_order_by(self, session):
        connection_model_1 = Connection(conn_id="test-connection-id-1", conn_type="test_type")
        connection_model_2 = Connection(conn_id="test-connection-id-2", conn_type="test_type")
        connections = [connection_model_1, connection_model_2]
        session.add_all(connections)
        session.commit()
        result = session.query(Connection).all()
        assert len(result) == 2
        response = self.client.get(
            "/api/v1/connections?order_by=-connection_id", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        # Using - means descending
        assert response.json == {
            "connections": [
                {
                    "connection_id": "test-connection-id-2",
                    "conn_type": "test_type",
                    "description": None,
                    "host": None,
                    "login": None,
                    "schema": None,
                    "port": None,
                },
                {
                    "connection_id": "test-connection-id-1",
                    "conn_type": "test_type",
                    "description": None,
                    "host": None,
                    "login": None,
                    "schema": None,
                    "port": None,
                },
            ],
            "total_entries": 2,
        }

    def test_should_raises_401_unauthenticated(self):
        response = self.client.get("/api/v1/connections")

        assert_401(response)

    @pytest.mark.parametrize(
        "set_auto_role_public, expected_status_code",
        (("Public", 403), ("Admin", 200)),
        indirect=["set_auto_role_public"],
    )
    def test_with_auth_role_public_set(self, set_auto_role_public, expected_status_code):
        response = self.client.get("/api/v1/connections")

        assert response.status_code == expected_status_code


class TestGetConnectionsPagination(TestConnectionEndpoint):
    @pytest.mark.parametrize(
        "url, expected_conn_ids",
        [
            ("/api/v1/connections?limit=1", ["TEST_CONN_ID1"]),
            ("/api/v1/connections?limit=2", ["TEST_CONN_ID1", "TEST_CONN_ID2"]),
            (
                "/api/v1/connections?offset=5",
                [
                    "TEST_CONN_ID6",
                    "TEST_CONN_ID7",
                    "TEST_CONN_ID8",
                    "TEST_CONN_ID9",
                    "TEST_CONN_ID10",
                ],
            ),
            (
                "/api/v1/connections?offset=0",
                [
                    "TEST_CONN_ID1",
                    "TEST_CONN_ID2",
                    "TEST_CONN_ID3",
                    "TEST_CONN_ID4",
                    "TEST_CONN_ID5",
                    "TEST_CONN_ID6",
                    "TEST_CONN_ID7",
                    "TEST_CONN_ID8",
                    "TEST_CONN_ID9",
                    "TEST_CONN_ID10",
                ],
            ),
            ("/api/v1/connections?limit=1&offset=5", ["TEST_CONN_ID6"]),
            ("/api/v1/connections?limit=1&offset=1", ["TEST_CONN_ID2"]),
            (
                "/api/v1/connections?limit=2&offset=2",
                ["TEST_CONN_ID3", "TEST_CONN_ID4"],
            ),
        ],
    )
    @provide_session
    def test_handle_limit_offset(self, url, expected_conn_ids, session):
        connections = self._create_connections(10)
        session.add_all(connections)
        session.commit()
        response = self.client.get(url, environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200
        assert response.json["total_entries"] == 10
        conn_ids = [conn["connection_id"] for conn in response.json["connections"] if conn]
        assert conn_ids == expected_conn_ids

    def test_should_respect_page_size_limit_default(self, session):
        connection_models = self._create_connections(200)
        session.add_all(connection_models)
        session.commit()

        response = self.client.get("/api/v1/connections", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200

        assert response.json["total_entries"] == 200
        assert len(response.json["connections"]) == 100

    def test_invalid_order_by_raises_400(self, session):
        connection_models = self._create_connections(200)
        session.add_all(connection_models)
        session.commit()

        response = self.client.get(
            "/api/v1/connections?order_by=invalid", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 400
        assert (
            response.json["detail"] == "Ordering with 'invalid' is disallowed or"
            " the attribute does not exist on the model"
        )

    def test_limit_of_zero_should_return_default(self, session):
        connection_models = self._create_connections(200)
        session.add_all(connection_models)
        session.commit()

        response = self.client.get("/api/v1/connections?limit=0", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200

        assert response.json["total_entries"] == 200
        assert len(response.json["connections"]) == 100

    @conf_vars({("api", "maximum_page_limit"): "150"})
    def test_should_return_conf_max_if_req_max_above_conf(self, session):
        connection_models = self._create_connections(200)
        session.add_all(connection_models)
        session.commit()

        response = self.client.get("/api/v1/connections?limit=180", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200
        assert len(response.json["connections"]) == 150

    def _create_connections(self, count):
        return [
            Connection(conn_id=f"TEST_CONN_ID{i}", conn_type=f"TEST_CONN_TYPE{i}")
            for i in range(1, count + 1)
        ]


class TestPatchConnection(TestConnectionEndpoint):
    @pytest.mark.parametrize(
        "payload",
        [
            {"connection_id": "test-connection-id", "conn_type": "test_type", "extra": '{"key": "var"}'},
            {"extra": '{"key": "var"}'},
        ],
    )
    @provide_session
    def test_patch_should_respond_200(self, payload, session):
        self._create_connection(session)

        response = self.client.patch(
            "/api/v1/connections/test-connection-id", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        _check_last_log(session, dag_id=None, event="api.connection.edit", execution_date=None)

    def test_patch_should_respond_200_with_update_mask(self, session):
        self._create_connection(session)
        test_connection = "test-connection-id"
        payload = {
            "connection_id": test_connection,
            "conn_type": "test_type_2",
            "extra": "{'key': 'var'}",
            "login": "login",
            "port": 80,
        }
        response = self.client.patch(
            "/api/v1/connections/test-connection-id?update_mask=port,login",
            json=payload,
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 200
        connection = session.query(Connection).filter_by(conn_id=test_connection).first()
        assert connection.password is None
        assert response.json == {
            "connection_id": test_connection,  # not updated
            "conn_type": "test_type",  # Not updated
            "description": None,  # Not updated
            "extra": None,  # Not updated
            "login": "login",  # updated
            "port": 80,  # updated
            "schema": None,
            "host": None,
        }

    @pytest.mark.parametrize(
        "payload, update_mask, error_message",
        [
            (
                {
                    "connection_id": "test-connection-id",
                    "conn_type": "test_type_2",
                    "extra": "{'key': 'var'}",
                    "login": "login",
                    "port": 80,
                },
                "update_mask=ports, login",  # posts is unknown
                "'ports' is unknown or cannot be updated.",
            ),
            (
                {
                    "connection_id": "test-connection-id",
                    "conn_type": "test_type_2",
                    "extra": "{'key': 'var'}",
                    "login": "login",
                    "port": 80,
                },
                "update_mask=port, login, conn_id",  # conn_id is unknown
                "'conn_id' is unknown or cannot be updated.",
            ),
            (
                {
                    "connection_id": "test-connection-id",
                    "conn_type": "test_type_2",
                    "extra": "{'key': 'var'}",
                    "login": "login",
                    "port": 80,
                },
                "update_mask=port, login, connection_id",  # connection_id cannot be updated
                "'connection_id' is unknown or cannot be updated.",
            ),
            (
                {
                    "connection_id": "test-connection",  # trying to change connection_id
                    "conn_type": "test-type",
                    "login": "login",
                },
                "",  # not necessary
                "The connection_id cannot be updated.",
            ),
        ],
    )
    @provide_session
    def test_patch_should_respond_400_for_invalid_fields_in_update_mask(
        self, payload, update_mask, error_message, session
    ):
        self._create_connection(session)
        response = self.client.patch(
            f"/api/v1/connections/test-connection-id?{update_mask}",
            json=payload,
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 400
        assert response.json["detail"] == error_message

    @pytest.mark.parametrize(
        "payload, error_message",
        [
            (
                {
                    "connection_id": "test-connection-id",
                    "conn_type": "test-type",
                    "extra": 0,  # expected string
                },
                "0 is not of type 'string' - 'extra'",
            ),
            (
                {
                    "connection_id": "test-connection-id",
                    "conn_type": "test-type",
                    "extras": "{}",  # extras not a known field e.g typo
                },
                "extras",
            ),
            (
                {
                    "connection_id": "test-connection-id",
                    "conn_type": "test-type",
                    "invalid_field": "invalid field",  # unknown field
                    "_password": "{}",  # _password not a known field
                },
                "_password",
            ),
        ],
    )
    @provide_session
    def test_patch_should_respond_400_for_invalid_update(self, payload, error_message, session):
        self._create_connection(session)
        response = self.client.patch(
            "/api/v1/connections/test-connection-id", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 400
        assert error_message in response.json["detail"]

    def test_patch_should_respond_404_not_found(self):
        payload = {"connection_id": "test-connection-id", "conn_type": "test-type", "port": 90}
        response = self.client.patch(
            "/api/v1/connections/test-connection-id", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 404
        assert {
            "detail": "The Connection with connection_id: `test-connection-id` was not found",
            "status": 404,
            "title": "Connection not found",
            "type": EXCEPTIONS_LINK_MAP[404],
        } == response.json

    def test_should_raises_401_unauthenticated(self, session):
        self._create_connection(session)

        response = self.client.patch(
            "/api/v1/connections/test-connection-id",
            json={"connection_id": "test-connection-id", "conn_type": "test_type", "extra": "{'key': 'var'}"},
        )

        assert_401(response)

    @pytest.mark.parametrize(
        "set_auto_role_public, expected_status_code",
        (("Public", 403), ("Admin", 200)),
        indirect=["set_auto_role_public"],
    )
    def test_with_auth_role_public_set(self, set_auto_role_public, expected_status_code, session):
        self._create_connection(session)

        response = self.client.patch(
            "/api/v1/connections/test-connection-id",
            json={"connection_id": "test-connection-id", "conn_type": "test_type", "extra": '{"key": "var"}'},
        )

        assert response.status_code == expected_status_code


class TestPostConnection(TestConnectionEndpoint):
    def test_post_should_respond_200(self, session):
        payload = {"connection_id": "test-connection-id", "conn_type": "test_type"}
        response = self.client.post(
            "/api/v1/connections", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        connection = session.query(Connection).all()
        assert len(connection) == 1
        assert connection[0].conn_id == "test-connection-id"
        _check_last_log(
            session, dag_id=None, event="api.connection.create", execution_date=None, expected_extra=payload
        )

    def test_post_should_respond_200_extra_null(self, session):
        payload = {"connection_id": "test-connection-id", "conn_type": "test_type", "extra": None}
        response = self.client.post(
            "/api/v1/connections", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        assert response.json["extra"] is None
        connection = session.query(Connection).all()
        assert len(connection) == 1
        assert connection[0].conn_id == "test-connection-id"
        assert connection[0].extra is None

    def test_post_should_respond_400_for_invalid_payload(self):
        payload = {
            "connection_id": "test-connection-id",
        }  # conn_type missing
        response = self.client.post(
            "/api/v1/connections", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 400
        assert response.json == {
            "detail": "{'conn_type': ['Missing data for required field.']}",
            "status": 400,
            "title": "Bad Request",
            "type": EXCEPTIONS_LINK_MAP[400],
        }

    def test_post_should_respond_400_for_invalid_conn_id(self):
        payload = {"connection_id": "****", "conn_type": "test_type"}
        response = self.client.post(
            "/api/v1/connections", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 400
        assert response.json == {
            "detail": "The key '****' has to be made of "
            "alphanumeric characters, dashes, dots and underscores exclusively",
            "status": 400,
            "title": "Bad Request",
            "type": EXCEPTIONS_LINK_MAP[400],
        }

    def test_post_should_respond_409_already_exist(self):
        payload = {"connection_id": "test-connection-id", "conn_type": "test_type"}
        response = self.client.post(
            "/api/v1/connections", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        # Another request
        response = self.client.post(
            "/api/v1/connections", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 409
        assert response.json == {
            "detail": "Connection already exist. ID: test-connection-id",
            "status": 409,
            "title": "Conflict",
            "type": EXCEPTIONS_LINK_MAP[409],
        }

    def test_should_raises_401_unauthenticated(self):
        response = self.client.post(
            "/api/v1/connections", json={"connection_id": "test-connection-id", "conn_type": "test_type"}
        )

        assert_401(response)

    @pytest.mark.parametrize(
        "set_auto_role_public, expected_status_code",
        (("Public", 403), ("Admin", 200)),
        indirect=["set_auto_role_public"],
    )
    def test_with_auth_role_public_set(self, set_auto_role_public, expected_status_code):
        response = self.client.post(
            "/api/v1/connections", json={"connection_id": "test-connection-id", "conn_type": "test_type"}
        )

        assert response.status_code == expected_status_code


class TestConnection(TestConnectionEndpoint):
    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    def test_should_respond_200(self):
        payload = {"connection_id": "test-connection-id", "conn_type": "sqlite"}
        response = self.client.post(
            "/api/v1/connections/test", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        assert response.json == {
            "status": True,
            "message": "Connection successfully tested",
        }

    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    def test_connection_env_is_cleaned_after_run(self):
        payload = {"connection_id": "test-connection-id", "conn_type": "sqlite"}
        self.client.post("/api/v1/connections/test", json=payload, environ_overrides={"REMOTE_USER": "test"})
        assert not any([key.startswith(CONN_ENV_PREFIX) for key in os.environ.keys()])

    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    def test_post_should_respond_400_for_invalid_payload(self):
        payload = {
            "connection_id": "test-connection-id",
        }  # conn_type missing
        response = self.client.post(
            "/api/v1/connections/test", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 400
        assert response.json == {
            "detail": "{'conn_type': ['Missing data for required field.']}",
            "status": 400,
            "title": "Bad Request",
            "type": EXCEPTIONS_LINK_MAP[400],
        }

    def test_should_raises_401_unauthenticated(self):
        response = self.client.post(
            "/api/v1/connections/test", json={"connection_id": "test-connection-id", "conn_type": "test_type"}
        )

        assert_401(response)

    def test_should_respond_403_by_default(self):
        payload = {"connection_id": "test-connection-id", "conn_type": "sqlite"}
        response = self.client.post(
            "/api/v1/connections/test", json=payload, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 403
        assert response.text == (
            "Testing connections is disabled in Airflow configuration. "
            "Contact your deployment admin to enable it."
        )

    @pytest.mark.parametrize(
        "set_auto_role_public, expected_status_code",
        (("Public", 403), ("Admin", 200)),
        indirect=["set_auto_role_public"],
    )
    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    def test_with_auth_role_public_set(self, set_auto_role_public, expected_status_code):
        payload = {"connection_id": "test-connection-id", "conn_type": "sqlite"}
        response = self.client.post("/api/v1/connections/test", json=payload)
        assert response.status_code == expected_status_code