python-telegram-bot · jh0ker · Jan 3, 2016 · Dec 25, 2015 · Dec 25, 2015 · Dec 25, 2015
diff --git a/.gitignore b/.gitignore
@@ -61,3 +61,10 @@ target/
 
 # Sublime Text 2
 *.sublime*
+
+# unitests files
+telegram.mp3
+telegram.mp4
+telegram.ogg
+telegram.png
+telegram.webp
diff --git a/.travis.yml b/.travis.yml
@@ -9,6 +9,7 @@ python:
 install:
   - pip install pylint flake8 coveralls
   - pip install -r requirements.txt
+  - 'if [[ $TRAVIS_PYTHON_VERSION == 2.6 ]]; then pip install unittest2; fi'
 script:
   - nosetests --with-coverage --cover-package telegram/
   - flake8 telegram

diff --git a/AUTHORS.rst b/AUTHORS.rst
@@ -19,6 +19,7 @@ The following wonderful people contributed directly or indirectly to this projec
 - `macrojames <https://github.com/macrojames>`_
 - `naveenvhegde <https://github.com/naveenvhegde>`_
 - `njittam <https://github.com/njittam>`_
+- `Noam Meltzer <https://github.com/tsnoam>`_
 - `Rahiel Kasim <https://github.com/rahiel>`_
 - `sooyhwang <https://github.com/sooyhwang>`_
 - `wjt <https://github.com/wjt>`_

diff --git a/telegram/bot.py b/telegram/bot.py
@@ -52,7 +52,7 @@ class Bot(TelegramObject):
     def __init__(self,
                  token,
                  base_url=None):
-        self.token = token
+        self.token = self._valid_token(token)
 
         if base_url is None:
             self.base_url = 'https://api.telegram.org/bot%s' % self.token
@@ -744,3 +744,11 @@ def to_dict(self):
     def __reduce__(self):
         return (self.__class__, (self.token,
                                  self.base_url.replace(self.token, '')))
+
+    @staticmethod
+    def _valid_token(token):
+        """a very basic validation on token"""
+        left, sep, _right = token.partition(':')
+        if (not sep) or (not left.isdigit()) or (len(left) < 3):
+            raise TelegramError('Invalid token')
+        return token
diff --git a/telegram/error.py b/telegram/error.py
@@ -18,24 +18,43 @@
 
 """This module contains a object that represents a Telegram Error"""
 
-import re
+
+def _lstrip_str(in_s, lstr):
+    """
+    Args:
+        in_s (str): in string
+        lstr (str): substr to strip from left side
+
+    Returns:
+        str:
+
+    """
+    if in_s.startswith(lstr):
+        res = in_s[len(lstr):]
+    else:
+        res = in_s
+    return res
 
 
 class TelegramError(Exception):
     """This object represents a Telegram Error."""
 
     def __init__(self, message):
         """
+        Args:
+            message (str):
+
         Returns:
-            str:
+
         """
         super(TelegramError, self).__init__()
 
-        api_error = re.match(r'^Error: (?P<message>.*)', message)
-        if api_error:
-            self.message = api_error.group('message').capitalize()
-        else:
-            self.message = message
+        msg = _lstrip_str(message, 'Error: ')
+        msg = _lstrip_str(msg, '[Error]: ')
+        if msg != message:
+            # api_error - capitalize the msg...
+            msg = msg.capitalize()
+        self.message = msg
 
     def __str__(self):
         return '%s' % (self.message)
diff --git a/telegram/updater.py b/telegram/updater.py
@@ -64,15 +64,26 @@ class Updater:
     Attributes:
 
     Args:
-        token (str): The bots token given by the @BotFather
+        token (Optional[str]): The bot's token given by the @BotFather
         base_url (Optional[str]):
         workers (Optional[int]): Amount of threads in the thread pool for
             functions decorated with @run_async
+        bot (Optional[Bot]):
+
+    Raises:
+        ValueError: If both `token` and `bot` are passed or none of them.
     """
 
-    def __init__(self, token, base_url=None, workers=4):
+    def __init__(self, token=None, base_url=None, workers=4, bot=None):
+        if (token is None) and (bot is None):
+            raise ValueError('`token` or `bot` must be passed')
+        if (token is not None) and (bot is not None):
+            raise ValueError('`token` and `bot` are mutually exclusive')
 
-        self.bot = Bot(token, base_url)
+        if bot is not None:
+            self.bot = bot
+        else:
+            self.bot = Bot(token, base_url)
         self.update_queue = Queue()
         self.dispatcher = Dispatcher(self.bot, self.update_queue,
                                      workers=workers)

diff --git a/telegram/utils/request.py b/telegram/utils/request.py
@@ -19,6 +19,7 @@
 
 """This module contains methods to make POST and GET requests"""
 
+import functools
 import json
 import socket
 from ssl import SSLError
@@ -45,14 +46,46 @@ def _parse(json_data):
     Returns:
       A JSON parsed as Python dict with results.
     """
-    data = json.loads(json_data.decode())
+    decoded_s = json_data.decode('utf-8')
+    try:
+        data = json.loads(decoded_s)
+    except ValueError:
+        raise TelegramError('Invalid server response')
 
     if not data.get('ok') and data.get('description'):
         return data['description']
 
     return data['result']
 
 
+def _try_except_req(func):
+    """Decorator for requests to handle known exceptions"""
+    @functools.wraps(func)
+    def decorator(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except HTTPError as error:
+            if error.getcode() == 403:
+                raise TelegramError('Unauthorized')
+            if error.getcode() == 502:
+                raise TelegramError('Bad Gateway')
+
+            try:
+                message = _parse(error.read())
+            except ValueError:
+                message = 'Unknown HTTPError {0}'.format(error.getcode())
+
+            raise TelegramError(message)
+        except (SSLError, socket.timeout) as error:
+            if "operation timed out" in str(error):
+                raise TelegramError("Timed out")
+
+            raise TelegramError(str(error))
+
+    return decorator
+
+
+@_try_except_req
 def get(url):
     """Request an URL.
     Args:
@@ -67,6 +100,7 @@ def get(url):
     return _parse(result)
 
 
+@_try_except_req
 def post(url,
          data,
          network_delay=2.):
@@ -91,39 +125,22 @@ def post(url,
     else:
         timeout = None
 
-    try:
-        if InputFile.is_inputfile(data):
-            data = InputFile(data)
-            request = Request(url,
-                              data=data.to_form(),
-                              headers=data.headers)
-        else:
-            data = json.dumps(data)
-            request = Request(url,
-                              data=data.encode(),
-                              headers={'Content-Type': 'application/json'})
-
-        result = urlopen(request, timeout=timeout).read()
-    except HTTPError as error:
-        if error.getcode() == 403:
-            raise TelegramError('Unauthorized')
-        if error.getcode() == 502:
-            raise TelegramError('Bad Gateway')
-
-        try:
-            message = _parse(error.read())
-        except ValueError:
-            message = 'Unknown HTTPError'
-
-        raise TelegramError(message)
-    except (SSLError, socket.timeout) as error:
-        if "operation timed out" in str(error):
-            raise TelegramError("Timed out")
+    if InputFile.is_inputfile(data):
+        data = InputFile(data)
+        request = Request(url,
+                          data=data.to_form(),
+                          headers=data.headers)
+    else:
+        data = json.dumps(data)
+        request = Request(url,
+                          data=data.encode(),
+                          headers={'Content-Type': 'application/json'})
 
-        raise TelegramError(str(error))
+    result = urlopen(request, timeout=timeout).read()
     return _parse(result)
 
 
+@_try_except_req
 def download(url,
              filename):
     """Download a file by its URL.

diff --git a/tests/test_bot.py b/tests/test_bot.py
@@ -20,9 +20,14 @@
 """This module contains a object that represents Tests for Telegram Bot"""
 
 import os
-import unittest
 from datetime import datetime
 import sys
+
+if sys.version_info[0:2] == (2, 6):
+    import unittest2 as unittest
+else:
+    import unittest
+
 sys.path.append('.')
 
 import telegram
@@ -137,5 +142,33 @@ def testGetUserProfilePhotos(self):
         self.assertTrue(self.is_json(upf.to_json()))
         self.assertEqual(upf.photos[0][0].file_size, 12421)
 
+    def _test_invalid_token(self, token):
+        print('Testing invalid token: {0}'.format(token))
+        self.assertRaisesRegexp(telegram.TelegramError, 'Invalid token', telegram.Bot, token)
+
+    def testInvalidToken1(self):
+        self._test_invalid_token('123')
+
+    def testInvalidToken2(self):
+        self._test_invalid_token('12a:')
+
+    def testInvalidToken3(self):
+        self._test_invalid_token('12:')
+
+    def testUnauthToken(self):
+        print('Testing unauthorized token')
+        with self.assertRaisesRegexp(telegram.TelegramError, 'Unauthorized'):
+            bot = telegram.Bot('1234:abcd1234')
+            bot.getMe()
+
+    def testInvalidSrvResp(self):
+        print('Testing invalid server response')
+        with self.assertRaisesRegexp(telegram.TelegramError, 'Invalid server response'):
+            # bypass the valid token check
+            bot_cls = type('bot_cls', (telegram.Bot, ), {'_valid_token': lambda self, token: token})
+            bot = bot_cls('12')
+            bot.getMe()
+
+
 if __name__ == '__main__':
     unittest.main()