python-telegram-bot/tests/test_jobqueue.py

204 lines
5.5 KiB
Python
Raw Normal View History

#!/usr/bin/env python
# encoding: utf-8
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2016
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""
2016-10-17 00:22:40 +02:00
This module contains an object that represents Tests for JobQueue
"""
import logging
import sys
import unittest
Job queue time units (#452) * Adding timeunit and day support to the jobqueue * Adding tests * Changed the file permission back to 644. * Changed AssertEqual argument order to (actual, expectd). * Removed the TimeUnit enum and unit param, instead use datetime.time for interval. * Removing the TimeUnits enum and unit param in favour of optionally using a datetime.time as the interval. * Removing the TimeUnits enumeration, forgot the remove it in the last one. * Removed some old docstrings refering to the TimeUnits enum. * Removed the old TimeUnits import. * Adding some error handling for the 'days' argument (only a 'tuple' with 'Days') * Writing the error message directly in the exception. * Moving a debug statement wrongfully saying a job would be running on days it wouldn't. * Writing error messages directly in the exceptions instead of making an extra variable. * Replacing datetime.time in favour of datetime.timedelta because of the get_seconds() method. * Adding error handling for the method . * Splitting the tests up in multiple ones, no float test because I haven't found a reliable way to test it. * Excluding .exrc file. * Removing \ at EOF of ValueError. * Replacing Enums with plain new-style classes. * Using numbers.number to check for ints/floats instead of seperate int/float checks. * Fixing typo, number -> Number. * Changed lower_case Days attributes to UPPER_CASE. * Different formatting for Days class, removed the get_days function in favour of a tuple. * Removed redundant function get_days. * Edited the docstring for next_t to also take datetime.timedelta. * Removed for-loop in favour of any(). * Changed docstring for interval. * Removed debug print. * Changing some docstrings. * Changing some docstrings (again).
2016-11-08 23:39:25 +01:00
import datetime
import time
from math import ceil
from time import sleep
from tests.test_updater import MockBot
sys.path.append('.')
from telegram.ext import JobQueue, Job, Updater
from tests.base import BaseTest
# Enable logging
root = logging.getLogger()
root.setLevel(logging.INFO)
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(logging.WARN)
2016-05-15 02:52:35 +02:00
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
root.addHandler(ch)
2016-01-04 11:08:43 +01:00
class JobQueueTest(BaseTest, unittest.TestCase):
"""
This object represents Tests for Updater, Dispatcher, WebhookServer and
WebhookHandler
"""
def setUp(self):
self.jq = JobQueue(MockBot('jobqueue_test'))
self.jq.start()
self.result = 0
Job queue time units (#452) * Adding timeunit and day support to the jobqueue * Adding tests * Changed the file permission back to 644. * Changed AssertEqual argument order to (actual, expectd). * Removed the TimeUnit enum and unit param, instead use datetime.time for interval. * Removing the TimeUnits enum and unit param in favour of optionally using a datetime.time as the interval. * Removing the TimeUnits enumeration, forgot the remove it in the last one. * Removed some old docstrings refering to the TimeUnits enum. * Removed the old TimeUnits import. * Adding some error handling for the 'days' argument (only a 'tuple' with 'Days') * Writing the error message directly in the exception. * Moving a debug statement wrongfully saying a job would be running on days it wouldn't. * Writing error messages directly in the exceptions instead of making an extra variable. * Replacing datetime.time in favour of datetime.timedelta because of the get_seconds() method. * Adding error handling for the method . * Splitting the tests up in multiple ones, no float test because I haven't found a reliable way to test it. * Excluding .exrc file. * Removing \ at EOF of ValueError. * Replacing Enums with plain new-style classes. * Using numbers.number to check for ints/floats instead of seperate int/float checks. * Fixing typo, number -> Number. * Changed lower_case Days attributes to UPPER_CASE. * Different formatting for Days class, removed the get_days function in favour of a tuple. * Removed redundant function get_days. * Edited the docstring for next_t to also take datetime.timedelta. * Removed for-loop in favour of any(). * Changed docstring for interval. * Removed debug print. * Changing some docstrings. * Changing some docstrings (again).
2016-11-08 23:39:25 +01:00
self.job_time = 0
def tearDown(self):
if self.jq is not None:
self.jq.stop()
Job queue time units (#452) * Adding timeunit and day support to the jobqueue * Adding tests * Changed the file permission back to 644. * Changed AssertEqual argument order to (actual, expectd). * Removed the TimeUnit enum and unit param, instead use datetime.time for interval. * Removing the TimeUnits enum and unit param in favour of optionally using a datetime.time as the interval. * Removing the TimeUnits enumeration, forgot the remove it in the last one. * Removed some old docstrings refering to the TimeUnits enum. * Removed the old TimeUnits import. * Adding some error handling for the 'days' argument (only a 'tuple' with 'Days') * Writing the error message directly in the exception. * Moving a debug statement wrongfully saying a job would be running on days it wouldn't. * Writing error messages directly in the exceptions instead of making an extra variable. * Replacing datetime.time in favour of datetime.timedelta because of the get_seconds() method. * Adding error handling for the method . * Splitting the tests up in multiple ones, no float test because I haven't found a reliable way to test it. * Excluding .exrc file. * Removing \ at EOF of ValueError. * Replacing Enums with plain new-style classes. * Using numbers.number to check for ints/floats instead of seperate int/float checks. * Fixing typo, number -> Number. * Changed lower_case Days attributes to UPPER_CASE. * Different formatting for Days class, removed the get_days function in favour of a tuple. * Removed redundant function get_days. * Edited the docstring for next_t to also take datetime.timedelta. * Removed for-loop in favour of any(). * Changed docstring for interval. * Removed debug print. * Changing some docstrings. * Changing some docstrings (again).
2016-11-08 23:39:25 +01:00
def getSeconds(self):
return int(ceil(time.time()))
def job1(self, bot, job):
self.result += 1
def job2(self, bot, job):
2016-01-04 02:36:15 +01:00
raise Exception("Test Error")
def job3(self, bot, job):
self.result += 1
job.schedule_removal()
2016-05-26 14:07:44 +02:00
def job4(self, bot, job):
self.result += job.context
Job queue time units (#452) * Adding timeunit and day support to the jobqueue * Adding tests * Changed the file permission back to 644. * Changed AssertEqual argument order to (actual, expectd). * Removed the TimeUnit enum and unit param, instead use datetime.time for interval. * Removing the TimeUnits enum and unit param in favour of optionally using a datetime.time as the interval. * Removing the TimeUnits enumeration, forgot the remove it in the last one. * Removed some old docstrings refering to the TimeUnits enum. * Removed the old TimeUnits import. * Adding some error handling for the 'days' argument (only a 'tuple' with 'Days') * Writing the error message directly in the exception. * Moving a debug statement wrongfully saying a job would be running on days it wouldn't. * Writing error messages directly in the exceptions instead of making an extra variable. * Replacing datetime.time in favour of datetime.timedelta because of the get_seconds() method. * Adding error handling for the method . * Splitting the tests up in multiple ones, no float test because I haven't found a reliable way to test it. * Excluding .exrc file. * Removing \ at EOF of ValueError. * Replacing Enums with plain new-style classes. * Using numbers.number to check for ints/floats instead of seperate int/float checks. * Fixing typo, number -> Number. * Changed lower_case Days attributes to UPPER_CASE. * Different formatting for Days class, removed the get_days function in favour of a tuple. * Removed redundant function get_days. * Edited the docstring for next_t to also take datetime.timedelta. * Removed for-loop in favour of any(). * Changed docstring for interval. * Removed debug print. * Changing some docstrings. * Changing some docstrings (again).
2016-11-08 23:39:25 +01:00
def job5(self, bot, job):
self.job_time = self.getSeconds()
def test_basic(self):
self.jq.put(Job(self.job1, 0.1))
2016-01-04 02:11:42 +01:00
sleep(1.5)
self.assertGreaterEqual(self.result, 10)
2016-05-26 14:07:44 +02:00
def test_job_with_context(self):
self.jq.put(Job(self.job4, 0.1, context=5))
sleep(1.5)
self.assertGreaterEqual(self.result, 50)
def test_noRepeat(self):
self.jq.put(Job(self.job1, 0.1, repeat=False))
sleep(0.5)
self.assertEqual(1, self.result)
def test_nextT(self):
self.jq.put(Job(self.job1, 0.1), next_t=0.5)
sleep(0.45)
self.assertEqual(0, self.result)
sleep(0.1)
self.assertEqual(1, self.result)
2016-01-04 02:36:15 +01:00
def test_multiple(self):
self.jq.put(Job(self.job1, 0.1, repeat=False))
self.jq.put(Job(self.job1, 0.2, repeat=False))
self.jq.put(Job(self.job1, 0.4))
2016-01-04 02:36:15 +01:00
sleep(1)
self.assertEqual(4, self.result)
def test_disabled(self):
j0 = Job(self.job1, 0.1)
j1 = Job(self.job1, 0.2)
self.jq.put(j0)
self.jq.put(Job(self.job1, 0.4))
self.jq.put(j1)
j0.enabled = False
j1.enabled = False
sleep(1)
self.assertEqual(2, self.result)
def test_schedule_removal(self):
j0 = Job(self.job1, 0.1)
j1 = Job(self.job1, 0.2)
self.jq.put(j0)
self.jq.put(Job(self.job1, 0.4))
self.jq.put(j1)
j0.schedule_removal()
j1.schedule_removal()
sleep(1)
self.assertEqual(2, self.result)
def test_schedule_removal_from_within(self):
self.jq.put(Job(self.job1, 0.4))
self.jq.put(Job(self.job3, 0.2))
sleep(1)
self.assertEqual(3, self.result)
def test_longer_first(self):
self.jq.put(Job(self.job1, 0.2, repeat=False))
self.jq.put(Job(self.job1, 0.1, repeat=False))
sleep(0.15)
self.assertEqual(1, self.result)
2016-01-04 02:36:15 +01:00
def test_error(self):
self.jq.put(Job(self.job2, 0.1))
self.jq.put(Job(self.job1, 0.2))
sleep(0.5)
self.assertEqual(2, self.result)
def test_jobs_tuple(self):
self.jq.stop()
jobs = tuple(Job(self.job1, t) for t in range(5, 25))
for job in jobs:
self.jq.put(job)
self.assertTupleEqual(jobs, self.jq.jobs())
2016-01-05 13:32:19 +01:00
def test_inUpdater(self):
u = Updater(bot="MockBot")
u.job_queue.start()
try:
u.job_queue.put(Job(self.job1, 0.5))
sleep(0.75)
self.assertEqual(1, self.result)
u.stop()
sleep(2)
self.assertEqual(1, self.result)
finally:
u.stop()
2016-01-04 11:08:43 +01:00
Job queue time units (#452) * Adding timeunit and day support to the jobqueue * Adding tests * Changed the file permission back to 644. * Changed AssertEqual argument order to (actual, expectd). * Removed the TimeUnit enum and unit param, instead use datetime.time for interval. * Removing the TimeUnits enum and unit param in favour of optionally using a datetime.time as the interval. * Removing the TimeUnits enumeration, forgot the remove it in the last one. * Removed some old docstrings refering to the TimeUnits enum. * Removed the old TimeUnits import. * Adding some error handling for the 'days' argument (only a 'tuple' with 'Days') * Writing the error message directly in the exception. * Moving a debug statement wrongfully saying a job would be running on days it wouldn't. * Writing error messages directly in the exceptions instead of making an extra variable. * Replacing datetime.time in favour of datetime.timedelta because of the get_seconds() method. * Adding error handling for the method . * Splitting the tests up in multiple ones, no float test because I haven't found a reliable way to test it. * Excluding .exrc file. * Removing \ at EOF of ValueError. * Replacing Enums with plain new-style classes. * Using numbers.number to check for ints/floats instead of seperate int/float checks. * Fixing typo, number -> Number. * Changed lower_case Days attributes to UPPER_CASE. * Different formatting for Days class, removed the get_days function in favour of a tuple. * Removed redundant function get_days. * Edited the docstring for next_t to also take datetime.timedelta. * Removed for-loop in favour of any(). * Changed docstring for interval. * Removed debug print. * Changing some docstrings. * Changing some docstrings (again).
2016-11-08 23:39:25 +01:00
def test_time_unit_int(self):
# Testing seconds in int
seconds_interval = 5
expected_time = self.getSeconds() + seconds_interval
self.jq.put(Job(self.job5, seconds_interval, repeat=False))
sleep(6)
self.assertEqual(self.job_time, expected_time)
def test_time_unit_dt_time(self):
# Testing seconds, minutes and hours as datetime.timedelta object
# This is sufficient to test that it actually works.
interval = datetime.timedelta(seconds=5)
expected_time = self.getSeconds() + interval.total_seconds()
self.jq.put(Job(self.job5, interval, repeat=False))
sleep(6)
self.assertEqual(self.job_time, expected_time)
2016-04-24 04:11:25 +02:00
if __name__ == '__main__':
unittest.main()