Enforce private instantiation and inheritance through metaclass

newsfragments/1021.misc.rst

@@ -0,0 +1,3 @@
+Inheriting from :exc:`trio.Cancelled` exception and :class:`CancelScope` is now blocked.
+Instantiating from :exc:`trio.Cancelled` has been blocked as well. Any of these actions will
+now raise a :exc:`TypeError`.

trio/_core/_exceptions.py

@@ -1,5 +1,7 @@
 import attr
 
+from trio._util import NoPublicConstructor
+
 
 class TrioInternalError(Exception):
     """Raised by :func:`run` if we encounter a bug in trio, or (possibly) a
@@ -31,7 +33,7 @@ class WouldBlock(Exception):
     pass
 
 
-class Cancelled(BaseException):
+class Cancelled(BaseException, metaclass=NoPublicConstructor):
     """Raised by blocking calls if the surrounding scope has been cancelled.
 
     You should let this exception propagate, to be caught by the relevant
@@ -47,7 +49,7 @@ class Cancelled(BaseException):
     then this *won't* catch a :exc:`Cancelled` exception.
 
     You cannot raise :exc:`Cancelled` yourself. Attempting to do so
-    will produce a :exc:`RuntimeError`. Use :meth:`cancel_scope.cancel()
+    will produce a :exc:`TypeError`. Use :meth:`cancel_scope.cancel()
     <trio.CancelScope.cancel>` instead.
 
     .. note::
@@ -63,27 +65,10 @@ class Cancelled(BaseException):
        everywhere.
 
     """
-    __marker = object()
-
-    def __init__(self, _marker=None):
-        if _marker is not self.__marker:
-            raise RuntimeError(
-                'Cancelled should not be raised directly. Use the cancel() '
-                'method on your cancel scope.'
-            )
-        super().__init__()
 
     def __str__(self):
         return "Cancelled"
 
-    @classmethod
-    def _init(cls):
-        """A private constructor so that a user-created instance of Cancelled
-        can raise an appropriate error. see `issue #342
-        <https://github.com/python-trio/trio/issues/342>`__.
-        """
-        return cls(_marker=cls.__marker)
-
 
 class BusyResourceError(Exception):
     """Raised when a task attempts to use a resource that some other task is

trio/_core/_run.py

@@ -37,6 +37,7 @@
 )
 from .. import _core
 from .._deprecate import deprecated
+from .._util import Final
 
 # At the bottom of this file there's also some "clever" code that generates
 # wrapper functions for runner and io manager methods, and adds them to
@@ -125,7 +126,7 @@ def deadline_to_sleep_time(self, deadline):
 
 
 @attr.s(cmp=False, repr=False, slots=True)
-class CancelScope:
+class CancelScope(metaclass=Final):
     """A *cancellation scope*: the link between a unit of cancellable
     work and Trio's cancellation system.
 
@@ -737,7 +738,7 @@ def _attempt_delivery_of_any_pending_cancel(self):
             return
 
         def raise_cancel():
-            raise Cancelled._init()
+            raise Cancelled._create()
 
         self._attempt_abort(raise_cancel)
 

trio/_core/tests/test_run.py

@@ -2057,7 +2057,7 @@ async def inner():
 
 def test_Cancelled_init():
     check_Cancelled_error = pytest.raises(
-        RuntimeError, match='should not be raised directly'
+        TypeError, match='no public constructor available'
     )
 
     with check_Cancelled_error:
@@ -2067,14 +2067,32 @@ def test_Cancelled_init():
         _core.Cancelled()
 
     # private constructor should not raise
-    _core.Cancelled._init()
+    _core.Cancelled._create()
 
 
 def test_Cancelled_str():
-    cancelled = _core.Cancelled._init()
+    cancelled = _core.Cancelled._create()
     assert str(cancelled) == 'Cancelled'
 
 
+def test_Cancelled_subclass():
+    with pytest.raises(
+        TypeError, match='Cancelled does not support subclassing'
+    ):
+
+        class Subclass(_core.Cancelled):
+            pass
+
+
+def test_CancelScope_subclass():
+    with pytest.raises(
+        TypeError, match='CancelScope does not support subclassing'
+    ):
+
+        class Subclass(_core.CancelScope):
+            pass
+
+
 def test_sniffio_integration():
     with pytest.raises(sniffio.AsyncLibraryNotFoundError):
         sniffio.current_async_library()

trio/_util.py

@@ -280,3 +280,53 @@ def __call__(self, *args, **kwargs):
 
     def __getitem__(self, _):
         return self
+
+
+class Final(type):
+    """Metaclass that enforces a class to be final (i.e., subclass not allowed).
+
+    If a class uses this metaclass like this::
+
+        class SomeClass(metaclass=Final):
+            pass
+
+    The metaclass will ensure that no sub class can be created.
+
+    Raises
+    ------
+    - TypeError if a sub class is created
+    """
+
+    def __new__(cls, name, bases, cls_namespace):
+        for base in bases:
+            if isinstance(base, Final):
+                raise TypeError(
+                    f"{base.__name__} does not support subclassing"
+                )
+        return type.__new__(cls, name, bases, cls_namespace)
+
+
+class NoPublicConstructor(Final):
+    """Metaclass that enforces a class to be final (i.e., subclass not allowed)
+    and ensures a private constructor.
+
+    If a class uses this metaclass like this::
+
+        class SomeClass(metaclass=NoPublicConstructor):
+            pass
+
+    The metaclass will ensure that no sub class can be created, and that no instance
+    can be initialized.
+
+    If you try to instantiate your class (SomeClass()), a TypeError will be thrown.
+
+    Raises
+    ------
+    - TypeError if a sub class or an instance is created.
+    """
+
+    def __call__(self, *args, **kwargs):
+        raise TypeError("no public constructor available")
+
+    def _create(self, *args, **kwargs):
+        return super().__call__(*args, **kwargs)

trio/tests/test_util.py

@@ -8,7 +8,8 @@
 from .. import _core
 from trio import run_sync_in_worker_thread
 from .._util import (
-    signal_raise, ConflictDetector, fspath, is_main_thread, generic_function
+    signal_raise, ConflictDetector, fspath, is_main_thread, generic_function,
+    Final, NoPublicConstructor
 )
 from ..testing import wait_all_tasks_blocked, assert_checkpoints
 
@@ -184,3 +185,33 @@ def test_func(arg):
     assert test_func.__qualname__ == "test_generic_function.<locals>.test_func"
     assert test_func.__name__ == "test_func"
     assert test_func.__module__ == __name__
+
+
+def test_final_metaclass():
+    class FinalClass(metaclass=Final):
+        pass
+
+    with pytest.raises(
+        TypeError, match="FinalClass does not support subclassing"
+    ):
+
+        class SubClass(FinalClass):
+            pass
+
+
+def test_no_public_constructor_metaclass():
+    class SpecialClass(metaclass=NoPublicConstructor):
+        pass
+
+    with pytest.raises(TypeError, match="no public constructor available"):
+        SpecialClass()
+
+    with pytest.raises(
+        TypeError, match="SpecialClass does not support subclassing"
+    ):
+
+        class SubClass(SpecialClass):
+            pass
+
+    # Private constructor should not raise
+    SpecialClass._create()