One Hat Cyber Team

Your IP: 216.73.216.144
Server IP: 157.245.143.252
Server: Linux www 6.11.0-9-generic #9-Ubuntu SMP PREEMPT_DYNAMIC Mon Oct 14 13:19:59 UTC 2024 x86_64
Server Software: nginx/1.26.0
PHP Version: 8.3.11
Buat File | Buat Folder

Dir : ~/lib/python3/dist-packages/twisted/internet/test/

View File Name : test_kqueuereactor.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.internet.kqueuereactor}.
"""
from __future__ import annotations

import errno

from zope.interface import implementer

from twisted.trial.unittest import TestCase

try:
    from twisted.internet.kqreactor import KQueueReactor, _IKQueue

    kqueueSkip = None
except ImportError:
    kqueueSkip = "KQueue not available."


def _fakeKEvent(*args: object, **kwargs: object) -> None:
    """
    Do nothing.
    """


def makeFakeKQueue(testKQueue: object, testKEvent: object) -> _IKQueue:
    """
    Create a fake that implements L{_IKQueue}.

    @param testKQueue: Something that acts like L{select.kqueue}.
    @param testKEvent: Something that acts like L{select.kevent}.
    @return: An implementation of L{_IKQueue} that includes C{testKQueue} and
        C{testKEvent}.
    """

    @implementer(_IKQueue)
    class FakeKQueue:
        kqueue = testKQueue
        kevent = testKEvent

    return FakeKQueue()


class KQueueTests(TestCase):
    """
    These are tests for L{KQueueReactor}'s implementation, not its real world
    behaviour. For that, look at
    L{twisted.internet.test.reactormixins.ReactorBuilder}.
    """

    skip = kqueueSkip

    def test_EINTR(self) -> None:
        """
        L{KQueueReactor} handles L{errno.EINTR} in C{doKEvent} by returning.
        """

        class FakeKQueue:
            """
            A fake KQueue that raises L{errno.EINTR} when C{control} is called,
            like a real KQueue would if it was interrupted.
            """

            def control(self, *args: object, **kwargs: object) -> None:
                raise OSError(errno.EINTR, "Interrupted")

        reactor = KQueueReactor(makeFakeKQueue(FakeKQueue, _fakeKEvent))
        # This should return cleanly -- should not raise the OSError we're
        # spawning, nor get upset and raise about the incomplete KQueue fake.
        reactor.doKEvent(0)