for k, v in env.items():
monkeypatch.setenv(k, v)
params = conninfo_to_dict(conninfo)
- params = await psycopg._dns.resolve_hostaddr_async(params)
+ params = await psycopg._dns.resolve_hostaddr_async( # type: ignore[attr-defined]
+ params
+ )
assert conninfo_to_dict(want) == params
@pytest.mark.asyncio
async def test_resolve_hostaddr_async(conninfo, want, env, fake_resolve):
params = conninfo_to_dict(conninfo)
- params = await psycopg._dns.resolve_hostaddr_async(params)
+ params = await psycopg._dns.resolve_hostaddr_async( # type: ignore[attr-defined]
+ params
+ )
assert conninfo_to_dict(want) == params
monkeypatch.setenv(k, v)
params = conninfo_to_dict(conninfo)
with pytest.raises((TypeError, psycopg.Error)):
- await psycopg._dns.resolve_hostaddr_async(params)
+ await psycopg._dns.resolve_hostaddr_async( # type: ignore[attr-defined]
+ params
+ )
@pytest.mark.asyncio
@classmethod
async def _get_connection_params(cls, conninfo, **kwargs):
params = await super()._get_connection_params(conninfo, **kwargs)
- params = await psycopg._dns.resolve_hostaddr_async(params)
+ params = await (
+ psycopg._dns.resolve_hostaddr_async( # type: ignore[attr-defined]
+ params
+ )
+ )
return params
with pytest.raises(ZeroDivisionError):
else:
return [dns.rdtypes.IN.A.A("IN", "A", addr)]
- monkeypatch.setattr(psycopg._dns.async_resolver, "resolve", fake_resolve_)
+ monkeypatch.setattr(
+ psycopg._dns.async_resolver, # type: ignore[attr-defined]
+ "resolve",
+ fake_resolve_,
+ )
@pytest.fixture
async def fail_resolve_(qname):
pytest.fail(f"shouldn't try to resolve {qname}")
- monkeypatch.setattr(psycopg._dns.async_resolver, "resolve", fail_resolve_)
+ monkeypatch.setattr(
+ psycopg._dns.async_resolver, # type: ignore[attr-defined]
+ "resolve",
+ fail_resolve_,
+ )
def import_dnspython():
for retry in retries:
with retry:
params = conninfo_to_dict(conninfo)
- params = psycopg._dns.resolve_srv(params)
+ params = psycopg._dns.resolve_srv(params) # type: ignore[attr-defined]
assert conninfo_to_dict(want) == params
async for retry in retries:
with retry:
params = conninfo_to_dict(conninfo)
- params = await psycopg._dns.resolve_srv_async(params)
+ params = await (
+ psycopg._dns.resolve_srv_async( # type: ignore[attr-defined]
+ params
+ )
+ )
assert conninfo_to_dict(want) == params
monkeypatch.setenv(k, v)
params = conninfo_to_dict(conninfo)
with pytest.raises(psycopg.OperationalError):
- psycopg._dns.resolve_srv(params)
+ psycopg._dns.resolve_srv(params) # type: ignore[attr-defined]
@pytest.mark.asyncio
monkeypatch.setenv(k, v)
params = conninfo_to_dict(conninfo)
with pytest.raises(psycopg.OperationalError):
- await psycopg._dns.resolve_srv_async(params)
+ await psycopg._dns.resolve_srv_async(params) # type: ignore[attr-defined]
@pytest.fixture
def fake_srv(monkeypatch):
f = get_fake_srv_function(monkeypatch)
- monkeypatch.setattr(psycopg._dns.resolver, "resolve", f)
+ monkeypatch.setattr(
+ psycopg._dns.resolver, # type: ignore[attr-defined]
+ "resolve",
+ f,
+ )
@pytest.fixture
async def af(qname, rdtype):
return f(qname, rdtype)
- monkeypatch.setattr(psycopg._dns.async_resolver, "resolve", af)
+ monkeypatch.setattr(
+ psycopg._dns.async_resolver, # type: ignore[attr-defined]
+ "resolve",
+ af,
+ )
def get_fake_srv_function(monkeypatch):