• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python ptrlookup.PtrLookup类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中slimta.util.ptrlookup.PtrLookup的典型用法代码示例。如果您正苦于以下问题:Python PtrLookup类的具体用法?Python PtrLookup怎么用?Python PtrLookup使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了PtrLookup类的16个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_run_nxdomain

 def test_run_nxdomain(self):
     self.mox.StubOutWithMock(dns_resolver, 'query')
     dns_resolver.query(IgnoreArg(), 'PTR').AndRaise(NXDOMAIN)
     self.mox.ReplayAll()
     ptr = PtrLookup('127.0.0.1')
     assert_is_instance(ptr, gevent.Greenlet)
     assert_is_none(ptr._run())
开发者ID:drewlander,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_util_ptrlookup.py


示例2: test_finish

 def test_finish(self):
     self.mox.StubOutWithMock(dns_resolver, 'query')
     dns_resolver.query(IgnoreArg(), 'PTR').AndReturn(['example.com'])
     self.mox.ReplayAll()
     ptr = PtrLookup('127.0.0.1')
     ptr.start()
     self.assertEqual('example.com', ptr.finish(runtime=1.0))
开发者ID:sk1p,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_util_ptrlookup.py


示例3: test_run_dnsexception

 def test_run_dnsexception(self):
     self.mox.StubOutWithMock(dns_resolver, 'query')
     dns_resolver.query(IgnoreArg(), 'PTR').AndRaise(DNSException)
     self.mox.ReplayAll()
     ptr = PtrLookup('127.0.0.1')
     self.assertIsInstance(ptr, gevent.Greenlet)
     self.assertIsNone(ptr._run())
开发者ID:sk1p,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_util_ptrlookup.py


示例4: test_finish

 def test_finish(self):
     self.mox.StubOutWithMock(socket, "gethostbyaddr")
     socket.gethostbyaddr("127.0.0.1").AndReturn(("example.com", None, None))
     self.mox.ReplayAll()
     ptr = PtrLookup("127.0.0.1")
     ptr.start()
     self.assertEqual("example.com", ptr.finish(runtime=1.0))
开发者ID:thestick613,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_util_ptrlookup.py


示例5: test_run_greenletexit

 def test_run_greenletexit(self):
     self.mox.StubOutWithMock(socket, "gethostbyaddr")
     socket.gethostbyaddr("127.0.0.1").AndRaise(gevent.GreenletExit)
     self.mox.ReplayAll()
     ptr = PtrLookup("abcd")
     self.assertIsInstance(ptr, gevent.Greenlet)
     self.assertIsNone(ptr._run())
开发者ID:thestick613,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_util_ptrlookup.py


示例6: test_run_no_result

 def test_run_no_result(self):
     self.mox.StubOutWithMock(socket, 'gethostbyaddr')
     socket.gethostbyaddr('127.0.0.1').AndRaise(socket.herror)
     self.mox.ReplayAll()
     ptr = PtrLookup('127.0.0.1')
     self.assertIsInstance(ptr, gevent.Greenlet)
     self.assertIsNone(ptr._run())
开发者ID:madhugb,项目名称:python-slimta,代码行数:7,代码来源:test_slimta_util_ptrlookup.py


示例7: test_finish

 def test_finish(self):
     self.mox.StubOutWithMock(socket, 'gethostbyaddr')
     socket.gethostbyaddr('127.0.0.1').AndReturn(
         ('example.com', None, None))
     self.mox.ReplayAll()
     ptr = PtrLookup('127.0.0.1')
     ptr.start()
     self.assertEqual('example.com', ptr.finish(runtime=1.0))
开发者ID:madhugb,项目名称:python-slimta,代码行数:8,代码来源:test_slimta_util_ptrlookup.py


示例8: test_finish_timeout

 def test_finish_timeout(self):
     self.mox.StubOutWithMock(dns_resolver, 'query')
     def long_sleep(*args):
         gevent.sleep(1.0)
     dns_resolver.query(IgnoreArg(), 'PTR').WithSideEffects(long_sleep)
     self.mox.ReplayAll()
     ptr = PtrLookup('127.0.0.1')
     ptr.start()
     self.assertIsNone(ptr.finish(runtime=0.0))
开发者ID:sk1p,项目名称:python-slimta,代码行数:9,代码来源:test_slimta_util_ptrlookup.py


示例9: test_finish_timeout

    def test_finish_timeout(self):
        def long_sleep(*args):
            gevent.sleep(1.0)

        self.mox.StubOutWithMock(socket, "gethostbyaddr")
        socket.gethostbyaddr("127.0.0.1").WithSideEffects(long_sleep)
        self.mox.ReplayAll()
        ptr = PtrLookup("127.0.0.1")
        ptr.start()
        self.assertIsNone(ptr.finish(runtime=0.001))
开发者ID:thestick613,项目名称:python-slimta,代码行数:10,代码来源:test_slimta_util_ptrlookup.py


示例10: test_from_getsockname

 def test_from_getsockname(self):
     sock = self.mox.CreateMockAnything()
     sock.getsockname().AndReturn(('127.0.0.1', 0))
     self.mox.ReplayAll()
     ptr, port = PtrLookup.from_getsockname(sock)
     self.assertEqual(0, port)
     self.assertIsInstance(ptr, PtrLookup)
     self.assertEqual('127.0.0.1', ptr.ip)
开发者ID:sk1p,项目名称:python-slimta,代码行数:8,代码来源:test_slimta_util_ptrlookup.py


示例11: __call__

 def __call__(self, environ, start_response):
     ptr_lookup = PtrLookup(environ.get('REMOTE_ADDR', '0.0.0.0'))
     ptr_lookup.start()
     try:
         self._validate_request(environ)
         env = self._get_envelope(environ)
         self._add_envelope_extras(environ, env, ptr_lookup)
         self._enqueue_envelope(env)
     except WsgiResponse as res:
         start_response(res.status, res.headers)
         return res.data
     except Exception as exc:
         logging.log_exception(__name__)
         msg = '{0!s}\n'.format(exc)
         content_length = str(len(msg))
         headers = [('Content-Length', content_length),
                    ('Content-Type', 'text/plain')]
         start_response('500 Internal Server Error', headers)
         return [msg]
     finally:
         ptr_lookup.kill(block=False)
开发者ID:slimta,项目名称:python-slimta,代码行数:21,代码来源:wsgi.py


示例12: SmtpSession

class SmtpSession(object):

    def __init__(self, address, validator_class, handoff):
        self.extended_smtp = False
        self.security = None
        self.address = address
        self.reverse_address = None
        self.handoff = handoff
        self.validators = validator_class(self) if validator_class else None

        self.envelope = None
        self.ehlo_as = None
        self.auth = None

    def _call_validator(self, command, *args):
        method = 'handle_'+command
        if hasattr(self.validators, method):
            getattr(self.validators, method)(*args)

    @property
    def protocol(self):
        proto = 'SMTP'
        if self.extended_smtp:
            proto = 'ESMTP'
        if self.security == 'TLS':
            proto += 'S'
        if self.auth:
            proto += 'A'
        return proto

    def BANNER_(self, reply):
        self._ptr_lookup = PtrLookup(self.address[0])
        self._ptr_lookup.start()
        self._call_validator('banner', reply, self.address)

    def EHLO(self, reply, ehlo_as):
        self._call_validator('ehlo', reply, ehlo_as)
        self.extended_smtp = True
        if reply.code == '250':
            self.ehlo_as = ehlo_as
            self.envelope = None

    def HELO(self, reply, helo_as):
        self._call_validator('helo', reply, helo_as)
        if reply.code == '250':
            self.ehlo_as = helo_as
            self.envelope = None

    def TLSHANDSHAKE(self):
        self._call_validator('tls')
        self.security = 'TLS'

    def AUTH(self, reply, creds):
        self._call_validator('auth', reply, creds)
        if reply.code == '235':
            self.auth = creds.authcid

    def RSET(self, reply):
        self.envelope = None

    def MAIL(self, reply, address, params):
        self._call_validator('mail', reply, address, params)
        if reply.code == '250':
            self.envelope = Envelope(sender=address)

    def RCPT(self, reply, address, params):
        self._call_validator('rcpt', reply, address, params)
        if reply.code == '250':
            self.envelope.recipients.append(address)

    def DATA(self, reply):
        self._call_validator('data', reply)

    def HAVE_DATA(self, reply, data, err):
        if isinstance(err, MessageTooBig):
            reply.code = '552'
            reply.message = '5.3.4 Message exceeded size limit'
            return
        elif err:
            raise err

        self._call_validator('have_data', reply, data)
        if reply.code != '250':
            return

        self.envelope.client['ip'] = self.address[0]
        self.envelope.client['host'] = self.reverse_address
        self.envelope.client['name'] = self.ehlo_as
        self.envelope.client['protocol'] = self.protocol
        self.envelope.client['auth'] = self.auth

        self.envelope.parse(data)

        results = self.handoff(self.envelope)
        if isinstance(results[0][1], QueueError):
            default_reply = Reply('451', '4.3.0 Error queuing message')
            queue_reply = getattr(results[0][1], 'reply', default_reply)
            reply.copy(queue_reply)
        elif isinstance(results[0][1], RelayError):
            relay_reply = results[0][1].reply
#.........这里部分代码省略.........
开发者ID:oasiswork,项目名称:python-slimta,代码行数:101,代码来源:smtp.py


示例13: BANNER_

 def BANNER_(self, reply):
     self._ptr_lookup = PtrLookup(self.address[0])
     self._ptr_lookup.start()
     self._call_validator('banner', reply, self.address)
开发者ID:oasiswork,项目名称:python-slimta,代码行数:4,代码来源:smtp.py


示例14: test_run_dnssyntaxerror

 def test_run_dnssyntaxerror(self):
     self.mox.StubOutWithMock(dns_resolver, 'query')
     self.mox.ReplayAll()
     ptr = PtrLookup('abcd')
     self.assertIsInstance(ptr, gevent.Greenlet)
     self.assertIsNone(ptr._run())
开发者ID:sk1p,项目名称:python-slimta,代码行数:6,代码来源:test_slimta_util_ptrlookup.py


示例15: SmtpSession

class SmtpSession(object):
    def __init__(self, address, validator_class, handoff):
        self.extended_smtp = False
        self.security = None
        self.address = address
        self.reverse_address = None
        self.handoff = handoff
        self.validators = validator_class(self) if validator_class else None

        self.envelope = None
        self.ehlo_as = None
        self.auth = None

    def _call_validator(self, command, *args):
        method = "handle_" + command
        if hasattr(self.validators, method):
            getattr(self.validators, method)(*args)

    @property
    def protocol(self):
        proto = "SMTP"
        if self.extended_smtp:
            proto = "ESMTP"
        if self.security == "TLS":
            proto += "S"
        if self.auth:
            proto += "A"
        return proto

    def BANNER_(self, reply):
        self._ptr_lookup = PtrLookup(self.address[0])
        self._ptr_lookup.start()
        self._call_validator("banner", reply, self.address)

    def EHLO(self, reply, ehlo_as):
        self._call_validator("ehlo", reply, ehlo_as)
        self.extended_smtp = True
        if reply.code == "250":
            self.ehlo_as = ehlo_as
            self.envelope = None

    def HELO(self, reply, helo_as):
        self._call_validator("helo", reply, helo_as)
        if reply.code == "250":
            self.ehlo_as = helo_as
            self.envelope = None

    def TLSHANDSHAKE(self):
        self._call_validator("tls")
        self.security = "TLS"

    def AUTH(self, reply, creds):
        self._call_validator("auth", reply, creds)
        if reply.code == "235":
            self.auth = creds.authcid

    def RSET(self, reply):
        self.envelope = None

    def MAIL(self, reply, address, params):
        self._call_validator("mail", reply, address, params)
        if reply.code == "250":
            self.envelope = Envelope(sender=address)

    def RCPT(self, reply, address, params):
        self._call_validator("rcpt", reply, address, params)
        if reply.code == "250":
            self.envelope.recipients.append(address)

    def DATA(self, reply):
        self._call_validator("data", reply)

    def HAVE_DATA(self, reply, data, err):
        if isinstance(err, MessageTooBig):
            reply.code = "552"
            reply.message = "5.3.4 Message exceeded size limit"
            return
        elif err:
            raise err

        self._call_validator("have_data", reply, data)
        if reply.code != "250":
            return

        self.envelope.client["ip"] = self.address[0]
        self.envelope.client["host"] = self.reverse_address
        self.envelope.client["name"] = self.ehlo_as
        self.envelope.client["protocol"] = self.protocol
        self.envelope.client["auth"] = self.auth

        self.envelope.parse(data)

        results = self.handoff(self.envelope)
        if isinstance(results[0][1], QueueError):
            reply.code = "550"
            reply.message = "5.6.0 Error queuing message"
        elif isinstance(results[0][1], RelayError):
            relay_reply = results[0][1].reply
            reply.copy(relay_reply)
        else:
#.........这里部分代码省略.........
开发者ID:sk1p,项目名称:python-slimta,代码行数:101,代码来源:smtp.py


示例16: test_run_bad_ip

 def test_run_bad_ip(self):
     self.mox.ReplayAll()
     ptr = PtrLookup("abcd")
     self.assertIsInstance(ptr, gevent.Greenlet)
     self.assertIsNone(ptr._run())
开发者ID:thestick613,项目名称:python-slimta,代码行数:5,代码来源:test_slimta_util_ptrlookup.py



注:本文中的slimta.util.ptrlookup.PtrLookup类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python base.register_plugin函数代码示例发布时间:2022-05-27
下一篇:
Python dns.DNSResolver类代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap