saghul / pycares

Python interface for c-ares
https://pypi.org/project/pycares/
MIT License
165 stars 73 forks source link

let "pycares" support ARES_OPT_LOOKUPS option. #100

Closed abaelhe closed 5 years ago

abaelhe commented 5 years ago

Reproduce: ############# import pycares from tornado.tcpclient import Resolver
Resolver.configure('tornado.platform.caresresolver.CaresResolver')
resolver = Resolver() resolver.channel = pycares.Channel(sock_state_cb=resolver._sock_state_cb, timeout=1, tries=1, socket_receive_buffer_size=4096, servers=['127.0.0.1'], tcp_port=5353, udp_port=5353, rotate=False)

################## Error: TypeError: initializer for ctype 'char *' must be a cdata pointer, not str

saghul commented 5 years ago

Thank you! Any chance you can add a test?

saghul commented 5 years ago

Btw, ignore the failing CI, one of the DNS entries which we are testing is failing, I'll fix that soon.

abaelhe commented 5 years ago

@saghul you turn.

saghul commented 5 years ago

Please don't make the test suite depend on Tornado. Just use pycares directly, like the other tests.

abaelhe commented 5 years ago

Please don't make the test suite depend on Tornado. Just use pycares directly, like the other tests.

@saghul latest version using the "select" module to monitor async event which already imported.

(qianxin_venv) Abael:pycares abaelhe$ PYTHONPATH=$(pwd) python tests/tests.py DNSTest.test_lookup test_lookup (main.DNSTest) ... ok


Ran 1 test in 0.035s

OK

Implements:

#####################################################

def test_lookup(self):
    channel = pycares.Channel(
        lookups="b",
        timeout=1,
        tries=1,
        socket_receive_buffer_size=4096,
        servers=["8.8.8.8", "8.8.4.4"],
        tcp_port=53,
        udp_port=53,
        rotate=True,
    )   

    def on_result(result, errorno):
        self.result, self.errorno = result, errorno

    for domain in [
        "google.com",
        "microsoft.com",
        "apple.com",
        "amazon.com",
        "baidu.com",
        "alipay.com",
        "tencent.com",
    ]:  
        self.result, self.errorno = None, None
        self.channel.query(domain, pycares.QUERY_TYPE_A, on_result)
        self.wait()
        self.assertNoError(self.errorno)
        self.assertTrue(len(self.result) > 0)
        for r in self.result:
            self.assertEqual(type(r), pycares.ares_query_a_result)
            self.assertNotEqual(r.host, None)
            self.assertTrue(r.type == 'A')
            self.assertTrue(r.ttl >= 0)
saghul commented 5 years ago

Neat, thanks! I’ll merge shortly! (Not on a computer at the moment).