fixed and tested

.env.example

@@ -0,0 +1,7 @@
+FROM_ADDRESS=<email address of the sender>
+SMTP_SSL_MODE=STARTLS or SSL or nothing
+SMTP_HOST=<your SMTP host domain>
+SMTP_PORT=<your SMTP port>
+SMTP_USERNAME=<user name for connecting to SMTP host>
+SMTP_PASSWORD=<passw ord for connecting to SMTP host>
+MAILDROP_INBOX=<your maildrop test inbox>

.gitignore

@@ -0,0 +1,2 @@
+.env
+__pycache__/

README.md

@@ -7,7 +7,9 @@ You create a new reader with `MailDropReader(<your maildrop.cc inbox name>)
 The methods are:
 * __status()__: provides the current maildrop.cc status. Returns 'operational' or an error string from the server
 * __ping(string)__: pings the maildrop.cc server with the given string. Returns 'pong <string>'
-* __inbox()__: returns all messages of your inbox (currently returns ALL messages, the filters aren't working). Returns a list of messages with only basic fields filled.
+* __inbox()__: returns all messages of your inbox 
+  Returns a list of messages with only basic fields filled.
+  __(currently returns ALL messages, the filters aren't working)__. 
 * __message(message_id)__: returns a full message including its body, its sender IP, ...
 * __delete__(message_id)__: deletes a message by its id. Returns True if ok
 * __statistics()__: returns maildrop.cc statistics. Returns a tuple (blocked, saved)
@@ -25,3 +27,7 @@ for msg in msgs:
   print(f"content: {message.html}, ip={message.ip}, headerfrom={message.headerfrom}"
 ```
 
+## Testing
+To test the module, clone the repo, then copy `.env.example` in `.env` and provide the email sending settings.
+These settings are used to send emails to maildrop.cc 
+Then run `python test_maildrop.py`

maildrop.py → maildrop/__init__.py

@@ -1,5 +1,15 @@
 import requests
 import json
+import certifi
+import urllib3
+import logging
+
+logger = logging.getLogger(__name__)
+
+http = urllib3.PoolManager(
+    cert_reqs="CERT_REQUIRED",
+    ca_certs=certifi.where()
+)
 
 class MailDropMessage:
       def __init__(self, id, headerfrom = None, subject = None, date = None, 
@@ -41,15 +51,17 @@ def _call_api(self, query: MailDropQuery):
         fields = ' '.join(s for s in query.fields)
         if fields: fields = f'{{ {fields} }}'
         query_data = f'{{"query": "{query.type} {query.name} {{ {query.query} {fields} }}"}}'
-        print(">>> query=", query_data)
+        logger.debug(">>> maildrop query=", query_data)
         response = requests.post(self._maildrop_api, 
                                 headers={'content-type': 'application/json'},
                                 data=query_data, 
-                                verify=False)
+                                verify=True)
         if response.status_code != 200:
+          logger.debug(f"inbox call result: {response.text}")
           raise ValueError(response.content)
-
-        return json.loads(response.text)
+
+        json_data = json.loads(response.text)
+        return json_data['data']
 
     def ping(self, message: str ="hello, world!") -> str:
         query = MailDropReader.MailDropQuery(
@@ -58,9 +70,10 @@ def ping(self, message: str ="hello, world!") -> str:
           query = f'ping(message:\\"{message}\\")',
         )
         res = self._call_api(query)
-        return res['data']['ping']
+        logger.debug(f"ping result: {res['ping']}")
+        return res['ping']
 
-    def inbox(self, filters: dict = None) -> list[MailDropMessage]:
+    def inbox(self, filters: dict = None):
         """gets all messages from maildrop inbox. Can be filtered by a dict of MailDropMessage fields"""
         filters_str = ' ' + ' '.join([f'{k}: \\"{v}\\"' for k, v in filters.items()]) if filters else ''
         query = MailDropReader.MailDropQuery(
@@ -71,8 +84,8 @@ def inbox(self, filters: dict = None) -> list[MailDropMessage]:
             fields = ['id', 'mailfrom', 'subject', 'date']
         )
         jdata = self._call_api(query)
-        # print(jdata['data']['inbox'])
-        return [MailDropMessage(**mess) for mess in jdata['data']['inbox']]
+        logger.debug(f"inbox call result: {jdata['inbox']}")
+        return [MailDropMessage(**mess) for mess in jdata['inbox']]
 
     def message(self, message_id) -> MailDropMessage:
         """get a full message content by its id"""
@@ -83,6 +96,8 @@ def message(self, message_id) -> MailDropMessage:
           fields =  ['id', 'headerfrom', 'subject', 'date', 'html', 'ip', 'mailfrom', 'data', 'rcptto', 'helo']
         )
         mess = self._call_api(query)
+        mess = mess['message']
+        logger.debug(f"message {message_id} call result: {mess}")
         return MailDropMessage(**mess)
 
     def delete(self, message_id) -> bool:
@@ -93,7 +108,8 @@ def delete(self, message_id) -> bool:
           query = f'delete(mailbox:\\"{self.inbox_name}\\", id:\\"{message_id}\\")',
         )
         res = self._call_api(query)
-        return res['data']['delete']
+        logger.debug(f"delete {message_id} call result: {res['delete']}")
+        return res['delete']
 
     def status(self) -> str:
         """Returns the maildrop platform status. Can be 'operational' or an error string"""
@@ -103,17 +119,17 @@ def status(self) -> str:
           query="status"
         )
         res = self._call_api(query)
-        return res['data']['status']
+        return res['status']
 
-    def statistics(self) -> tuple[int,int]:
+    def statistics(self):
         """returns maildrop statistics in the form of a tuple (blocked, saved)"""
         query = MailDropReader.MailDropQuery(
           qtype = "query",
           name = "statistics",
           query="statistics { blocked saved }"
         )
         res = self._call_api(query)
-        return (res['data']['statistics']['blocked'], res['data']['statistics']['saved'])
+        return (res['statistics']['blocked'], res['statistics']['saved'])
 
     def altinbox(self) -> str:
         """returns an alias for the inbox that can be used in subsequent MailDropReaders"""
@@ -123,4 +139,4 @@ def altinbox(self) -> str:
           query = f'altinbox(mailbox:\\"{self.inbox_name}\\")',
         )
         res = self._call_api(query)
-        return res['data']['altinbox']
+        return res['altinbox']

test_maildrop.py

@@ -1,11 +1,11 @@
-from .maildrop import MailDropReader
-from django.test import TestCase
-from django.core import mail
-from django.conf import settings
-from django.utils.html import strip_tags
+from maildrop import MailDropReader
+import smtplib
+from email.message import EmailMessage
+from dotenv import load_dotenv
+import re, os, random, string
 
 # change to true to trace all requests to maildrop.cc
-TRACE_REQUESTS=True
+TRACE_REQUESTS=False
 if TRACE_REQUESTS:
 	import http.client as http_client
 	import logging
@@ -16,91 +16,133 @@
 	requests_log.setLevel(logging.DEBUG)
 	requests_log.propagate = True
 
+load_dotenv()
 
-maildrop_inbox = 'test-maildrop'
-message_test_body = """
+def generate_random_string(length):
+    return ''.join(random.choice(string.digits) for _ in range(length))
+
+def do(func, *args):
+	if not callable(func):
+		raise ValueError(f"{func.name} is not callable")
+	print(f">>> running test {func.__name__} with args: {args}")
+	res = func(*args) if len(args) > 0 else func()
+	print(f"<<< end test {func.__name__} with result: {res}")
+
+def strip_tags(raw_html):
+  cleanr = re.compile('<.*?>')
+  cleantext = re.sub(cleanr, '', raw_html)
+  return cleantext
+
+maildrop_inbox = os.environ['MAILDROP_INBOX']
+message_test_body = f"""
 <html>
 	<header>
 		<style>
-			body { color: red;}
+			body {{ color: red;}}
 		</style>
 	</header>
 	<body>
-		<p>This is the test mail body</p>
+		<h1>Test maildropy on inbox {maildrop_inbox}</h1>
+		<p>This is the test mail body #{generate_random_string(8)}</p>
 	</body>
 </html>
 """
-
-class MailDropTests(TestCase):
-
-	def setUp(self):
-		self.maildrop = MailDropReader(maildrop_inbox)
-
-	def test_ping(self):
-		ping_str = "test python maildrop"
-		res = self.maildrop.ping(ping_str)
-		self.assertEqual(res, f'pong {ping_str}')
-		# print(res)
-
-	def send_test_mail(self, subject='test maildrop'):
-		self.assertIsNotNone(settings.EMAIL_HOST_USER)
-		mail.send_mail(f'{subject}', strip_tags(message_test_body), settings.DEFAULT_FROM_EMAIL,
-			recipient_list=[maildrop_inbox], html_message=message_test_body, fail_silently=False)
-		self.assertEqual(len(mail.outbox), 1)
-		self.assertEqual(mail.outbox[0].subject, subject)
-
-	def test_inbox(self):
-		self.send_test_mail('testing inbox')
-		msgs = self.maildrop.inbox()
-		self.assertEqual(len(msgs), 1)
-		msg = msgs[0]
-		# print("headerfrom=", msg.headerfrom)
-		self.assertEqual(msg.mailfrom, settings.DEFAULT_FROM_EMAIL)
-
-	# def test_filtered_inbox(self):
-	# 	subject = 'testing delete'
-	# 	self.send_test_mail(subject)
-	# 	msgs = self.maildrop.inbox({'subject': subject})
-	# 	self.assertCountEqual(msgs, 1)
-	# 	msg = msgs[0]
-	# 	self.assertEqual(msg.subject, subject)
-
-	def test_status(self):
-		self.assertEqual(self.maildrop.status(), 'operational')
-
-	def test_statistics(self):
-		blocked, saved = self.maildrop.statistics()
-		self.assertGreater(blocked, 1)
-		self.assertGreater(saved, 1)
-
-	def test_alias(self):
-		alias = self.maildrop.altinbox()
-		print('alias=', alias)
-		self.assertIsNotNone(alias)
-
-	def test_message(self):
-		subject = 'test read message'
-		self.send_test_mail(subject)
-		msgs = self.maildrop.inbox()
-
-		for m in msgs:
-			msg = self.maildrop.message(m.id)
-			self.assertIsNotNone(msg)
-			self.assertEqual(msg.mailfrom, settings.DEFAULT_FROM_EMAIL)
-
-		self.assertIn(subject, [msg.subject for msg in msgs])
-
-	def test_delete_message(self):
-		msgs = self.maildrop.inbox()
-		nmsgs = len(msgs)
-		if nmsgs == 0:
-			self.send_test_mail('testing inbox')
-			msgs = self.maildrop.inbox()
-			nmsgs = len(msgs)
-		self.assertGreater(nmsgs, 0)
-		msg = msgs[0]
-		self.maildrop.delete(msg.id)
-		msgs = self.maildrop.inbox()
-		self.assertEqual(len(msgs), nmsgs - 1)
-
-
+txt_message_test_body = strip_tags(message_test_body)
+
+def send_test_mail(subject='test maildrop'):
+	msg = EmailMessage()
+	msg.set_content(txt_message_test_body)
+	msg['Subject'] = subject
+	msg['From'] = os.environ['FROM_ADDRESS']
+	msg['To'] = f'{maildrop_inbox}@maildrop.cc'
+	msg.add_alternative(message_test_body, subtype='html')
+	if os.environ['SMTP_SSL_MODE'] == 'SSL':
+		s = smtplib.SMTP_SSL(os.environ['SMTP_HOST'], os.environ['SMTP_PORT'])
+	else: 
+		s = smtplib.SMTP(os.environ['SMTP_HOST'], os.environ['SMTP_PORT'])
+	if os.environ['SMTP_SSL_MODE'] == 'STARTLS':
+		s.starttls()
+	s.login(os.environ['SMTP_USERNAME'], os.environ['SMTP_PASSWORD'])
+	s.send_message(msg)
+	s.quit()
+
+reader = MailDropReader(maildrop_inbox)
+
+def test_ping():
+	ping_str = "test python maildrop"
+	res = reader.ping(ping_str)
+	assert res == f'pong {ping_str}', f'unexpected pong: {res}'
+	return res
+
+def test_inbox(nbmsgs):
+	msgs = reader.inbox()
+	assert len(msgs) == nbmsgs, f'unexpected number of messages: {len(msgs)}'
+	msg = msgs[0]
+	assert msg.mailfrom == os.environ['FROM_ADDRESS'], f'unexpected sender: {msg.mailfrom}'
+	return len(msgs)
+
+# DOES NOT WORK CURRENTLY
+# def test_filtered_inbox():
+# 	subject = 'testing delete'
+# 	send_test_mail(subject)
+# 	msgs = reader.inbox({'subject': subject})
+# 	assert len(msgs) == 1
+# 	msg = msgs[0]
+# 	assert msg.subject == subject
+
+def test_status():
+	status = reader.status()
+	assert status == 'operational', "maildrop status not operational"
+	return status
+
+def test_statistics():
+	blocked, saved = reader.statistics()
+	assert blocked >= 1, f'unexpected stat: blocked = {blocked}'
+	assert saved >= 1, f'unexpected stat: saved = {saved}'
+	return (blocked, saved)
+
+def test_alias():
+	alias = reader.altinbox()
+	assert alias is not None, "alias not given by maildrop"
+	return alias
+
+def test_message(subject):
+	msgs = reader.inbox()
+	msg_found = 0
+	for m in msgs:
+		msg = reader.message(m.id)
+		assert msg is not None, "null msg"
+		assert msg.mailfrom == os.environ['FROM_ADDRESS'], f"msg not sent by right sender: {msg.mailfrom}"
+		assert msg.html == message_test_body, f"unexpected msg content: {msg.html}"
+		if msg.subject == subject:
+			msg_found += 1
+			content = msg.html
+
+	assert msg_found == 1, f"subject '{subject} not found in messages or found several times: {msg_found}"
+	assert content is not None, f"content of message with subject:{subject} not found"
+	return content
+
+def test_delete_message():
+	msgs = reader.inbox()
+	nmsgs = len(msgs)
+	msg = msgs[0]
+	id = msg.id
+	reader.delete(id)
+	msgs = reader.inbox()
+	assert len(msgs) == nmsgs - 1, f"messages number should have decreased of one"
+	assert id not in [msg.id for msg in msgs], f"message id {id} still in msgs list after deletion"
+	return "deleted"
+
+do(test_status)
+do(test_statistics)
+do(test_ping)
+do(test_alias)
+
+nbmsgs = 3
+for _ in range(nbmsgs):
+	subject = f'testing message #{generate_random_string(8)}'
+	do(send_test_mail, subject)
+
+do(test_inbox, nbmsgs)
+do(test_message, subject)
+do(test_delete_message)