test_api_legacy.py

import flask
import pytest

from json import dumps, JSONEncoder

from flask import Blueprint, redirect, views
from werkzeug.exceptions import HTTPException, Unauthorized, BadRequest

import flask_restx as restx


# Add a dummy Resource to verify that the app is properly set.
class HelloWorld(restx.Resource):
    def get(self):
        return {}


class APITest(object):
    def test_unauthorized_no_challenge_by_default(self, api, mocker):
        response = mocker.Mock()
        response.headers = {}
        response = api.unauthorized(response)
        assert "WWW-Authenticate" not in response.headers

    @pytest.mark.api(serve_challenge_on_401=True)
    def test_unauthorized(self, api, mocker):
        response = mocker.Mock()
        response.headers = {}
        response = api.unauthorized(response)
        assert response.headers["WWW-Authenticate"] == 'Basic realm="flask-restx"'

    @pytest.mark.options(HTTP_BASIC_AUTH_REALM="Foo")
    @pytest.mark.api(serve_challenge_on_401=True)
    def test_unauthorized_custom_realm(self, api, mocker):
        response = mocker.Mock()
        response.headers = {}
        response = api.unauthorized(response)
        assert response.headers["WWW-Authenticate"] == 'Basic realm="Foo"'

    def test_handle_error_401_no_challenge_by_default(self, api):
        resp = api.handle_error(Unauthorized())
        assert resp.status_code == 401
        assert "WWW-Autheneticate" not in resp.headers

    @pytest.mark.api(serve_challenge_on_401=True)
    def test_handle_error_401_sends_challege_default_realm(self, api):
        exception = HTTPException()
        exception.code = 401
        exception.data = {"foo": "bar"}

        resp = api.handle_error(exception)
        assert resp.status_code == 401
        assert resp.headers["WWW-Authenticate"] == 'Basic realm="flask-restx"'

    @pytest.mark.api(serve_challenge_on_401=True)
    @pytest.mark.options(HTTP_BASIC_AUTH_REALM="test-realm")
    def test_handle_error_401_sends_challege_configured_realm(self, api):
        resp = api.handle_error(Unauthorized())
        assert resp.status_code == 401
        assert resp.headers["WWW-Authenticate"] == 'Basic realm="test-realm"'

    def test_handle_error_does_not_swallow_exceptions(self, api):
        exception = BadRequest("x")

        resp = api.handle_error(exception)
        assert resp.status_code == 400
        assert resp.get_data() == b'{"message": "x"}\n'

    def test_api_representation(self, api):
        @api.representation("foo")
        def foo():
            pass

        assert api.representations["foo"] == foo

    def test_api_base(self, app):
        api = restx.Api(app)
        assert api.urls == {}
        assert api.prefix == ""
        assert api.default_mediatype == "application/json"

    def test_api_delayed_initialization(self, app, client):
        api = restx.Api()
        api.add_resource(HelloWorld, "/", endpoint="hello")
        api.init_app(app)
        assert client.get("/").status_code == 200

    def test_api_prefix(self, app):
        api = restx.Api(app, prefix="/foo")
        assert api.prefix == "/foo"

    @pytest.mark.api(serve_challenge_on_401=True)
    def test_handle_auth(self, api):
        resp = api.handle_error(Unauthorized())
        assert resp.status_code == 401
        expected_data = dumps({"message": Unauthorized.description}) + "\n"
        assert resp.data.decode() == expected_data

        assert "WWW-Authenticate" in resp.headers

    def test_media_types(self, app):
        api = restx.Api(app)

        with app.test_request_context("/foo", headers={"Accept": "application/json"}):
            assert api.mediatypes() == ["application/json"]

    def test_media_types_method(self, app, mocker):
        api = restx.Api(app)

        with app.test_request_context(
            "/foo", headers={"Accept": "application/xml; q=0.5"}
        ):
            assert api.mediatypes_method()(mocker.Mock()) == [
                "application/xml",
                "application/json",
            ]

    def test_media_types_q(self, app):
        api = restx.Api(app)

        with app.test_request_context(
            "/foo",
            headers={"Accept": "application/json; q=1.0, application/xml; q=0.5"},
        ):
            assert api.mediatypes() == ["application/json", "application/xml"]

    def test_decorator(self, mocker, mock_app):
        def return_zero(func):
            return 0

        view = mocker.Mock()
        api = restx.Api(mock_app)
        api.decorators.append(return_zero)
        api.output = mocker.Mock()
        api.add_resource(view, "/foo", endpoint="bar")

        mock_app.add_url_rule.assert_called_with("/foo", view_func=0)

    def test_add_resource_endpoint(self, app, mocker):
        view = mocker.Mock(**{"as_view.return_value.__name__": str("test_view")})

        api = restx.Api(app)
        api.add_resource(view, "/foo", endpoint="bar")

        view.as_view.assert_called_with("bar", api)

    def test_add_two_conflicting_resources_on_same_endpoint(self, app):
        api = restx.Api(app)

        class Foo1(restx.Resource):
            def get(self):
                return "foo1"

        class Foo2(restx.Resource):
            def get(self):
                return "foo2"

        api.add_resource(Foo1, "/foo", endpoint="bar")
        with pytest.raises(ValueError):
            api.add_resource(Foo2, "/foo/toto", endpoint="bar")

    def test_add_the_same_resource_on_same_endpoint(self, app):
        api = restx.Api(app)

        class Foo1(restx.Resource):
            def get(self):
                return "foo1"

        api.add_resource(Foo1, "/foo", endpoint="bar")
        api.add_resource(Foo1, "/foo/toto", endpoint="blah")

        with app.test_client() as client:
            foo1 = client.get("/foo")
            assert foo1.data == b'"foo1"\n'
            foo2 = client.get("/foo/toto")
            assert foo2.data == b'"foo1"\n'

    def test_add_resource(self, mocker, mock_app):
        api = restx.Api(mock_app)
        api.output = mocker.Mock()
        api.add_resource(views.MethodView, "/foo")

        mock_app.add_url_rule.assert_called_with("/foo", view_func=api.output())

    def test_add_resource_kwargs(self, mocker, mock_app):
        api = restx.Api(mock_app)
        api.output = mocker.Mock()
        api.add_resource(views.MethodView, "/foo", defaults={"bar": "baz"})

        mock_app.add_url_rule.assert_called_with(
            "/foo", view_func=api.output(), defaults={"bar": "baz"}
        )

    def test_add_resource_forward_resource_class_parameters(self, app, client):
        api = restx.Api(app)

        class Foo(restx.Resource):
            def __init__(self, api, *args, **kwargs):
                self.one = args[0]
                self.two = kwargs["secret_state"]
                super(Foo, self).__init__(api, *args, **kwargs)

            def get(self):
                return "{0} {1}".format(self.one, self.two)

        api.add_resource(
            Foo,
            "/foo",
            resource_class_args=("wonderful",),
            resource_class_kwargs={"secret_state": "slurm"},
        )

        foo = client.get("/foo")
        assert foo.data == b'"wonderful slurm"\n'

    def test_output_unpack(self, app):
        def make_empty_response():
            return {"foo": "bar"}

        api = restx.Api(app)

        with app.test_request_context("/foo"):
            wrapper = api.output(make_empty_response)
            resp = wrapper()
            assert resp.status_code == 200
            assert resp.data.decode() == '{"foo": "bar"}\n'

    def test_output_func(self, app):
        def make_empty_resposne():
            return flask.make_response("")

        api = restx.Api(app)

        with app.test_request_context("/foo"):
            wrapper = api.output(make_empty_resposne)
            resp = wrapper()
            assert resp.status_code == 200
            assert resp.data.decode() == ""

    def test_resource(self, app, mocker):
        resource = restx.Resource()
        resource.get = mocker.Mock()
        with app.test_request_context("/foo"):
            resource.dispatch_request()

    def test_resource_resp(self, app, mocker):
        resource = restx.Resource()
        resource.get = mocker.Mock()
        with app.test_request_context("/foo"):
            resource.get.return_value = flask.make_response("")
            resource.dispatch_request()

    def test_resource_text_plain(self, app):
        def text(data, code, headers=None):
            return flask.make_response(str(data))

        class Foo(restx.Resource):
            representations = {
                "text/plain": text,
            }

            def get(self):
                return "hello"

        with app.test_request_context("/foo", headers={"Accept": "text/plain"}):
            resource = Foo(None)
            resp = resource.dispatch_request()
            assert resp.data.decode() == "hello"

    @pytest.mark.request_context("/foo")
    def test_resource_error(self, app):
        resource = restx.Resource()
        with pytest.raises(AssertionError):
            resource.dispatch_request()

    @pytest.mark.request_context("/foo", method="HEAD")
    def test_resource_head(self, app):
        resource = restx.Resource()
        with pytest.raises(AssertionError):
            resource.dispatch_request()

    def test_endpoints(self, app):
        api = restx.Api(app)
        api.add_resource(HelloWorld, "/ids/<int:id>", endpoint="hello")
        with app.test_request_context("/foo"):
            assert api._has_fr_route() is False

        with app.test_request_context("/ids/3"):
            assert api._has_fr_route() is True

    def test_url_for(self, app):
        api = restx.Api(app)
        api.add_resource(HelloWorld, "/ids/<int:id>")
        with app.test_request_context("/foo"):
            assert api.url_for(HelloWorld, id=123) == "/ids/123"

    def test_url_for_with_blueprint(self, app):
        """Verify that url_for works when an Api object is mounted on a
        Blueprint.
        """
        api_bp = Blueprint("api", __name__)
        api = restx.Api(api_bp)
        api.add_resource(HelloWorld, "/foo/<string:bar>")
        app.register_blueprint(api_bp)
        with app.test_request_context("/foo"):
            assert api.url_for(HelloWorld, bar="baz") == "/foo/baz"

    def test_exception_header_forwarding_doesnt_duplicate_headers(self, api):
        """Test that HTTPException's headers do not add a duplicate
        Content-Length header

        https://github.com/flask-restful/flask-restful/issues/534
        """
        r = api.handle_error(BadRequest())
        assert len(r.headers.getlist("Content-Length")) == 1

    def test_read_json_settings_from_config(self, app, client):
        class TestConfig(object):
            RESTX_JSON = {"indent": 2, "sort_keys": True, "separators": (", ", ": ")}

        app.config.from_object(TestConfig)
        api = restx.Api(app)

        class Foo(restx.Resource):
            def get(self):
                return {"foo": "bar", "baz": "qux"}

        api.add_resource(Foo, "/foo")

        data = client.get("/foo").data

        expected = b'{\n  "baz": "qux", \n  "foo": "bar"\n}\n'

        assert data == expected

    def test_use_custom_jsonencoder(self, app, client):
        def default(obj):
            return "cabbage"

        app.json.default = default
        api = restx.Api(app)

        class Cabbage(restx.Resource):
            def get(self):
                return {"frob": object()}

        api.add_resource(Cabbage, "/cabbage")

        data = client.get("/cabbage").data

        expected = b'{"frob": "cabbage"}\n'
        assert data == expected

    def test_json_with_no_settings(self, api, client):
        class Foo(restx.Resource):
            def get(self):
                return {"foo": "bar"}

        api.add_resource(Foo, "/foo")

        data = client.get("/foo").data

        expected = b'{"foo": "bar"}\n'
        assert data == expected

    def test_redirect(self, api, client):
        class FooResource(restx.Resource):
            def get(self):
                return redirect("/")

        api.add_resource(FooResource, "/api")

        resp = client.get("/api")
        assert resp.status_code == 302
        # FIXME: The behavior changed somewhere between Flask 2.0.3 and 2.2.x
        assert resp.headers["Location"].endswith("/")

    def test_calling_owns_endpoint_before_api_init(self):
        api = restx.Api()
        api.owns_endpoint("endpoint")
        # with pytest.raises(AttributeError):
        # try:
        # except AttributeError as ae:
        #     self.fail(ae.message)