本文整理汇总了Python中slimta.relay.smtp.client.SmtpRelayClient类的典型用法代码示例。如果您正苦于以下问题:Python SmtpRelayClient类的具体用法?Python SmtpRelayClient怎么用?Python SmtpRelayClient使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了SmtpRelayClient类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_rcptto
def test_rcptto(self):
self.sock.sendall(b'RCPT TO:<recipient>\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Ok\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=self._socket_creator)
client._connect()
client._rcptto('recipient')
开发者ID:thestick613,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_relay_smtp_client.py
示例2: test_rset
def test_rset(self):
self.sock.sendall(b'RSET\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Ok\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=self._socket_creator)
client._connect()
client._rset()
开发者ID:thestick613,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_relay_smtp_client.py
示例3: test_connect_failure
def test_connect_failure(self):
def socket_creator(address):
raise socket_error(None, None)
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=socket_creator)
with self.assertRaises(TransientRelayError):
client._connect()
开发者ID:thestick613,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_relay_smtp_client.py
示例4: test_run_multiple
def test_run_multiple(self):
result1 = AsyncResult()
result2 = AsyncResult()
env1 = Envelope('[email protected]', ['[email protected]'])
env1.parse(b'From: [email protected]\r\n\r\ntest test\r\n')
env2 = Envelope('[email protected]', ['[email protected]'])
env2.parse(b'From: [email protected]\r\n\r\ntest test\r\n')
queue = BlockingDeque()
queue.append((result1, env1))
queue.append((result2, env2))
self.sock.recv(IsA(int)).AndReturn(b'220 Welcome\r\n')
self.sock.sendall(b'EHLO test\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250-Hello\r\n250 PIPELINING\r\n')
self.sock.sendall(b'MAIL FROM:<[email protected]>\r\nRCPT TO:<[email protected]>\r\nDATA\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Ok\r\n250 Ok\r\n354 Go ahead\r\n')
self.sock.sendall(b'From: [email protected]\r\n\r\ntest test\r\n.\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Ok\r\n')
self.sock.sendall(b'MAIL FROM:<[email protected]>\r\nRCPT TO:<[email protected]>\r\nDATA\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Ok\r\n250 Ok\r\n354 Go ahead\r\n')
self.sock.sendall(b'From: [email protected]\r\n\r\ntest test\r\n.\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Ok\r\n')
self.sock.sendall(b'QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn(b'221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient('addr', queue, socket_creator=self._socket_creator, ehlo_as='test', idle_timeout=0.0)
client._run()
self.assertEqual({'[email protected]': Reply('250', 'Ok')}, result1.get_nowait())
self.assertEqual({'[email protected]': Reply('250', 'Ok')}, result2.get_nowait())
开发者ID:thestick613,项目名称:python-slimta,代码行数:29,代码来源:test_slimta_relay_smtp_client.py
示例5: test_run_multiple
def test_run_multiple(self):
result1 = self.mox.CreateMock(AsyncResult)
result2 = self.mox.CreateMock(AsyncResult)
env1 = Envelope('[email protected]', ['[email protected]'])
env1.parse('From: [email protected]\r\n\r\ntest test\r\n')
env2 = Envelope('[email protected]', ['[email protected]'])
env2.parse('From: [email protected]\r\n\r\ntest test\r\n')
queue = BlockingDeque()
queue.append((result1, env1))
queue.append((result2, env2))
self.sock.recv(IsA(int)).AndReturn('220 Welcome\r\n')
self.sock.sendall('EHLO test\r\n')
self.sock.recv(IsA(int)).AndReturn('250-Hello\r\n250 PIPELINING\r\n')
self.sock.sendall('MAIL FROM:<[email protected]>\r\nRCPT TO:<[email protected]>\r\nDATA\r\n')
self.sock.recv(IsA(int)).AndReturn('250 Ok\r\n250 Ok\r\n354 Go ahead\r\n')
self.sock.sendall('From: [email protected]\r\n\r\ntest test\r\n.\r\n')
self.sock.recv(IsA(int)).AndReturn('250 Ok\r\n')
result1.set([None])
self.sock.sendall('MAIL FROM:<[email protected]>\r\nRCPT TO:<[email protected]>\r\nDATA\r\n')
self.sock.recv(IsA(int)).AndReturn('250 Ok\r\n250 Ok\r\n354 Go ahead\r\n')
self.sock.sendall('From: [email protected]\r\n\r\ntest test\r\n.\r\n')
self.sock.recv(IsA(int)).AndReturn('250 Ok\r\n')
result2.set([None])
self.sock.sendall('QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn('221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient(None, queue, socket_creator=self._socket_creator, ehlo_as='test', idle_timeout=0.0)
client._run()
开发者ID:MichaelEvanchik,项目名称:python-slimta,代码行数:29,代码来源:test_slimta_relay_smtp_client.py
示例6: test_handshake_starttls
def test_handshake_starttls(self):
sock = self.mox.CreateMockAnything()
sock.fileno = lambda: -1
def socket_creator(address):
return sock
sock.recv(IsA(int)).AndReturn("220 Welcome\r\n")
sock.sendall("EHLO test\r\n")
sock.recv(IsA(int)).AndReturn("250-Hello\r\n250 STARTTLS\r\n")
sock.sendall("STARTTLS\r\n")
sock.recv(IsA(int)).AndReturn("220 Go ahead\r\n")
sock.tls_wrapper(sock, self.tls_args).AndReturn(sock)
sock.sendall("EHLO test\r\n")
sock.recv(IsA(int)).AndReturn("250 Hello\r\n")
self.mox.ReplayAll()
client = SmtpRelayClient(
None,
self.queue,
socket_creator=socket_creator,
tls=self.tls_args,
tls_wrapper=sock.tls_wrapper,
ehlo_as="test",
)
client._connect()
client._handshake()
开发者ID:rafaelnovello,项目名称:python-slimta,代码行数:26,代码来源:test_slimta_relay_smtp_client.py
示例7: test_disconnect
def test_disconnect(self):
self.sock.sendall("QUIT\r\n")
self.sock.recv(IsA(int)).AndReturn("221 Goodbye\r\n")
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient(None, self.queue, socket_creator=self._socket_creator, ehlo_as="test")
client._connect()
client._disconnect()
开发者ID:rafaelnovello,项目名称:python-slimta,代码行数:8,代码来源:test_slimta_relay_smtp_client.py
示例8: test_disconnect
def test_disconnect(self):
self.sock.sendall(b'QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn(b'221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=self._socket_creator, ehlo_as='test')
client._connect()
client._disconnect()
开发者ID:thestick613,项目名称:python-slimta,代码行数:8,代码来源:test_slimta_relay_smtp_client.py
示例9: test_disconnect_failure
def test_disconnect_failure(self):
self.sock.sendall(b'QUIT\r\n')
self.sock.recv(IsA(int)).AndRaise(socket_error(None, None))
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=self._socket_creator, ehlo_as='test')
client._connect()
client._disconnect()
开发者ID:thestick613,项目名称:python-slimta,代码行数:8,代码来源:test_slimta_relay_smtp_client.py
示例10: test_handshake_tls_immediately
def test_handshake_tls_immediately(self):
tls_wrapper = self.mox.CreateMockAnything()
tls_wrapper(self.sock, self.tls_args).AndReturn(self.sock)
self.sock.recv(IsA(int)).AndReturn(b'220 Welcome\r\n')
self.sock.sendall(b'EHLO there\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250 Hello\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=self._socket_creator, tls=self.tls_args, tls_wrapper=tls_wrapper, tls_immediately=True, ehlo_as='there')
client._connect()
client._handshake()
开发者ID:dldorsey,项目名称:python-slimta,代码行数:10,代码来源:test_slimta_relay_smtp_client.py
示例11: test_handshake_authenticate_badcreds
def test_handshake_authenticate_badcreds(self):
self.sock.recv(IsA(int)).AndReturn(b'220 Welcome\r\n')
self.sock.sendall(b'EHLO there\r\n')
self.sock.recv(IsA(int)).AndReturn(b'250-Hello\r\n250 AUTH PLAIN\r\n')
self.sock.sendall(b'AUTH PLAIN AHRlc3RAZXhhbXBsZS5jb20AcGFzc3dk\r\n')
self.sock.recv(IsA(int)).AndReturn(b'535 Nope!\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient(('addr', 0), self.queue, socket_creator=self._socket_creator, credentials=('[email protected]', 'passwd'), ehlo_as='there', auth_mechanism=b'PLAIN')
client._connect()
with self.assertRaises(PermanentRelayError):
client._handshake()
开发者ID:slimta,项目名称:python-slimta,代码行数:11,代码来源:test_slimta_relay_smtp_client.py
示例12: test_run_nomessages
def test_run_nomessages(self):
queue = PriorityQueue()
self.sock.recv(IsA(int)).AndReturn('220 Welcome\r\n')
self.sock.sendall('EHLO test\r\n')
self.sock.recv(IsA(int)).AndReturn('250-Hello\r\n250 PIPELINING\r\n')
self.sock.sendall('QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn('221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient(None, queue, socket_creator=self._socket_creator, ehlo_as='test', idle_timeout=0)
client._run()
开发者ID:RoboticCheese,项目名称:python-slimta,代码行数:11,代码来源:test_slimta_relay_smtp_client.py
示例13: test_handshake_tls_immediately
def test_handshake_tls_immediately(self):
sock = self.mox.CreateMockAnything()
sock.fileno = lambda: -1
def socket_creator(address):
return sock
sock.tls_wrapper(sock, self.tls_args).AndReturn(sock)
sock.recv(IsA(int)).AndReturn(b'220 Welcome\r\n')
sock.sendall(b'EHLO test\r\n')
sock.recv(IsA(int)).AndReturn(b'250 Hello\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=socket_creator, tls=self.tls_args, tls_wrapper=sock.tls_wrapper, tls_immediately=True, ehlo_as='test')
client._connect()
client._handshake()
开发者ID:thestick613,项目名称:python-slimta,代码行数:13,代码来源:test_slimta_relay_smtp_client.py
示例14: test_handshake_authenticate
def test_handshake_authenticate(self):
sock = self.mox.CreateMockAnything()
sock.fileno = lambda: -1
def socket_creator(address):
return sock
sock.recv(IsA(int)).AndReturn(b'220 Welcome\r\n')
sock.sendall(b'EHLO test\r\n')
sock.recv(IsA(int)).AndReturn(b'250-Hello\r\n250 AUTH PLAIN\r\n')
sock.sendall(b'AUTH PLAIN AHRlc3RAZXhhbXBsZS5jb20AcGFzc3dk\r\n')
sock.recv(IsA(int)).AndReturn(b'235 Ok\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient('addr', self.queue, socket_creator=socket_creator, credentials=('[email protected]', 'passwd'), ehlo_as='test')
client._connect()
client._handshake()
开发者ID:thestick613,项目名称:python-slimta,代码行数:14,代码来源:test_slimta_relay_smtp_client.py
示例15: test_run_banner_failure
def test_run_banner_failure(self):
result = self.mox.CreateMock(AsyncResult)
env = Envelope('[email protected]', ['[email protected]'])
env.parse('From: [email protected]\r\n\r\ntest test\r\n')
queue = BlockingDeque()
queue.append((result, env))
self.sock.recv(IsA(int)).AndReturn('520 Not Welcome\r\n')
result.set_exception(IsA(PermanentRelayError))
self.sock.sendall('QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn('221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient(None, queue, socket_creator=self._socket_creator, ehlo_as='test')
client._run()
开发者ID:MichaelEvanchik,项目名称:python-slimta,代码行数:14,代码来源:test_slimta_relay_smtp_client.py
示例16: test_run_timeout
def test_run_timeout(self):
result = self.mox.CreateMock(AsyncResult)
env = Envelope('[email protected]', ['[email protected]'])
env.parse('From: [email protected]\r\n\r\ntest test\r\n')
queue = BlockingDeque()
queue.append((result, env))
self.sock.recv(IsA(int)).AndRaise(Timeout(0.0))
result.ready().AndReturn(False)
result.set_exception(IsA(TransientRelayError))
self.sock.sendall('QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn('221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient(None, queue, socket_creator=self._socket_creator, ehlo_as='test')
client._run()
开发者ID:MichaelEvanchik,项目名称:python-slimta,代码行数:15,代码来源:test_slimta_relay_smtp_client.py
示例17: test_run_banner_failure
def test_run_banner_failure(self):
result = AsyncResult()
env = Envelope('[email protected]', ['[email protected]'])
env.parse(b'From: [email protected]\r\n\r\ntest test\r\n')
queue = BlockingDeque()
queue.append((result, env))
self.sock.recv(IsA(int)).AndReturn(b'520 Not Welcome\r\n')
self.sock.sendall(b'QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn(b'221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient('addr', queue, socket_creator=self._socket_creator, ehlo_as='test')
client._run()
with self.assertRaises(PermanentRelayError):
result.get_nowait()
开发者ID:thestick613,项目名称:python-slimta,代码行数:15,代码来源:test_slimta_relay_smtp_client.py
示例18: test_run_smtperror
def test_run_smtperror(self):
result = self.mox.CreateMock(AsyncResult)
env = Envelope("[email protected]", ["[email protected]"])
env.parse("From: [email protected]\r\n\r\ntest test\r\n")
queue = BlockingDeque()
queue.append((result, env))
self.sock.recv(IsA(int)).AndRaise(SmtpError("test error"))
result.ready().AndReturn(False)
result.set_exception(IsA(TransientRelayError))
self.sock.sendall("QUIT\r\n")
self.sock.recv(IsA(int)).AndReturn("221 Goodbye\r\n")
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient(None, queue, socket_creator=self._socket_creator, ehlo_as="test")
client._run()
开发者ID:rafaelnovello,项目名称:python-slimta,代码行数:15,代码来源:test_slimta_relay_smtp_client.py
示例19: test_handshake_authenticate_badcreds
def test_handshake_authenticate_badcreds(self):
sock = self.mox.CreateMockAnything()
sock.fileno = lambda: -1
def socket_creator(address):
return sock
sock.recv(IsA(int)).AndReturn('220 Welcome\r\n')
sock.sendall('EHLO test\r\n')
sock.recv(IsA(int)).AndReturn('250-Hello\r\n250 AUTH PLAIN\r\n')
sock.sendall('AUTH PLAIN AHRlc3RAZXhhbXBsZS5jb20AcGFzc3dk\r\n')
sock.recv(IsA(int)).AndReturn('535 Nope!\r\n')
self.mox.ReplayAll()
client = SmtpRelayClient(None, self.queue, socket_creator=socket_creator, credentials=('[email protected]', 'passwd'), ehlo_as='test')
client._connect()
with self.assertRaises(PermanentRelayError):
client._handshake()
开发者ID:MichaelEvanchik,项目名称:python-slimta,代码行数:15,代码来源:test_slimta_relay_smtp_client.py
示例20: test_run_random_exception
def test_run_random_exception(self):
result = AsyncResult()
env = Envelope('[email protected]', ['[email protected]'])
env.parse(b'From: [email protected]\r\n\r\ntest test\r\n')
queue = BlockingDeque()
queue.append((result, env))
self.sock.recv(IsA(int)).AndRaise(ValueError('test error'))
self.sock.sendall(b'QUIT\r\n')
self.sock.recv(IsA(int)).AndReturn(b'221 Goodbye\r\n')
self.sock.close()
self.mox.ReplayAll()
client = SmtpRelayClient('addr', queue, socket_creator=self._socket_creator, ehlo_as='test')
with self.assertRaises(ValueError):
client._run()
with self.assertRaises(ValueError):
result.get_nowait()
开发者ID:thestick613,项目名称:python-slimta,代码行数:16,代码来源:test_slimta_relay_smtp_client.py
注:本文中的slimta.relay.smtp.client.SmtpRelayClient类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论