summaryrefslogtreecommitdiff
path: root/tests/test_canarydb.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_canarydb.py')
-rw-r--r--tests/test_canarydb.py232
1 files changed, 116 insertions, 116 deletions
diff --git a/tests/test_canarydb.py b/tests/test_canarydb.py
index a51afdc..195387e 100644
--- a/tests/test_canarydb.py
+++ b/tests/test_canarydb.py
@@ -5,119 +5,119 @@ import datetime
from emailcanary import canarydb
class TestCanaryDB(unittest.TestCase):
- def setUp(self):
- self.tempdir = tempfile.mkdtemp()
- self.db = canarydb.CanaryDB(self.tempdir + "canary.db")
-
- def tearDown(self):
- self.db.close()
- shutil.rmtree(self.tempdir)
-
- def testPingCheckPong(self):
- listAddress = "list@example.com"
- address = "test@example.com"
- time = datetime.datetime(2015, 10, 24, 9, 00)
- uuid = "1234"
- expectedDelta = datetime.datetime.now() - time
-
- # Record a Ping
- self.db.ping(listAddress, address, time, uuid)
-
- # Check for missing pongs
- missing = self.db.get_missing_pongs()
-
- self.assertEqual(1, len(missing))
- firstMissing = missing[0]
- self.assertEqual(4, len(firstMissing))
- self.assertEqual(listAddress, firstMissing[0])
- self.assertEqual(uuid, firstMissing[1])
- self.assertEqual(address, firstMissing[2])
- delta = firstMissing[3].total_seconds() - expectedDelta.total_seconds()
- self.assertTrue(delta <= 10)
-
- # Record a pong
- pongtime = datetime.datetime(2015, 10, 24, 9, 05)
- self.db.pong(address, pongtime, uuid)
-
- # Check for missing pongs
- missing = self.db.get_missing_pongs()
- self.assertEqual(0, len(missing))
-
- def testCloseReopen(self):
- listAddress = "list@example.com"
- address = "test@example.com"
- time = datetime.datetime(2015, 10, 24, 9, 00)
- uuid = "1234"
- expectedDelta = datetime.datetime.now() - time
-
- # Record a Ping
- self.db.ping(listAddress, address, time, uuid)
-
- # Close, Reopen
- self.db.close()
- self.db = canarydb.CanaryDB(self.tempdir + "canary.db")
-
- # Check for missing pongs
- missing = self.db.get_missing_pongs()
-
- self.assertEqual(1, len(missing))
- firstMissing = missing[0]
- self.assertEqual(4, len(firstMissing))
- self.assertEqual(listAddress, firstMissing[0])
- self.assertEqual(uuid, firstMissing[1])
- self.assertEqual(address, firstMissing[2])
- delta = firstMissing[3].total_seconds() - expectedDelta.total_seconds()
- self.assertTrue(delta <= 10)
-
- def testAccounts(self):
- listAddress = "list@example.org"
- address = "user@example.net"
- imapserver = "imap.example.net"
- password = "secretpassword"
-
- # Verify that no accounts exist
- accounts = self.db.get_accounts()
- self.assertEqual(0, len(accounts))
-
- # Add one account
- self.db.add_account(listAddress, address, imapserver, password)
-
- # Verify that the account exists
- accounts = self.db.get_accounts()
- self.assertEqual(1, len(accounts))
- self.assertEqual(listAddress, accounts[0][0])
- self.assertEqual(address, accounts[0][1])
- self.assertEqual(imapserver, accounts[0][2])
- self.assertEqual(password, accounts[0][3])
-
- # Remove the account
- self.db.remove_account(listAddress, address)
- accounts = self.db.get_accounts()
- self.assertEqual(0, len(accounts))
-
- def testGetRecipientsForList(self):
- listAddress1 = "list1@example.org"
- listAddress2 = "list2@example.org"
- imapserver = "imap.example.net"
- password = "secretpassword"
- address1 = "user1@example.net"
- address2 = "user2@example.net"
-
- # No accounts
- self.assertEqual([], self.db.get_recipients_for_list(listAddress1));
- self.assertEqual([], self.db.get_recipients_for_list(listAddress2));
-
- # One account
- self.db.add_account(listAddress1, address1, imapserver, password)
- self.assertEqual([address1], self.db.get_recipients_for_list(listAddress1));
- self.assertEqual([], self.db.get_recipients_for_list(listAddress2));
-
- # Two accounts
- self.db.add_account(listAddress1, address2, imapserver, password)
- self.assertEqual([address1, address2], self.db.get_recipients_for_list(listAddress1));
- self.assertEqual([], self.db.get_recipients_for_list(listAddress2));
-
- # Two lists
- self.db.add_account(listAddress2, address1, imapserver, password)
- self.assertEqual([address1, address2], self.db.get_recipients_for_list(listAddress1));
- self.assertEqual([address1], self.db.get_recipients_for_list(listAddress2));
+ def setUp(self):
+ self.tempdir = tempfile.mkdtemp()
+ self.db = canarydb.CanaryDB(self.tempdir + "canary.db")
+
+ def tearDown(self):
+ self.db.close()
+ shutil.rmtree(self.tempdir)
+
+ def testPingCheckPong(self):
+ listAddress = "list@example.com"
+ address = "test@example.com"
+ time = datetime.datetime(2015, 10, 24, 9, 00)
+ uuid = "1234"
+ expectedDelta = datetime.datetime.now() - time
+
+ # Record a Ping
+ self.db.ping(listAddress, address, time, uuid)
+
+ # Check for missing pongs
+ missing = self.db.get_missing_pongs()
+
+ self.assertEqual(1, len(missing))
+ firstMissing = missing[0]
+ self.assertEqual(4, len(firstMissing))
+ self.assertEqual(listAddress, firstMissing[0])
+ self.assertEqual(uuid, firstMissing[1])
+ self.assertEqual(address, firstMissing[2])
+ delta = firstMissing[3].total_seconds() - expectedDelta.total_seconds()
+ self.assertTrue(delta <= 10)
+
+ # Record a pong
+ pongtime = datetime.datetime(2015, 10, 24, 9, 05)
+ self.db.pong(address, pongtime, uuid)
+
+ # Check for missing pongs
+ missing = self.db.get_missing_pongs()
+ self.assertEqual(0, len(missing))
+
+ def testCloseReopen(self):
+ listAddress = "list@example.com"
+ address = "test@example.com"
+ time = datetime.datetime(2015, 10, 24, 9, 00)
+ uuid = "1234"
+ expectedDelta = datetime.datetime.now() - time
+
+ # Record a Ping
+ self.db.ping(listAddress, address, time, uuid)
+
+ # Close, Reopen
+ self.db.close()
+ self.db = canarydb.CanaryDB(self.tempdir + "canary.db")
+
+ # Check for missing pongs
+ missing = self.db.get_missing_pongs()
+
+ self.assertEqual(1, len(missing))
+ firstMissing = missing[0]
+ self.assertEqual(4, len(firstMissing))
+ self.assertEqual(listAddress, firstMissing[0])
+ self.assertEqual(uuid, firstMissing[1])
+ self.assertEqual(address, firstMissing[2])
+ delta = firstMissing[3].total_seconds() - expectedDelta.total_seconds()
+ self.assertTrue(delta <= 10)
+
+ def testAccounts(self):
+ listAddress = "list@example.org"
+ address = "user@example.net"
+ imapserver = "imap.example.net"
+ password = "secretpassword"
+
+ # Verify that no accounts exist
+ accounts = self.db.get_accounts()
+ self.assertEqual(0, len(accounts))
+
+ # Add one account
+ self.db.add_account(listAddress, address, imapserver, password)
+
+ # Verify that the account exists
+ accounts = self.db.get_accounts()
+ self.assertEqual(1, len(accounts))
+ self.assertEqual(listAddress, accounts[0][0])
+ self.assertEqual(address, accounts[0][1])
+ self.assertEqual(imapserver, accounts[0][2])
+ self.assertEqual(password, accounts[0][3])
+
+ # Remove the account
+ self.db.remove_account(listAddress, address)
+ accounts = self.db.get_accounts()
+ self.assertEqual(0, len(accounts))
+
+ def testGetRecipientsForList(self):
+ listAddress1 = "list1@example.org"
+ listAddress2 = "list2@example.org"
+ imapserver = "imap.example.net"
+ password = "secretpassword"
+ address1 = "user1@example.net"
+ address2 = "user2@example.net"
+
+ # No accounts
+ self.assertEqual([], self.db.get_recipients_for_list(listAddress1));
+ self.assertEqual([], self.db.get_recipients_for_list(listAddress2));
+
+ # One account
+ self.db.add_account(listAddress1, address1, imapserver, password)
+ self.assertEqual([address1], self.db.get_recipients_for_list(listAddress1));
+ self.assertEqual([], self.db.get_recipients_for_list(listAddress2));
+
+ # Two accounts
+ self.db.add_account(listAddress1, address2, imapserver, password)
+ self.assertEqual([address1, address2], self.db.get_recipients_for_list(listAddress1));
+ self.assertEqual([], self.db.get_recipients_for_list(listAddress2));
+
+ # Two lists
+ self.db.add_account(listAddress2, address1, imapserver, password)
+ self.assertEqual([address1, address2], self.db.get_recipients_for_list(listAddress1));
+ self.assertEqual([address1], self.db.get_recipients_for_list(listAddress2));