From 48dfa5a8b9729be325c4294bb49888661a8fcb38 Mon Sep 17 00:00:00 2001 From: Timo Stollenwerk Date: Fri, 22 Sep 2023 21:30:48 +0200 Subject: [PATCH] Run pre-commit.ci --- CHANGES.rst | 4 +- README.rst | 2 +- src/plone/rest/__init__.py | 1 - src/plone/rest/cors.py | 6 +- src/plone/rest/demo.py | 5 +- src/plone/rest/errors.py | 30 +++--- src/plone/rest/events.py | 4 +- src/plone/rest/interfaces.py | 1 - src/plone/rest/negotiation.py | 4 +- src/plone/rest/patches.py | 1 - src/plone/rest/service.py | 9 +- src/plone/rest/testing.py | 12 +-- src/plone/rest/tests/__init__.py | 1 - src/plone/rest/tests/test_cors.py | 19 ++-- src/plone/rest/tests/test_dexterity.py | 106 ++++++++++---------- src/plone/rest/tests/test_dispatching.py | 24 ++--- src/plone/rest/tests/test_error_handling.py | 31 +++--- src/plone/rest/tests/test_named_services.py | 29 +++--- src/plone/rest/tests/test_negotiation.py | 25 ++--- src/plone/rest/tests/test_permissions.py | 29 +++--- src/plone/rest/tests/test_redirects.py | 23 +++-- src/plone/rest/tests/test_siteroot.py | 41 ++++---- src/plone/rest/tests/test_traversal.py | 30 +++--- src/plone/rest/traverse.py | 21 ++-- src/plone/rest/zcml.py | 91 ++++++++--------- 25 files changed, 264 insertions(+), 285 deletions(-) diff --git a/CHANGES.rst b/CHANGES.rst index fc4e18a..dbb7900 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -245,7 +245,7 @@ Bugfixes: [buchi] - Fallback to regular views during traversal to ensure compatibility with - views beeing called with a specific Accept header. + views being called with a specific Accept header. [buchi] @@ -295,7 +295,7 @@ Bugfixes: - Refactor traversal of REST requests by using a traversal adapter on the site root instead of a traversal adapter for each REST service. This prevents - REST services from being overriden by other traversal adapters. + REST services from being overridden by other traversal adapters. [buchi] diff --git a/README.rst b/README.rst index 86e24f7..6c3a78d 100644 --- a/README.rst +++ b/README.rst @@ -271,7 +271,7 @@ allow_origin allow_methods A comma separated list of HTTP method names that are allowed by this CORS policy, e.g. "DELETE,GET,OPTIONS,PATCH,POST,PUT". If not specified, all - methods for which there's a service registerd are allowed. + methods for which there's a service registered are allowed. allow_credentials Indicates whether the resource supports user credentials in the request. diff --git a/src/plone/rest/__init__.py b/src/plone/rest/__init__.py index 44646e4..ab37f22 100644 --- a/src/plone/rest/__init__.py +++ b/src/plone/rest/__init__.py @@ -1,2 +1 @@ -# -*- coding: utf-8 -*- from plone.rest.service import Service # noqa diff --git a/src/plone/rest/cors.py b/src/plone/rest/cors.py index dbc2035..18b84d9 100644 --- a/src/plone/rest/cors.py +++ b/src/plone/rest/cors.py @@ -1,7 +1,7 @@ -# -*- coding: utf-8 -*- -from plone.rest.interfaces import ICORSPolicy from zope.interface import implementer +from plone.rest.interfaces import ICORSPolicy + # CORS preflight service registry # A mapping of method -> service_id _services = {} @@ -19,7 +19,7 @@ def lookup_preflight_service_id(method): @implementer(ICORSPolicy) -class CORSPolicy(object): +class CORSPolicy: def __init__(self, context, request): self.context = context self.request = request diff --git a/src/plone/rest/demo.py b/src/plone/rest/demo.py index 4622f97..91d8e8d 100644 --- a/src/plone/rest/demo.py +++ b/src/plone/rest/demo.py @@ -1,8 +1,7 @@ -# -*- coding: utf-8 -*- -from plone.rest import Service - import json +from plone.rest import Service + class BaseService(Service): def render(self): diff --git a/src/plone/rest/errors.py b/src/plone/rest/errors.py index 454d48e..5402d0b 100644 --- a/src/plone/rest/errors.py +++ b/src/plone/rest/errors.py @@ -4,32 +4,30 @@ from plone.app.redirector.interfaces import IRedirectionStorage except ImportError: IRedirectionStorage = None +from urllib.parse import quote, unquote + from plone.memoize.instance import memoize -from plone.rest.interfaces import IAPIRequest -from plone.rest.interfaces import ICORSPolicy from Products.CMFCore.permissions import ManagePortal from Products.Five.browser import BrowserView from six.moves import urllib -from six.moves.urllib.parse import quote -from six.moves.urllib.parse import unquote from zExceptions import NotFound +from plone.rest.interfaces import IAPIRequest, ICORSPolicy + try: from ZPublisher.HTTPRequest import WSGIRequest HAS_WSGI = True except ImportError: HAS_WSGI = False -from zope.component import adapter -from zope.component import queryMultiAdapter -from zope.component import queryUtility -from zope.component.hooks import getSite - import json -import six import sys import traceback +import six +from zope.component import adapter, queryMultiAdapter, queryUtility +from zope.component.hooks import getSite + @adapter(Exception, IAPIRequest) class ErrorHandling(BrowserView): @@ -61,7 +59,7 @@ def render_exception(self, exception): if six.PY2: name = name.decode("utf-8") message = message.decode("utf-8") - result = {u"type": name, u"message": message} + result = {"type": name, "message": message} policy = queryMultiAdapter((self.context, self.request), ICORSPolicy) if policy is not None: @@ -77,10 +75,10 @@ def render_exception(self, exception): # NotFound exceptions need special handling because their # exception message gets turned into HTML by ZPublisher url = self.request.getURL() - result[u"message"] = u"Resource not found: %s" % url + result["message"] = "Resource not found: %s" % url if getSecurityManager().checkPermission(ManagePortal, getSite()): - result[u"traceback"] = self.render_traceback(exception) + result["traceback"] = self.render_traceback(exception) return result @@ -101,8 +99,8 @@ def render_traceback(self, exception): pass else: return ( - u"ERROR: Another exception happened before we could " - u"render the traceback." + "ERROR: Another exception happened before we could " + "render the traceback." ) raw = "\n".join(traceback.format_tb(exc_traceback)) @@ -192,7 +190,7 @@ def attempt_redirect(self): query_string = self.request.QUERY_STRING if query_string: - new_path = storage.get("%s?%s" % (old_path, query_string)) + new_path = storage.get(f"{old_path}?{query_string}") # if we matched on the query_string we don't want to include it # in redirect if new_path: diff --git a/src/plone/rest/events.py b/src/plone/rest/events.py index 49554d8..a40b403 100644 --- a/src/plone/rest/events.py +++ b/src/plone/rest/events.py @@ -1,8 +1,8 @@ -# -*- coding: utf-8 -*- +from zope.interface import alsoProvides + from plone.rest.cors import lookup_preflight_service_id from plone.rest.interfaces import IAPIRequest from plone.rest.negotiation import lookup_service_id -from zope.interface import alsoProvides def mark_as_api_request(request, accept): diff --git a/src/plone/rest/interfaces.py b/src/plone/rest/interfaces.py index ac18224..bf07258 100644 --- a/src/plone/rest/interfaces.py +++ b/src/plone/rest/interfaces.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- from zope.interface import Interface diff --git a/src/plone/rest/negotiation.py b/src/plone/rest/negotiation.py index 7a47d5a..63f2f31 100644 --- a/src/plone/rest/negotiation.py +++ b/src/plone/rest/negotiation.py @@ -1,5 +1,3 @@ -# -*- coding: utf-8 -*- - # Service registry # A mapping of method -> type name -> subtype name -> service id _services = {} @@ -42,7 +40,7 @@ def register_service(method, media_type): """Register a service for the given request method and media type and return it's service id. """ - service_id = u"{}_{}_{}_".format(method, media_type[0], media_type[1]) + service_id = f"{method}_{media_type[0]}_{media_type[1]}_" types = _services.setdefault(method, {}) subtypes = types.setdefault(media_type[0], {}) subtypes[media_type[1]] = service_id diff --git a/src/plone/rest/patches.py b/src/plone/rest/patches.py index 97b618f..8f4cb5d 100644 --- a/src/plone/rest/patches.py +++ b/src/plone/rest/patches.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- from plone.rest.interfaces import IAPIRequest diff --git a/src/plone/rest/service.py b/src/plone/rest/service.py index 351b111..ebc3dbd 100644 --- a/src/plone/rest/service.py +++ b/src/plone/rest/service.py @@ -1,12 +1,11 @@ -# -*- coding: utf-8 -*- -from plone.rest.interfaces import ICORSPolicy -from plone.rest.interfaces import IService from zope.component import queryMultiAdapter from zope.interface import implementer +from plone.rest.interfaces import ICORSPolicy, IService + @implementer(IService) -class Service(object): +class Service: def __call__(self): policy = queryMultiAdapter((self.context, self.request), ICORSPolicy) if policy is not None: @@ -29,4 +28,4 @@ def __getattribute__(self, name): # include credentials if name == "__roles__" and self.request._rest_cors_preflight: return ["Anonymous"] - return super(Service, self).__getattribute__(name) + return super().__getattribute__(name) diff --git a/src/plone/rest/testing.py b/src/plone/rest/testing.py index 05268d9..4c6d203 100644 --- a/src/plone/rest/testing.py +++ b/src/plone/rest/testing.py @@ -1,16 +1,12 @@ -# -*- coding: utf-8 -*- from plone.app.contenttypes.testing import PLONE_APP_CONTENTTYPES_FIXTURE -from plone.app.testing import FunctionalTesting -from plone.app.testing import IntegrationTesting -from plone.app.testing import PloneSandboxLayer -from plone.rest.service import Service +from plone.app.testing import FunctionalTesting, IntegrationTesting, PloneSandboxLayer from plone.testing import z2 - from zope.configuration import xmlconfig +from plone.rest.service import Service -class PloneRestLayer(PloneSandboxLayer): +class PloneRestLayer(PloneSandboxLayer): defaultBases = (PLONE_APP_CONTENTTYPES_FIXTURE,) def setUpZope(self, app, configurationContext): @@ -31,7 +27,7 @@ def setUpZope(self, app, configurationContext): class InternalServerErrorService(Service): def __call__(self): - from six.moves.urllib.error import HTTPError + from urllib.error import HTTPError raise HTTPError( "http://nohost/plone/500-internal-server-error", diff --git a/src/plone/rest/tests/__init__.py b/src/plone/rest/tests/__init__.py index 40a96af..e69de29 100644 --- a/src/plone/rest/tests/__init__.py +++ b/src/plone/rest/tests/__init__.py @@ -1 +0,0 @@ -# -*- coding: utf-8 -*- diff --git a/src/plone/rest/tests/test_cors.py b/src/plone/rest/tests/test_cors.py index 964ea1d..b85cd6e 100644 --- a/src/plone/rest/tests/test_cors.py +++ b/src/plone/rest/tests/test_cors.py @@ -1,21 +1,19 @@ -# -*- coding: utf-8 -*- -from ZPublisher.pubevents import PubStart -from plone.app.testing import popGlobalRegistry -from plone.app.testing import pushGlobalRegistry -from plone.rest.cors import CORSPolicy -from plone.rest.interfaces import ICORSPolicy -from plone.rest.testing import PLONE_REST_INTEGRATION_TESTING +import unittest + +from plone.app.testing import popGlobalRegistry, pushGlobalRegistry from zExceptions import Unauthorized from zope.component import provideAdapter from zope.event import notify from zope.interface import Interface from zope.publisher.interfaces.browser import IDefaultBrowserLayer +from ZPublisher.pubevents import PubStart -import unittest +from plone.rest.cors import CORSPolicy +from plone.rest.interfaces import ICORSPolicy +from plone.rest.testing import PLONE_REST_INTEGRATION_TESTING class TestCORSPolicy(unittest.TestCase): - layer = PLONE_REST_INTEGRATION_TESTING def setUp(self): @@ -195,7 +193,6 @@ def test_preflight_cors_sets_status_code_200(self): class TestCORS(unittest.TestCase): - layer = PLONE_REST_INTEGRATION_TESTING def setUp(self): @@ -231,7 +228,7 @@ def test_simple_cors_gets_processed(self): def test_preflight_request_without_cors_policy_doesnt_render_service(self): # "Unregister" the current CORS policy - class NoCORSPolicy(object): + class NoCORSPolicy: def __new__(cls, context, request): return None diff --git a/src/plone/rest/tests/test_dexterity.py b/src/plone/rest/tests/test_dexterity.py index 288ae48..74b5574 100644 --- a/src/plone/rest/tests/test_dexterity.py +++ b/src/plone/rest/tests/test_dexterity.py @@ -1,25 +1,25 @@ -# -*- coding: utf-8 -*- +import os +import unittest from datetime import datetime -from plone.app.testing import setRoles -from plone.app.testing import TEST_USER_ID -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD + +import requests +import transaction +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + setRoles, +) from plone.app.textfield.value import RichTextValue -from plone.namedfile.file import NamedBlobImage -from plone.namedfile.file import NamedBlobFile -from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING +from plone.namedfile.file import NamedBlobFile, NamedBlobImage from z3c.relationfield import RelationValue from zope.component import getUtility from zope.intid.interfaces import IIntIds -import unittest -import os -import requests -import transaction +from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING class TestDexterityServiceEndpoints(unittest.TestCase): - layer = PLONE_REST_FUNCTIONAL_TESTING def setUp(self): @@ -39,8 +39,8 @@ def test_dexterity_document_get(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(200, response.status_code) - self.assertEqual(u"doc1", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("doc1", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_document_post(self): response = requests.post( @@ -49,8 +49,8 @@ def test_dexterity_document_post(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(200, response.status_code) - self.assertEqual(u"doc1", response.json().get("id")) - self.assertEqual(u"POST", response.json().get("method")) + self.assertEqual("doc1", response.json().get("id")) + self.assertEqual("POST", response.json().get("method")) def test_dexterity_document_put(self): response = requests.put( @@ -59,8 +59,8 @@ def test_dexterity_document_put(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(200, response.status_code) - self.assertEqual(u"doc1", response.json().get("id")) - self.assertEqual(u"PUT", response.json().get("method")) + self.assertEqual("doc1", response.json().get("id")) + self.assertEqual("PUT", response.json().get("method")) def test_dexterity_document_patch(self): response = requests.patch( @@ -69,8 +69,8 @@ def test_dexterity_document_patch(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(200, response.status_code) - self.assertEqual(u"doc1", response.json().get("id")) - self.assertEqual(u"PATCH", response.json().get("method")) + self.assertEqual("doc1", response.json().get("id")) + self.assertEqual("PATCH", response.json().get("method")) def test_dexterity_document_delete(self): response = requests.delete( @@ -79,8 +79,8 @@ def test_dexterity_document_delete(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(200, response.status_code) - self.assertEqual(u"doc1", response.json().get("id")) - self.assertEqual(u"DELETE", response.json().get("method")) + self.assertEqual("doc1", response.json().get("id")) + self.assertEqual("DELETE", response.json().get("method")) def test_dexterity_document_options(self): response = requests.options( @@ -89,8 +89,8 @@ def test_dexterity_document_options(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(200, response.status_code) - self.assertEqual(u"doc1", response.json().get("id")) - self.assertEqual(u"OPTIONS", response.json().get("method")) + self.assertEqual("doc1", response.json().get("id")) + self.assertEqual("OPTIONS", response.json().get("method")) def test_dexterity_folder_get(self): self.portal.invokeFactory("Folder", id="folder") @@ -103,23 +103,23 @@ def test_dexterity_folder_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"folder", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("folder", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_news_item_get(self): self.portal.invokeFactory("News Item", id="newsitem") self.portal.newsitem.title = "My News Item" - self.portal.newsitem.description = u"This is a news item" + self.portal.newsitem.description = "This is a news item" self.portal.newsitem.text = RichTextValue( - u"Lorem ipsum", "text/plain", "text/html" + "Lorem ipsum", "text/plain", "text/html" ) - image_file = os.path.join(os.path.dirname(__file__), u"image.png") + image_file = os.path.join(os.path.dirname(__file__), "image.png") fd = open(image_file, "rb") self.portal.newsitem.image = NamedBlobImage( - data=fd.read(), contentType="image/png", filename=u"image.png" + data=fd.read(), contentType="image/png", filename="image.png" ) fd.close() - self.portal.newsitem.image_caption = u"This is an image caption." + self.portal.newsitem.image_caption = "This is an image caption." import transaction transaction.commit() @@ -130,13 +130,13 @@ def test_dexterity_news_item_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"newsitem", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("newsitem", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_event_get(self): self.portal.invokeFactory("Event", id="event") self.portal.event.title = "Event" - self.portal.event.description = u"This is an event" + self.portal.event.description = "This is an event" self.portal.event.start = datetime(2013, 1, 1, 10, 0) self.portal.event.end = datetime(2013, 1, 1, 12, 0) import transaction @@ -149,13 +149,13 @@ def test_dexterity_event_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"event", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("event", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_link_get(self): self.portal.invokeFactory("Link", id="link") self.portal.link.title = "My Link" - self.portal.link.description = u"This is a link" + self.portal.link.description = "This is a link" self.portal.remoteUrl = "http://plone.org" import transaction @@ -167,17 +167,17 @@ def test_dexterity_link_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"link", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("link", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_file_get(self): self.portal.invokeFactory("File", id="file") self.portal.file.title = "My File" - self.portal.file.description = u"This is a file" - pdf_file = os.path.join(os.path.dirname(__file__), u"file.pdf") + self.portal.file.description = "This is a file" + pdf_file = os.path.join(os.path.dirname(__file__), "file.pdf") fd = open(pdf_file, "rb") self.portal.file.file = NamedBlobFile( - data=fd.read(), contentType="application/pdf", filename=u"file.pdf" + data=fd.read(), contentType="application/pdf", filename="file.pdf" ) fd.close() intids = getUtility(IIntIds) @@ -194,17 +194,17 @@ def test_dexterity_file_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"file", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("file", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_image_get(self): self.portal.invokeFactory("Image", id="image") self.portal.image.title = "My Image" - self.portal.image.description = u"This is an image" - image_file = os.path.join(os.path.dirname(__file__), u"image.png") + self.portal.image.description = "This is an image" + image_file = os.path.join(os.path.dirname(__file__), "image.png") fd = open(image_file, "rb") self.portal.image.image = NamedBlobImage( - data=fd.read(), contentType="image/png", filename=u"image.png" + data=fd.read(), contentType="image/png", filename="image.png" ) fd.close() import transaction @@ -218,13 +218,13 @@ def test_dexterity_image_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"image", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("image", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_dexterity_collection_get(self): self.portal.invokeFactory("Collection", id="collection") self.portal.collection.title = "My Collection" - self.portal.collection.description = u"This is a collection with two documents" + self.portal.collection.description = "This is a collection with two documents" self.portal.collection.query = [ { "i": "portal_type", @@ -243,5 +243,5 @@ def test_dexterity_collection_get(self): ) self.assertEqual(200, response.status_code) - self.assertEqual(u"collection", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("collection", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) diff --git a/src/plone/rest/tests/test_dispatching.py b/src/plone/rest/tests/test_dispatching.py index fb9b7b8..f7ab6ce 100644 --- a/src/plone/rest/tests/test_dispatching.py +++ b/src/plone/rest/tests/test_dispatching.py @@ -1,15 +1,16 @@ -# -*- coding: utf-8 -*- -from plone.app.testing import setRoles -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD -from plone.app.testing import TEST_USER_ID -from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING -from Products.CMFCore.utils import getToolByName +import unittest import requests import transaction -import unittest +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + setRoles, +) +from Products.CMFCore.utils import getToolByName +from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING CREDS = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD) INVALID_CREDS = ("invalid", "password") @@ -17,7 +18,6 @@ class DispatchingTestCase(unittest.TestCase): - layer = PLONE_REST_FUNCTIONAL_TESTING def setUp(self): @@ -54,7 +54,7 @@ def validate(self, expectations, follow_redirects=False): if failures: msg = "" - for (request_args, expected_status, actual_status) in failures: + for request_args, expected_status, actual_status in failures: msg += ( "\n" "Request: %s\n" @@ -137,7 +137,7 @@ def test_not_found_invalid_creds(self): class TestDispatchingDexterity(DispatchingTestCase): def setUp(self): - super(TestDispatchingDexterity, self).setUp() + super().setUp() self.portal.invokeFactory("Folder", id="private") self.portal.invokeFactory("Folder", id="public") @@ -216,7 +216,7 @@ def test_public_dx_folder_invalid_creds(self): class TestDispatchingRedirects(DispatchingTestCase): def setUp(self): - super(TestDispatchingRedirects, self).setUp() + super().setUp() self.portal.invokeFactory("Folder", id="private-old") self.portal.manage_renameObject("private-old", "private-new") diff --git a/src/plone/rest/tests/test_error_handling.py b/src/plone/rest/tests/test_error_handling.py index b2cf511..61cb2b5 100644 --- a/src/plone/rest/tests/test_error_handling.py +++ b/src/plone/rest/tests/test_error_handling.py @@ -1,19 +1,20 @@ -# -*- coding: utf-8 -*- -from plone.app.testing import setRoles -from plone.app.testing import TEST_USER_ID -from plone.app.testing import TEST_USER_PASSWORD -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD -from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING - import json +import unittest + import requests import transaction -import unittest +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + TEST_USER_PASSWORD, + setRoles, +) + +from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING class TestErrorHandling(unittest.TestCase): - layer = PLONE_REST_FUNCTIONAL_TESTING def setUp(self): @@ -83,7 +84,7 @@ def test_500_internal_server_error(self): self.assertEqual("HTTPError", response.json()["type"]) self.assertEqual( - {u"type": u"HTTPError", u"message": u"HTTP Error 500: InternalServerError"}, + {"type": "HTTPError", "message": "HTTP Error 500: InternalServerError"}, response.json(), ) @@ -94,7 +95,7 @@ def test_500_traceback_only_for_manager_users(self): headers={"Accept": "application/json"}, auth=(TEST_USER_ID, TEST_USER_PASSWORD), ) - self.assertNotIn(u"traceback", response.json()) + self.assertNotIn("traceback", response.json()) # Manager user response = requests.get( @@ -102,10 +103,10 @@ def test_500_traceback_only_for_manager_users(self): headers={"Accept": "application/json"}, auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) - self.assertIn(u"traceback", response.json()) + self.assertIn("traceback", response.json()) - traceback = response.json()[u"traceback"] + traceback = response.json()["traceback"] self.assertIsInstance(traceback, list) - self.assertRegexpMatches( + self.assertRegex( traceback[0], r'^File "[^"]*", line \d*, in (publish|transaction_pubevents)' ) diff --git a/src/plone/rest/tests/test_named_services.py b/src/plone/rest/tests/test_named_services.py index fe1977b..044042f 100644 --- a/src/plone/rest/tests/test_named_services.py +++ b/src/plone/rest/tests/test_named_services.py @@ -1,17 +1,18 @@ -# -*- coding: utf-8 -*- -from plone.app.testing import setRoles -from plone.app.testing import TEST_USER_ID -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD -from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING - import unittest + import requests import transaction +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + setRoles, +) +from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING -class TestNamedServiceEndpoints(unittest.TestCase): +class TestNamedServiceEndpoints(unittest.TestCase): layer = PLONE_REST_FUNCTIONAL_TESTING def setUp(self): @@ -31,7 +32,7 @@ def test_dexterity_named_get(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual({u"service": u"named get"}, response.json()) + self.assertEqual({"service": "named get"}, response.json()) def test_dexterity_named_post(self): response = requests.post( @@ -40,7 +41,7 @@ def test_dexterity_named_post(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual({u"service": u"named post"}, response.json()) + self.assertEqual({"service": "named post"}, response.json()) def test_dexterity_named_put(self): response = requests.put( @@ -49,7 +50,7 @@ def test_dexterity_named_put(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual({u"service": u"named put"}, response.json()) + self.assertEqual({"service": "named put"}, response.json()) def test_dexterity_named_patch(self): response = requests.patch( @@ -58,7 +59,7 @@ def test_dexterity_named_patch(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual({u"service": u"named patch"}, response.json()) + self.assertEqual({"service": "named patch"}, response.json()) def test_dexterity_named_delete(self): response = requests.delete( @@ -67,7 +68,7 @@ def test_dexterity_named_delete(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual({u"service": u"named delete"}, response.json()) + self.assertEqual({"service": "named delete"}, response.json()) def test_dexterity_named_options(self): response = requests.options( @@ -76,4 +77,4 @@ def test_dexterity_named_options(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual({u"service": u"named options"}, response.json()) + self.assertEqual({"service": "named options"}, response.json()) diff --git a/src/plone/rest/tests/test_negotiation.py b/src/plone/rest/tests/test_negotiation.py index 70edd5a..f4a979c 100644 --- a/src/plone/rest/tests/test_negotiation.py +++ b/src/plone/rest/tests/test_negotiation.py @@ -1,10 +1,11 @@ -# -*- coding: utf-8 -*- -from plone.rest.negotiation import lookup_service_id -from plone.rest.negotiation import parse_accept_header -from plone.rest.negotiation import register_service - import unittest +from plone.rest.negotiation import ( + lookup_service_id, + parse_accept_header, + register_service, +) + class TestAcceptHeaderParser(unittest.TestCase): def test_parse_application_json_accept_header(self): @@ -61,24 +62,24 @@ def test_parse_invalid_accept_header(self): class TestServiceRegistry(unittest.TestCase): def test_register_media_type(self): self.assertEqual( - u"GET_application_json_", register_service("GET", ("application", "json")) + "GET_application_json_", register_service("GET", ("application", "json")) ) self.assertEqual( - u"GET_application_json_", lookup_service_id("GET", "application/json") + "GET_application_json_", lookup_service_id("GET", "application/json") ) def test_register_wildcard_subtype(self): - self.assertEqual(u"PATCH_text_*_", register_service("PATCH", ("text", "*"))) - self.assertEqual(u"PATCH_text_*_", lookup_service_id("PATCH", "text/xml")) + self.assertEqual("PATCH_text_*_", register_service("PATCH", ("text", "*"))) + self.assertEqual("PATCH_text_*_", lookup_service_id("PATCH", "text/xml")) def test_register_wilcard_type(self): - self.assertEqual(u"PATCH_*_*_", register_service("PATCH", ("*", "*"))) - self.assertEqual(u"PATCH_*_*_", lookup_service_id("PATCH", "foo/bar")) + self.assertEqual("PATCH_*_*_", register_service("PATCH", ("*", "*"))) + self.assertEqual("PATCH_*_*_", lookup_service_id("PATCH", "foo/bar")) def test_service_id_for_multiple_media_types_is_none(self): register_service("GET", "application/json") self.assertEqual( - None, lookup_service_id("GET", "application/json,application/javascipt") + None, lookup_service_id("GET", "application/json,application/javascript") ) def test_service_id_for_invalid_media_type_is_none(self): diff --git a/src/plone/rest/tests/test_permissions.py b/src/plone/rest/tests/test_permissions.py index 2ab0df6..7a11bef 100644 --- a/src/plone/rest/tests/test_permissions.py +++ b/src/plone/rest/tests/test_permissions.py @@ -1,23 +1,24 @@ -# -*- coding: utf-8 -*- -from Products.CMFCore.utils import getToolByName -from ZPublisher.pubevents import PubStart +import unittest from base64 import b64encode -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import TEST_USER_ID -from plone.app.testing import TEST_USER_NAME -from plone.app.testing import TEST_USER_PASSWORD -from plone.app.testing import login -from plone.app.testing import setRoles -from plone.rest.service import Service -from plone.rest.testing import PLONE_REST_INTEGRATION_TESTING + +from plone.app.testing import ( + SITE_OWNER_NAME, + TEST_USER_ID, + TEST_USER_NAME, + TEST_USER_PASSWORD, + login, + setRoles, +) +from Products.CMFCore.utils import getToolByName from zExceptions import Unauthorized from zope.event import notify +from ZPublisher.pubevents import PubStart -import unittest +from plone.rest.service import Service +from plone.rest.testing import PLONE_REST_INTEGRATION_TESTING class TestPermissions(unittest.TestCase): - layer = PLONE_REST_INTEGRATION_TESTING def setUp(self): @@ -36,7 +37,7 @@ def traverse(self, path="/plone", accept="application/json", method="GET"): request.environ["PATH_TRANSLATED"] = path request.environ["HTTP_ACCEPT"] = accept request.environ["REQUEST_METHOD"] = method - auth = "%s:%s" % (TEST_USER_NAME, TEST_USER_PASSWORD) + auth = f"{TEST_USER_NAME}:{TEST_USER_PASSWORD}" b64auth = b64encode(auth.encode("utf8")) request._auth = "Basic %s" % b64auth.decode("utf8") notify(PubStart(request)) diff --git a/src/plone/rest/tests/test_redirects.py b/src/plone/rest/tests/test_redirects.py index 30122da..2096594 100644 --- a/src/plone/rest/tests/test_redirects.py +++ b/src/plone/rest/tests/test_redirects.py @@ -1,21 +1,22 @@ -# -*- coding: utf-8 -*- +import unittest + +import requests +import transaction from BTrees.OOBTree import OOSet from plone.app.redirector.interfaces import IRedirectionStorage -from plone.app.testing import setRoles -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD -from plone.app.testing import TEST_USER_ID -from plone.rest.errors import ErrorHandling -from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + setRoles, +) from zope.component import queryUtility -import requests -import transaction -import unittest +from plone.rest.errors import ErrorHandling +from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING class TestRedirects(unittest.TestCase): - layer = PLONE_REST_FUNCTIONAL_TESTING def setUp(self): diff --git a/src/plone/rest/tests/test_siteroot.py b/src/plone/rest/tests/test_siteroot.py index 6cf71d5..3bcdc79 100644 --- a/src/plone/rest/tests/test_siteroot.py +++ b/src/plone/rest/tests/test_siteroot.py @@ -1,16 +1,17 @@ -# -*- coding: utf-8 -*- -from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING -from plone.app.testing import setRoles -from plone.app.testing import TEST_USER_ID -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD - import unittest + import requests +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + setRoles, +) +from plone.rest.testing import PLONE_REST_FUNCTIONAL_TESTING -class TestSiteRootServiceEndpoints(unittest.TestCase): +class TestSiteRootServiceEndpoints(unittest.TestCase): layer = PLONE_REST_FUNCTIONAL_TESTING def setUp(self): @@ -33,8 +34,8 @@ def test_siteroot_get(self): response.status_code ), ) - self.assertEqual(u"plone", response.json().get("id")) - self.assertEqual(u"GET", response.json().get("method")) + self.assertEqual("plone", response.json().get("id")) + self.assertEqual("GET", response.json().get("method")) def test_siteroot_post(self): response = requests.post( @@ -49,8 +50,8 @@ def test_siteroot_post(self): response.status_code ), ) - self.assertEqual(u"plone", response.json().get("id")) - self.assertEqual(u"POST", response.json().get("method")) + self.assertEqual("plone", response.json().get("id")) + self.assertEqual("POST", response.json().get("method")) def test_siteroot_delete(self): response = requests.delete( @@ -59,8 +60,8 @@ def test_siteroot_delete(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual(u"plone", response.json().get("id")) - self.assertEqual(u"DELETE", response.json().get("method")) + self.assertEqual("plone", response.json().get("id")) + self.assertEqual("DELETE", response.json().get("method")) def test_siteroot_put(self): response = requests.put( @@ -70,8 +71,8 @@ def test_siteroot_put(self): ) self.assertEqual(response.status_code, 200) - self.assertEqual(u"plone", response.json().get("id")) - self.assertEqual(u"PUT", response.json().get("method")) + self.assertEqual("plone", response.json().get("id")) + self.assertEqual("PUT", response.json().get("method")) def test_siteroot_patch(self): response = requests.patch( @@ -81,8 +82,8 @@ def test_siteroot_patch(self): ) self.assertEqual(response.status_code, 200) - self.assertEqual(u"plone", response.json().get("id")) - self.assertEqual(u"PATCH", response.json().get("method")) + self.assertEqual("plone", response.json().get("id")) + self.assertEqual("PATCH", response.json().get("method")) def test_siteroot_options(self): response = requests.options( @@ -91,5 +92,5 @@ def test_siteroot_options(self): auth=(SITE_OWNER_NAME, SITE_OWNER_PASSWORD), ) self.assertEqual(response.status_code, 200) - self.assertEqual(u"plone", response.json().get("id")) - self.assertEqual(u"OPTIONS", response.json().get("method")) + self.assertEqual("plone", response.json().get("id")) + self.assertEqual("OPTIONS", response.json().get("method")) diff --git a/src/plone/rest/tests/test_traversal.py b/src/plone/rest/tests/test_traversal.py index 5d5389d..26c5816 100644 --- a/src/plone/rest/tests/test_traversal.py +++ b/src/plone/rest/tests/test_traversal.py @@ -1,26 +1,26 @@ -# -*- coding: utf-8 -*- -from Products.SiteAccess.VirtualHostMonster import VirtualHostMonster -from ZPublisher import BeforeTraverse -from ZPublisher.pubevents import PubStart +import unittest from base64 import b64encode + from plone.app.layout.navigation.interfaces import INavigationRoot -from plone.app.testing import setRoles -from plone.app.testing import SITE_OWNER_NAME -from plone.app.testing import SITE_OWNER_PASSWORD -from plone.app.testing import TEST_USER_ID -from plone.rest.service import Service -from plone.rest.testing import PLONE_REST_INTEGRATION_TESTING -from zExceptions import NotFound -from zExceptions import Redirect +from plone.app.testing import ( + SITE_OWNER_NAME, + SITE_OWNER_PASSWORD, + TEST_USER_ID, + setRoles, +) +from Products.SiteAccess.VirtualHostMonster import VirtualHostMonster +from zExceptions import NotFound, Redirect from zope.event import notify from zope.interface import alsoProvides from zope.publisher.interfaces.browser import IBrowserView +from ZPublisher import BeforeTraverse +from ZPublisher.pubevents import PubStart -import unittest +from plone.rest.service import Service +from plone.rest.testing import PLONE_REST_INTEGRATION_TESTING class TestTraversal(unittest.TestCase): - layer = PLONE_REST_INTEGRATION_TESTING def setUp(self): @@ -34,7 +34,7 @@ def traverse(self, path="/plone", accept="application/json", method="GET"): request.environ["PATH_TRANSLATED"] = path request.environ["HTTP_ACCEPT"] = accept request.environ["REQUEST_METHOD"] = method - auth = "%s:%s" % (SITE_OWNER_NAME, SITE_OWNER_PASSWORD) + auth = f"{SITE_OWNER_NAME}:{SITE_OWNER_PASSWORD}" b64auth = b64encode(auth.encode("utf8")) request._auth = "Basic %s" % b64auth.decode("utf8") notify(PubStart(request)) diff --git a/src/plone/rest/traverse.py b/src/plone/rest/traverse.py index 0a151c8..393b00e 100644 --- a/src/plone/rest/traverse.py +++ b/src/plone/rest/traverse.py @@ -1,24 +1,21 @@ -# -*- coding: utf-8 -*- -from Products.CMFCore.interfaces import ISiteRoot +from Products.CMFCore.interfaces import IContentish, ISiteRoot from Products.SiteAccess.VirtualHostMonster import VirtualHostMonster -from ZPublisher.BaseRequest import DefaultPublishTraverse -from plone.rest.interfaces import IAPIRequest -from plone.rest.interfaces import IService -from plone.rest.events import mark_as_api_request from zExceptions import Redirect -from zope.component import adapter -from zope.component import queryMultiAdapter +from zope.component import adapter, queryMultiAdapter from zope.interface import implementer from zope.publisher.interfaces.browser import IBrowserPublisher from zope.traversing.interfaces import ITraversable -from Products.CMFCore.interfaces import IContentish +from ZPublisher.BaseRequest import DefaultPublishTraverse + +from plone.rest.events import mark_as_api_request +from plone.rest.interfaces import IAPIRequest, IService @adapter(ISiteRoot, IAPIRequest) class RESTTraverse(DefaultPublishTraverse): def publishTraverse(self, request, name): try: - obj = super(RESTTraverse, self).publishTraverse(request, name) + obj = super().publishTraverse(request, name) if not IContentish.providedBy(obj) and not IService.providedBy(obj): if isinstance(obj, VirtualHostMonster): return obj @@ -54,7 +51,7 @@ def browserDefault(self, request): @implementer(ITraversable) -class MarkAsRESTTraverser(object): +class MarkAsRESTTraverser: """ Traversal adapter for the ``++api++`` namespace. It marks the request as API request. @@ -82,7 +79,7 @@ def traverse(self, name_ignored, subpath_ignored): @implementer(IBrowserPublisher) -class RESTWrapper(object): +class RESTWrapper: """A wrapper for objects traversed during a REST request.""" def __init__(self, context, request): diff --git a/src/plone/rest/zcml.py b/src/plone/rest/zcml.py index 3e33602..7ac81c7 100644 --- a/src/plone/rest/zcml.py +++ b/src/plone/rest/zcml.py @@ -1,70 +1,65 @@ -# -*- coding: utf-8 -*- from AccessControl.class_init import InitializeClass -from AccessControl.security import getSecurityInfo -from AccessControl.security import protectClass +from AccessControl.security import getSecurityInfo, protectClass from Products.Five.browser import BrowserView -from plone.rest.cors import CORSPolicy -from plone.rest.cors import register_method_for_preflight -from plone.rest.interfaces import ICORSPolicy -from plone.rest.negotiation import parse_accept_header -from plone.rest.negotiation import register_service from zope.browserpage.metaconfigure import _handle_for from zope.component.zcml import handler -from zope.configuration.fields import GlobalInterface -from zope.configuration.fields import GlobalObject +from zope.configuration.fields import GlobalInterface, GlobalObject from zope.interface import Interface from zope.publisher.interfaces.browser import IDefaultBrowserLayer -from zope.schema import Bool -from zope.schema import TextLine +from zope.schema import Bool, TextLine from zope.security.zcml import Permission +from plone.rest.cors import CORSPolicy, register_method_for_preflight +from plone.rest.interfaces import ICORSPolicy +from plone.rest.negotiation import parse_accept_header, register_service + class IService(Interface): """ """ method = TextLine( - title=u"The name of the view that should be the default. " - + u"[get|post|put|delete]", - description=u""" + title="The name of the view that should be the default. " + + "[get|post|put|delete]", + description=""" This name refers to view that should be the view used by default (if no view name is supplied explicitly).""", ) accept = TextLine( - title=u"Acceptable media types", - description=u"""Specifies the media type used for content negotiation. + title="Acceptable media types", + description="""Specifies the media type used for content negotiation. The service is limited to the given media type and only called if the request contains an "Accept" header with the given media type. Multiple media types can be given by separating them with a comma.""", - default=u"application/json", + default="application/json", ) for_ = GlobalObject( - title=u"The interface this view is the default for.", - description=u"""Specifies the interface for which the view is + title="The interface this view is the default for.", + description="""Specifies the interface for which the view is registered. All objects implementing this interface can make use of this view. If this attribute is not specified, the view is available for all objects.""", ) factory = GlobalObject( - title=u"The factory for this service", - description=u"The factory is usually subclass of the Service class.", + title="The factory for this service", + description="The factory is usually subclass of the Service class.", ) name = TextLine( - title=u"The name of the service.", - description=u"""When no name is defined, the service is available at + title="The name of the service.", + description="""When no name is defined, the service is available at the object's absolute URL. When defining a name, the service is available at the object's absolute URL appended with a slash and the service name.""", required=False, - default=u"", + default="", ) layer = GlobalInterface( - title=u"The browser layer for which this service is registered.", - description=u"""Useful for overriding existing services or for making + title="The browser layer for which this service is registered.", + description="""Useful for overriding existing services or for making services available only if a specific add-on has been installed.""", required=False, @@ -72,8 +67,8 @@ class IService(Interface): ) permission = Permission( - title=u"Permission", - description=u"The permission needed to access the service.", + title="Permission", + description="The permission needed to access the service.", required=True, ) @@ -86,9 +81,8 @@ def serviceDirective( for_, permission, layer=IDefaultBrowserLayer, - name=u"", + name="", ): - _handle_for(_context, for_) media_types = parse_accept_header(accept) @@ -137,16 +131,16 @@ class ICORSPolicyDirective(Interface): """Directive for defining CORS policies""" for_ = GlobalObject( - title=u"The interface this CORS policy is for.", - description=u"""Specifies the interface for which the CORS policy is + title="The interface this CORS policy is for.", + description="""Specifies the interface for which the CORS policy is registered. If this attribute is not specified, the CORS policy applies to all objects.""", required=False, ) layer = GlobalInterface( - title=u"The browser layer for which this CORS policy is registered.", - description=u"""Useful for overriding existing policies or for making + title="The browser layer for which this CORS policy is registered.", + description="""Useful for overriding existing policies or for making them available only if a specific add-on has been installed.""", required=False, @@ -154,45 +148,45 @@ class ICORSPolicyDirective(Interface): ) allow_origin = TextLine( - title=u"Origins", - description=u"""Origins that are allowed access to the resource. Either + title="Origins", + description="""Origins that are allowed access to the resource. Either a comma separated list of origins, e.g. "http://example.net, http://mydomain.com" or "*".""", ) allow_methods = TextLine( - title=u"Methods", - description=u"""A comma separated list of HTTP method names that are + title="Methods", + description="""A comma separated list of HTTP method names that are allowed by this CORS policy, e.g. "DELETE,GET,OPTIONS,PATCH,POST,PUT". """, required=False, ) allow_headers = TextLine( - title=u"Headers", - description=u"""A comma separated list of request headers allowed to be + title="Headers", + description="""A comma separated list of request headers allowed to be sent by the client, e.g. "X-My-Header".""", required=False, ) expose_headers = TextLine( - title=u"Exposed Headers", - description=u"""A comma separated list of response headers clients can + title="Exposed Headers", + description="""A comma separated list of response headers clients can access, e.g. "Content-Length,X-My-Header".""", required=False, ) allow_credentials = Bool( - title=u"Support Credentials", - description=u"""Indicates whether the resource supports user + title="Support Credentials", + description="""Indicates whether the resource supports user credentials in the request.""", required=True, default=False, ) max_age = TextLine( - title=u"Max Age", - description=u"""Indicates how long the results of a preflight request + title="Max Age", + description="""Indicates how long the results of a preflight request can be cached.""", required=False, ) @@ -209,7 +203,6 @@ def cors_policy_directive( for_=Interface, layer=IDefaultBrowserLayer, ): - _handle_for(_context, for_) # Create a new policy class and store the CORS policy configuration in @@ -240,7 +233,7 @@ def cors_policy_directive( new_class, (for_, layer), ICORSPolicy, - u"", + "", _context.info, ), )