tests/test_integration.py

   1 from shutil import rmtree
   2 import os
   3 import smtplib
   4 import subprocess
   5 import sys
   6 import time
   7
   8 from test_app.config import settings as server_settings
   9
  10 from salmon import queue
  11
  12 from .setup_env import SalmonTestCase, dirs
  13
  14
  15 class IntegrationTestCase(SalmonTestCase):
  16     @classmethod
  17     def setUpClass(cls):
  18         cls._cwd = "tests/data/test_app"
  19         for path in dirs:
  20             rmtree(os.path.join(cls._cwd, path), ignore_errors=True)
  21             os.mkdir(os.path.join(cls._cwd, path))
  22         cls._server = subprocess.Popen(["salmon", "start", "--boot", "config.dump", "--no-daemon"],
  23                                        cwd=cls._cwd, stdout=sys.stdout, stderr=sys.stderr)
  24         for i in range(5):
  25             try:
  26                 conn = smtplib.SMTP(**server_settings.receiver_config)
  27             except Exception:
  28                 time.sleep(2**i)
  29                 continue
  30             else:
  31                 conn.quit()
  32                 return
  33
  34         raise Exception("Server still not ready, something must be wrong")
  35
  36     @classmethod
  37     def tearDownClass(cls):
  38         cls._server.kill()
  39         cls._server = None
  40         for path in dirs:
  41             rmtree(os.path.join(cls._cwd, path), ignore_errors=True)
  42
  43     def setUp(self):
  44         super().setUp()
  45         # re-create destoryed queues
  46         queue.Queue(os.path.join(self._cwd, server_settings.UNDELIVERABLE_QUEUE)).clear()
  47         queue.Queue(os.path.join(self._cwd, server_settings.QUEUE_PATH)).clear()
  48
  49     def test_we_get_message(self):
  50         client = smtplib.SMTP(**server_settings.receiver_config)
  51
  52         client.helo()
  53         client.sendmail("me@example.com", "you@example.com", "hello")
  54
  55         undelivered = queue.Queue(os.path.join(self._cwd, server_settings.UNDELIVERABLE_QUEUE))
  56         self.assertEqual(len(undelivered), 0)
  57
  58         inbox = queue.Queue(os.path.join(self._cwd, server_settings.QUEUE_PATH))
  59         self.assertEqual(len(inbox), 1)
  60
  61     def test_we_dont_get_message(self):
  62         client = smtplib.SMTP(**server_settings.receiver_config)
  63
  64         client.helo()
  65         client.sendmail("me@example.com", "you@example1.com", "hello")
  66
  67         undelivered = queue.Queue(os.path.join(self._cwd, server_settings.UNDELIVERABLE_QUEUE))
  68         self.assertEqual(len(undelivered), 1)
  69
  70         inbox = queue.Queue(os.path.join(self._cwd, server_settings.QUEUE_PATH))
  71         self.assertEqual(len(inbox), 0)