This file is indexed.

/usr/lib/python3/dist-packages/tftp/util.py is in python3-txtftp 0.1~bzr42-0ubuntu2.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
'''
@author: shylent
'''
from functools import wraps
from twisted.internet import reactor
from twisted.internet.defer import maybeDeferred


__all__ = ['SequentialCall', 'Spent', 'Cancelled', 'deferred',
           'int_to_byte_string']


class Spent(Exception):
    """Trying to iterate a L{SequentialCall}, that is exhausted"""

class Cancelled(Exception):
    """Trying to iterate a L{SequentialCall}, that's been cancelled"""

def no_op(*args, **kwargs):
    pass

class SequentialCall(object):
    """Calls a given callable at intervals, specified by the L{timeout} iterable.
    Optionally calls a timeout handler, if provided, when there are no more timeout values.

    @param timeout: an iterable, that yields valid _seconds arguments to
    L{callLater<twisted.internet.interfaces.IReactorTime.callLater>} (floats)
    @type timeout: any iterable

    @param run_now: whether or not the callable should be called immediately
    upon initialization. Relinquishes control to the reactor
    (calls callLater(0,...)). Default: C{False}.
    @type run_now: C{bool}

    @param callable: the callable, that will be called at the specified intervals
    @param callable_args: the arguments to call it with
    @param callable_kwargs: the keyword arguments to call it with

    @param on_timeout: the callable, that will be called when there are no more
    timeout values
    @param on_timeout_args: the arguments to call it with
    @param on_timeout_kwargs: the keyword arguments to call it with 
    
    """

    @classmethod
    def run(cls, timeout, callable, callable_args=None, callable_kwargs=None,
                 on_timeout=None, on_timeout_args=None, on_timeout_kwargs=None,
                 run_now=False, _clock=None):
        """Create a L{SequentialCall} object and start its scheduler cycle

        @see: L{SequentialCall}

        """
        inst = cls(timeout, callable, callable_args, callable_kwargs,
                   on_timeout, on_timeout_args, on_timeout_kwargs,
                   run_now, _clock)
        inst.reschedule()
        return inst

    def __init__(self, timeout,
                 callable, callable_args=None, callable_kwargs=None,
                 on_timeout=None, on_timeout_args=None, on_timeout_kwargs=None,
                 run_now=False, _clock=None):
        self._timeout = iter(timeout)
        self.callable = callable
        self.callable_args = callable_args or []
        self.callable_kwargs = callable_kwargs or {}
        self.on_timeout = on_timeout or no_op
        self.on_timeout_args = on_timeout_args or []
        self.on_timeout_kwargs = on_timeout_kwargs or {}
        self._wd = None
        self._spent = self._cancelled = False
        self._ran_first = not run_now
        if _clock is None:
            self._clock = reactor
        else:
            self._clock = _clock

    def _call_and_schedule(self):
        self.callable(*self.callable_args, **self.callable_kwargs)
        self._ran_first = True
        if not self._spent:
            self.reschedule()

    def reschedule(self):
        """Schedule the next L{callable} call

        @raise Spent: if the timeout iterator has been exhausted and on_timeout
        handler has been already called
        @raise Cancelled: if this L{SequentialCall} has already been cancelled

        """
        if not self._ran_first:
            self._wd = self._clock.callLater(0, self._call_and_schedule)
            return
        if self._cancelled:
            raise Cancelled("This SequentialCall has already been cancelled")
        if self._spent:
            raise Spent("This SequentialCall has already timed out")
        try:
            next_timeout = next(self._timeout)
            self._wd = self._clock.callLater(next_timeout, self._call_and_schedule)
        except StopIteration:
            self.on_timeout(*self.on_timeout_args, **self.on_timeout_kwargs)
            self._spent = True

    def cancel(self):
        """Cancel the next scheduled call

        @raise Cancelled: if this SequentialCall has already been cancelled
        @raise Spent: if this SequentialCall has expired

        """
        if self._cancelled:
            raise Cancelled("This SequentialCall has already been cancelled")
        if self._spent:
            raise Spent("This SequentialCall has already timed out")
        if self._wd is not None and self._wd.active():
            self._wd.cancel()
        self._spent = self._cancelled = True

    def active(self):
        """Whether or not this L{SequentialCall} object is considered active"""
        return not (self._spent or self._cancelled)


def deferred(func):
    """Decorates a function to ensure that it always returns a `Deferred`.

    This also serves a secondary documentation purpose; functions decorated
    with this are readily identifiable as asynchronous.
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        return maybeDeferred(func, *args, **kwargs)
    return wrapper


def int_to_byte_string(integer):
    """Convert `integer` into a byte string.

    For example, the number 1234 will be returned as b"1234". The obvious
    approach of c{bytes(n)} does not work on Python 3; in fact it will return
    a C{bytes} object of length C{n}.
    """
    if str is bytes:
        return str(integer)
    else:
        return str(integer).encode("ascii")