Blah
wat?
(testing fedmsg)
(testing fedmsg some more..)
Awesome:
pagure.issue.comment.added -- ralph commented on the ticket pagure#2: "testing" https://pagure.io/pagure/issue/2
Been working on some code today:
import datetime class Denomination(object): def __init__(self, value): self.value = list(set(value or self.default)) self.value.sort() for item in self.value: if item not in self.default: raise ValueError("%r is not allowed" % item) def next(self, now, remainder=0): T = getattr(now, type(self).__name__.lower()) + remainder try: return next(v for v in self.value if v >= T), 0 except StopIteration: return next(iter(self.value)), 1 # self.value[0] class Second(Denomination): default = range(0, 60) class Minute(Denomination): default = range(0, 60) class Hour(Denomination): default = range(0, 24) class Schedule(object): def __init__(self, second=None, minute=None, hour=None): self.second = Second(second) self.minute = Minute(minute) self.hour = Hour(hour) def __iter__(self): return self def next(self): """ Returns the datetime object representing when we'll fire next. """ now = datetime.datetime.utcnow() s, remainder = self.second.next(now) m, remainder = self.minute.next(now, remainder) h, remainder = self.hour.next(now, remainder) return datetime.datetime(now.year, now.month, now.day, h, m, s) + \ datetime.timedelta(days=remainder) # For nice rendering def __str__(self): return str(self.next() - datetime.datetime.utcnow()) def __repr__(self): "<Schedule, fires next in %s>" % str(self) # For serialization def __json__(self): return float(self) # For usage in 'time.sleep(self)' def __float__(self): """ Return the number of seconds until we should fire next """ return (self.next() - datetime.datetime.utcnow()).total_seconds() import time import unittest import freezegun import nose.tools class TestSchedule(unittest.TestCase): @freezegun.freeze_time('2012-01-14 00:00:00') def test_basic(self): s = Schedule(minute=[15], hour=[5]) self.assertEquals(float(s), 5 * 60 * 60 + 15 * 60) @freezegun.freeze_time('2012-01-14 12:00:00') def test_wrap_day(self): s = Schedule(minute=[15], hour=[5]) self.assertEquals(float(s), 17 * 60 * 60 + 15 * 60) @freezegun.freeze_time('2012-01-14 12:15:00') def test_wrap_day_match_minute(self): s = Schedule(minute=[15], hour=[5]) self.assertEquals(float(s), 17 * 60 * 60) @freezegun.freeze_time('2012-01-14 12:16:00') def test_wrap_day_wrap_minute(self): s = Schedule(minute=[15], hour=[5]) self.assertEquals(float(s), 16 * 60 * 60 + 59 * 60) @freezegun.freeze_time('2012-01-14 00:00:00') def test_working_with_time_sleep(self): s = Schedule(second=[1]) time.sleep(s) # Let's just make sure this doesn't crash @nose.tools.raises(ValueError) def test_wrap_invalid_item(self): Schedule(minute=['lol']) @nose.tools.raises(TypeError) def test_wrap_invalid_item_int(self): Schedule(minute=1) if __name__ == '__main__': unittest.main()
Looks like we're almost there but there is still a bug to fix
o/
Let's fix that nasty bug !
ORA ORA ORA ORA ORA ORA :fist:
now ?
Works :fireworks: :dancers:
Testing to see if fedmsg makes it to Fedora Infra.
Success?
For example
Just a test of fedmsg to #fedora-apps.
I was told async comment don't work anymore?
Some fun test
Enough playing with this one, let's consider it invalid :)