Anonymous avatar Anonymous committed 2968eaa

added some basic test cases

Comments (0)

Files changed (3)

pymta/processor.py

         self.state = StateMachine('_state', initial_state='new')
         self._add_state('new',     'GREET', 'greeted')
         self._add_state('greeted', 'HELO',  'identify')
-        self._add_state('greeted', 'EHLO',  'identify')
+        self._add_state('identify', 'QUIT',  'finished')
         self._add_state('greeted', 'QUIT',  'finished')
+        self.valid_commands = [command for new_state, command in self.state.states]
     
     
     def _dispatch_commands(self, from_state, to_state, smtp_command, ob):
             # SMTP commands must be treated as case-insensitive
             self.state.execute(self, command)
         except StateMachineError:
-            base_msg = 'Command "%s" is not allowed here, expected on of %s'
-            msg = base_msg % (command, self.state.transitions(self))
-            print msg
-            self.reply(502, 'Error: %s not allowed here' % smtp_command)
+            if command not in self.valid_commands:
+                self.reply(500, 'unrecognized command')
+            else:
+                msg = 'Command "%s" is not allowed here' % smtp_command
+                allowed_transitions = self.state.transitions(self)
+                if len(allowed_transitions) > 0:
+                      msg += ', expected on of %s' % allowed_transitions
+                self.reply(502, msg)
         self._command_arguments = None
     
     
     def reply(self, code, text):
         """This method returns a message to the client (actually the session 
         object is responsible of actually pushing the bits)."""
-        self._session.push('%s %s' % (code, text))
+        self._session.push(code, text)
     
     
     def close_connection(self):
     def smtp_greet(self):
         """This method handles not a real smtp command. It is called when a new
         connection was accepted by the server."""
-        self.reply(220, '%s %s' % (self._fqdn, self._server.version))
+        primary_hostname = self._session.primary_hostname
+        reply_text = '%s Hello %s' % (primary_hostname, self.remote_ip_string)
+        self.reply(220, reply_text)
+    
+    def smtp_quit(self):
+        primary_hostname = self._session.primary_hostname
+        reply_text = '%s closing connection' % primary_hostname
+        self.reply(221, reply_text)
+        self._session.close_when_done()
+    
+    def smtp_helo(self):
+        helo_string = self._command_arguments
+        # We could store the helo string for a later check
+        primary_hostname = self._session.primary_hostname
+        self.reply(250, primary_hostname)
+        
+        
 
 

pymta/smtp_session.py

     
     def primary_hostname(self):
         return self._server.primary_hostname
-    _fqdn = property(primary_hostname)
+    primary_hostname = property(primary_hostname)
     
     
     # -------------------------------------------------------------------------

tests/basic_message_sending_test.py

+# -*- coding: UTF-8 -*-
+
+from unittest import TestCase
+
+from pymta import SMTPProcessor
+
+class MockSession(object):
+    primary_hostname = 'localhost'
+    
+    def __init__(self):
+        self.replies = []
+        self.open = True
+    
+    def push(self, code, text):
+        self.replies.append((code, text))
+    
+    def close_when_done(self):
+        assert self.open
+        self.open = False
+    
+
+
+class BasicMessageSendingTest(TestCase):
+
+    def setUp(self):
+        self.session = MockSession()
+        self.processor = SMTPProcessor(session=self.session)
+        self.processor.new_connection('127.0.0.1', 4567)
+    
+    def _close_connection(self):
+        self.processor.handle_input('quit')
+        self.assertTrue(len(self.session.replies) >= 1)
+        code, reply_text = self.session.replies[-1]
+        self.assertTrue(221, code)
+        self.assertEqual('localhost closing connection', reply_text)
+        self.assertEqual(False, self.session.open)
+    
+    def test_new_connection(self):
+        self.assertEqual(1, len(self.session.replies))
+        code, reply_text = self.session.replies[-1]
+        self.assertEqual(220, code)
+        self.assertEqual('localhost Hello 127.0.0.1', reply_text)
+        self._close_connection()
+    
+    def test_send_helo(self):
+        self.processor.handle_input('helo', 'foo.example.com')
+        self.assertEqual(2, len(self.session.replies))
+        code, reply_text = self.session.replies[-1]
+        self.assertEqual(250, code)
+        self.assertEqual('localhost', reply_text)
+        self._close_connection()
+
+    def test_reject_duplicated_helo(self):
+        self.processor.handle_input('helo', 'foo.example.com')
+        self.processor.handle_input('helo', 'foo.example.com')
+        self.assertEqual(3, len(self.session.replies))
+        code, reply_text = self.session.replies[-1]
+        self.assertEqual(502, code)
+        expected_message = 'Command "helo" is not allowed here'
+        self.assertTrue(reply_text.startswith(expected_message), reply_text)
+        self._close_connection()
+
+    def test_invalid_commands_are_recognized(self):
+        self.processor.handle_input('invalid')
+        self.assertEqual(2, len(self.session.replies))
+        code, reply_text = self.session.replies[-1]
+        self.assertEqual(500, code)
+        self.assertEqual('unrecognized command', reply_text)
+        self._close_connection()
+
+    def test_send_simple_mail(self):
+        self.processor.handle_input('MAIL FROM', 'foo@example.com')
+        self.processor.handle_input('RCPT TO', 'bar@example.com')
+        msg = 'Subject: Test\r\n\r\nJust testing...\r\n'
+        self.processor.handle_input('DATA', msg)
+        self._close_connection()
+
+
+
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.