Jason R. Coombs avatar Jason R. Coombs committed b7816c2

Moved scheduled commands to a schedule module

Comments (0)

Files changed (5)

+7.1
+===
+
+* Moved scheduled command classes to irc.schedule module. Kept references for
+  backwards-compatibility.
+
 7.0
 ===
 
 import socket
 import string
 import time
-import datetime
 import struct
 import logging
 import itertools
 from . import strings
 from . import util
 from . import buffer
+from . import schedule
 
 log = logging.getLogger(__name__)
 
 # ERROR from the server triggers the error event and the disconnect event.
 # dropping of the connection triggers the disconnect event.
 
+# for backward-compatibility
+from .schedule import (PeriodicCommand, DelayedCommand,
+    PeriodicCommandFixedDelay)
+
 class IRCError(Exception):
     "An IRC exception"
 
                 if not command.due():
                     break
                 command.function(*command.arguments)
-                if isinstance(command, PeriodicCommand):
+                if isinstance(command, schedule.PeriodicCommand):
                     self._schedule_command(command.next())
                 del self.delayed_commands[0]
 
             function -- Function to call.
             arguments -- Arguments to give the function.
         """
-        command = DelayedCommand.at_time(at, function, arguments)
+        command = schedule.DelayedCommand.at_time(at, function, arguments)
         self._schedule_command(command)
 
     def execute_delayed(self, delay, function, arguments=()):
         function -- Function to call.
         arguments -- Arguments to give the function.
         """
-        command = DelayedCommand(delay, function, arguments)
+        command = schedule.DelayedCommand(delay, function, arguments)
         self._schedule_command(command)
 
     def execute_every(self, period, function, arguments=()):
         function -- Function to call.
         arguments -- Arguments to give the function.
         """
-        command = PeriodicCommand(period, function, arguments)
+        command = schedule.PeriodicCommand(period, function, arguments)
         self._schedule_command(command)
 
     def _schedule_command(self, command):
             self.connections.remove(connection)
             self._on_disconnect(connection.socket)
 
-class DelayedCommand(datetime.datetime):
-    """
-    A command to be executed after some delay (seconds or timedelta).
-
-    Clients may override .now() to have dates interpreted in a different
-    manner, such as to use UTC or to have timezone-aware times.
-    """
-    def __new__(cls, delay, function, arguments):
-        if not isinstance(delay, datetime.timedelta):
-            delay = datetime.timedelta(seconds=delay)
-        at = cls.now() + delay
-        cmd = datetime.datetime.__new__(cls, at.year,
-            at.month, at.day, at.hour, at.minute, at.second,
-            at.microsecond, at.tzinfo)
-        cmd.delay = delay
-        cmd.function = function
-        cmd.arguments = arguments
-        return cmd
-
-    @classmethod
-    def now(self, tzinfo=None):
-        return datetime.datetime.now(tzinfo)
-
-    @classmethod
-    def at_time(cls, at, function, arguments):
-        """
-        Construct a DelayedCommand to come due at `at`, where `at` may be
-        a datetime or timestamp. If `at` is a timestamp, it will be
-        interpreted as a naive local timestamp.
-        """
-        if isinstance(at, int):
-            at = datetime.datetime.fromtimestamp(at)
-        delay = at - cls.now()
-        return cls(delay, function, arguments)
-
-    def due(self):
-        return self.now() >= self
-
-class PeriodicCommandBase(DelayedCommand):
-    def next(self):
-        return PeriodicCommand(self.delay, self.function,
-            self.arguments)
-
-    def _check_delay(self):
-        if not self.delay > datetime.timedelta():
-            raise ValueError("A PeriodicCommand must have a positive, "
-                "non-zero delay.")
-
-class PeriodicCommand(PeriodicCommandBase):
-    """
-    Like a delayed command, but expect this command to run every delay
-    seconds.
-    """
-    def __init__(self, *args, **kwargs):
-        super(PeriodicCommand, self).__init__(*args, **kwargs)
-        self._check_delay()
-
-class PeriodicCommandFixedDelay(PeriodicCommandBase):
-    """
-    Like a periodic command, but don't calculate the delay based on
-    the current time. Instead use a fixed delay following the initial
-    run.
-    """
-
-    @classmethod
-    def at_time(cls, at, delay, function, arguments):
-        cmd = super(PeriodicCommandFixedDelay, cls).at_time(
-            at, function, arguments)
-        if not isinstance(delay, datetime.timedelta):
-            delay = datetime.timedelta(seconds=delay)
-        cmd.delay = delay
-        cmd._check_delay()
-        return cmd
-
 _rfc_1459_command_regexp = re.compile("^(:(?P<prefix>[^ ]+) +)?(?P<command>[^ ]+)( *(?P<argument> .+))?")
 
 class Connection(object):
+"""
+Classes for calling functions a schedule.
+"""
+
+import datetime
+
+class DelayedCommand(datetime.datetime):
+    """
+    A command to be executed after some delay (seconds or timedelta).
+
+    Clients may override .now() to have dates interpreted in a different
+    manner, such as to use UTC or to have timezone-aware times.
+    """
+    def __new__(cls, delay, function, arguments):
+        if not isinstance(delay, datetime.timedelta):
+            delay = datetime.timedelta(seconds=delay)
+        at = cls.now() + delay
+        cmd = datetime.datetime.__new__(cls, at.year,
+            at.month, at.day, at.hour, at.minute, at.second,
+            at.microsecond, at.tzinfo)
+        cmd.delay = delay
+        cmd.function = function
+        cmd.arguments = arguments
+        return cmd
+
+    @classmethod
+    def now(self, tzinfo=None):
+        return datetime.datetime.now(tzinfo)
+
+    @classmethod
+    def at_time(cls, at, function, arguments):
+        """
+        Construct a DelayedCommand to come due at `at`, where `at` may be
+        a datetime or timestamp. If `at` is a timestamp, it will be
+        interpreted as a naive local timestamp.
+        """
+        if isinstance(at, int):
+            at = datetime.datetime.fromtimestamp(at)
+        delay = at - cls.now()
+        return cls(delay, function, arguments)
+
+    def due(self):
+        return self.now() >= self
+
+class PeriodicCommandBase(DelayedCommand):
+    def next(self):
+        return PeriodicCommand(self.delay, self.function,
+            self.arguments)
+
+    def _check_delay(self):
+        if not self.delay > datetime.timedelta():
+            raise ValueError("A PeriodicCommand must have a positive, "
+                "non-zero delay.")
+
+class PeriodicCommand(PeriodicCommandBase):
+    """
+    Like a delayed command, but expect this command to run every delay
+    seconds.
+    """
+    def __init__(self, *args, **kwargs):
+        super(PeriodicCommand, self).__init__(*args, **kwargs)
+        self._check_delay()
+
+class PeriodicCommandFixedDelay(PeriodicCommandBase):
+    """
+    Like a periodic command, but don't calculate the delay based on
+    the current time. Instead use a fixed delay following the initial
+    run.
+    """
+
+    @classmethod
+    def at_time(cls, at, delay, function, arguments):
+        cmd = super(PeriodicCommandFixedDelay, cls).at_time(
+            at, function, arguments)
+        if not isinstance(delay, datetime.timedelta):
+            delay = datetime.timedelta(seconds=delay)
+        cmd.delay = delay
+        cmd._check_delay()
+        return cmd

irc/tests/test_client.py

 	assert isinstance(irc.client.VERSION, tuple)
 	assert irc.client.VERSION, "No VERSION detected."
 
-def test_delayed_command_order():
-	"""
-	delayed commands should be sorted by delay time
-	"""
-	null = lambda: None
-	delays = [random.randint(0, 99) for x in range(5)]
-	cmds = sorted([
-		irc.client.DelayedCommand(delay, null, tuple())
-		for delay in delays
-	])
-	assert [c.delay.seconds for c in cmds] == sorted(delays)
-
-def test_periodic_command_delay():
-	"A PeriodicCommand must have a positive, non-zero delay."
-	with pytest.raises(ValueError) as exc_info:
-		irc.client.PeriodicCommand(0, None, None)
-	assert str(exc_info.value) == test_periodic_command_delay.__doc__
-
-def test_periodic_command_fixed_delay():
-	"""
-	Test that we can construct a periodic command with a fixed initial
-	delay.
-	"""
-	fd = irc.client.PeriodicCommandFixedDelay.at_time(
-		at = datetime.datetime.now(),
-		delay = datetime.timedelta(seconds=2),
-		function = lambda: None,
-		arguments = [],
-		)
-	assert fd.due() == True
-	assert fd.next().due() == False
-
 @mock.patch('irc.connection.socket')
 def test_privmsg_sends_msg(socket_mod):
 	server = irc.client.IRC().server()

irc/tests/test_schedule.py

+import random
+import datetime
+
+import pytest
+
+from irc import schedule
+
+
+def test_delayed_command_order():
+	"""
+	delayed commands should be sorted by delay time
+	"""
+	null = lambda: None
+	delays = [random.randint(0, 99) for x in range(5)]
+	cmds = sorted([
+		schedule.DelayedCommand(delay, null, tuple())
+		for delay in delays
+	])
+	assert [c.delay.seconds for c in cmds] == sorted(delays)
+
+def test_periodic_command_delay():
+	"A PeriodicCommand must have a positive, non-zero delay."
+	with pytest.raises(ValueError) as exc_info:
+		schedule.PeriodicCommand(0, None, None)
+	assert str(exc_info.value) == test_periodic_command_delay.__doc__
+
+def test_periodic_command_fixed_delay():
+	"""
+	Test that we can construct a periodic command with a fixed initial
+	delay.
+	"""
+	fd = schedule.PeriodicCommandFixedDelay.at_time(
+		at = datetime.datetime.now(),
+		delay = datetime.timedelta(seconds=2),
+		function = lambda: None,
+		arguments = [],
+		)
+	assert fd.due() == True
+	assert fd.next().due() == False
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.