-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Moving test code around, and adding to the test suite a bit
- Loading branch information
Murray Christopherson
committed
Apr 10, 2018
1 parent
c96462f
commit 520eaef
Showing
5 changed files
with
197 additions
and
101 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| import os | ||
| import sys | ||
| import logging | ||
|
|
||
| LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO') | ||
|
|
||
| _logger = None | ||
| def logger(): | ||
| global _logger | ||
|
|
||
| if _logger is None: | ||
| _logger = logging.getLogger(__name__) | ||
| _logger.setLevel(logging.getLevelName(LOG_LEVEL)) | ||
|
|
||
| stream_handler = logging.StreamHandler(sys.stdout) | ||
| stream_handler.setLevel(logging.DEBUG) | ||
| stream_handler.setFormatter( | ||
| logging.Formatter( | ||
| fmt='%(asctime)s (%(levelname)s): %(message)s', | ||
| datefmt='%Y-%m-%d %H:%M:%S')) | ||
| _logger.addHandler(stream_handler) | ||
|
|
||
| return _logger |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| import os | ||
| import pprint | ||
| import time | ||
| import six | ||
|
|
||
| import requests | ||
|
|
||
| from tests.logger import logger | ||
|
|
||
| MASHAPE_KEY = os.environ['MASHAPE_KEY'] | ||
|
|
||
| class PostShift: | ||
| ENDPOINT_URL = 'https://reuleaux-post-shift-v1.p.mashape.com/api.php' | ||
| REQUEST_INTERVAL = 10.0 | ||
|
|
||
| @staticmethod | ||
| def _headers(): | ||
| return { | ||
| 'X-Mashape-Key': MASHAPE_KEY, | ||
| 'Accept': 'application/json', | ||
| } | ||
|
|
||
| def __init__(self): | ||
| self.email_jsons = [] | ||
| self.last_request_time = None | ||
|
|
||
| def sleep_if_needed(self): | ||
| if self.last_request_time is not None: | ||
| diff = (time.time() - self.last_request_time) | ||
|
|
||
| if diff < PostShift.REQUEST_INTERVAL: | ||
| logger().debug('sleeping for %s seconds...', diff) | ||
| time.sleep(diff) | ||
|
|
||
| def create(self): | ||
| self.sleep_if_needed() | ||
| self.last_request_time = time.time() | ||
| r = requests.get(PostShift.ENDPOINT_URL, params={ | ||
| 'action': 'new', | ||
| 'type': 'json', | ||
| }, headers=PostShift._headers()) | ||
|
|
||
| _json = r.json() | ||
|
|
||
| logger().debug('`create`: %s', pprint.pformat(_json)) | ||
|
|
||
| if not isinstance(_json, dict): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if 'email' not in _json or 'key' not in _json: | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(_json['email'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(_json['key'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| return _json | ||
|
|
||
| def get_list(self, email_key): | ||
| self.sleep_if_needed() | ||
| self.last_request_time = time.time() | ||
| r = requests.get(PostShift.ENDPOINT_URL, params={ | ||
| 'action': 'getlist', | ||
| 'key': email_key, | ||
| 'type': 'json', | ||
| }, headers=PostShift._headers()) | ||
|
|
||
| _json = r.json() | ||
|
|
||
| logger().debug('`get_list`: %s', pprint.pformat(_json)) | ||
|
|
||
| if not isinstance(_json, list): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| for e_json in _json: | ||
| if not isinstance(e_json, dict): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if 'id' not in e_json or 'date' not in e_json or 'subject' not in e_json or 'from' not in e_json: | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(e_json['id'], int): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(e_json['date'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(e_json['subject'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(e_json['from'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| return _json | ||
|
|
||
| def clear(self, email_key): | ||
| self.sleep_if_needed() | ||
| self.last_request_time = time.time() | ||
| r = requests.get(PostShift.ENDPOINT_URL, params={ | ||
| 'action': 'clear', | ||
| 'key': email_key, | ||
| 'type': 'json', | ||
| }, headers=PostShift._headers()) | ||
|
|
||
| _json = r.json() | ||
|
|
||
| logger().debug('`clear`: %s', pprint.pformat(_json)) | ||
|
|
||
| if not isinstance(_json, dict): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if 'clear' not in _json or 'key' not in _json: | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(_json['clear'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if not isinstance(_json['key'], six.string_types): | ||
| raise RuntimeError('bad response') | ||
|
|
||
| if _json['clear'] != 'ok': | ||
| raise RuntimeError('bad response') | ||
|
|
||
| return _json |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| import unittest | ||
| import os | ||
| import time | ||
|
|
||
| from quick_email import send_email | ||
|
|
||
| from tests.postshift import PostShift | ||
|
|
||
| SMTP_HOST = os.environ['SMTP_HOST'] | ||
| SMTP_PORT = int(os.environ['SMTP_PORT']) | ||
| SMTP_USER = os.environ['SMTP_USER'] | ||
| SMTP_PASSWORD = os.environ['SMTP_PASSWORD'] | ||
| SMTP_IS_TLS = os.environ.get('SMTP_IS_TLS') == 'true' | ||
| SMTP_SENDER = os.environ['SMTP_SENDER'] | ||
|
|
||
|
|
||
| class TestQuickEmail(unittest.TestCase): | ||
| @classmethod | ||
| def setUpClass(cls): | ||
| cls.postshift = PostShift() | ||
|
|
||
| for i in range(6): | ||
| cls.postshift.create() | ||
|
|
||
| def test_send_email(self): | ||
| test_email_json = TestQuickEmail.postshift.email_jsons[0] | ||
| send_email(SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASSWORD, SMTP_IS_TLS, SMTP_SENDER, test_email_json['email'], None, None, 'The Subject', 'Some Text', '<b>Some Bold Text</b>', attachment_list=None, inline_attachment_dict=None) | ||
|
|
||
| # wait for email to arrive | ||
| time.sleep(30.0) | ||
|
|
||
| _json = TestQuickEmail.postshift.get_list(test_email_json['key']) | ||
|
|
||
| self.assertIsInstance(_json, list) | ||
| self.assertEqual(len(_json), 1) | ||
|
|
||
|
|
||
| if __name__ == '__main__': | ||
| unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters