137 lines
5.0 KiB
Python
137 lines
5.0 KiB
Python
"""Tests for the BIND9 DNS client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
|
|
from conduit_client import AsyncConduitClient, ConduitClient
|
|
from conduit_client.models.dns import (
|
|
AddZoneRequest,
|
|
BatchRecordUpdate,
|
|
BatchUpdateRecordsRequest,
|
|
CacheFlushRequest,
|
|
ServerReloadRequest,
|
|
)
|
|
|
|
|
|
@respx.mock
|
|
def test_server_status(api_key: str) -> None:
|
|
respx.get("https://api.example.com/dns/servers/default/status").mock(
|
|
return_value=Response(200, text="named is running")
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.dns.server_status("default")
|
|
assert result == "named is running"
|
|
|
|
|
|
@respx.mock
|
|
def test_server_reload_and_stats(api_key: str) -> None:
|
|
reload_route = respx.post("https://api.example.com/dns/servers/default/reload").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.get("https://api.example.com/dns/servers/default/stats").mock(
|
|
return_value=Response(200, json={"zones": 5})
|
|
)
|
|
respx.get("https://api.example.com/dns/servers/default/config").mock(
|
|
return_value=Response(200, json={"options": {}})
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
client.dns.reload_server("default", ServerReloadRequest(mode="reconfig"))
|
|
assert client.dns.server_stats("default") == {"zones": 5}
|
|
assert client.dns.server_config("default") == {"options": {}}
|
|
assert json.loads(reload_route.calls[0].request.content) == {"mode": "reconfig"}
|
|
|
|
|
|
@respx.mock
|
|
def test_cache_flush(api_key: str) -> None:
|
|
route = respx.post("https://api.example.com/dns/servers/default/cache/flush").mock(
|
|
return_value=Response(204)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
client.dns.flush_cache(
|
|
"default", CacheFlushRequest(mode="flushname", name="example.com")
|
|
)
|
|
assert json.loads(route.calls[0].request.content) == {
|
|
"mode": "flushname",
|
|
"name": "example.com",
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
def test_zone_crud(api_key: str) -> None:
|
|
create_route = respx.post("https://api.example.com/dns/zones").mock(
|
|
return_value=Response(200, json={"zone": "example.com"})
|
|
)
|
|
respx.get("https://api.example.com/dns/zones/example.com").mock(
|
|
return_value=Response(200, json={"zone": "example.com", "type": "master"})
|
|
)
|
|
patch_route = respx.patch("https://api.example.com/dns/zones/example.com").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.delete("https://api.example.com/dns/zones/example.com").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.post("https://api.example.com/dns/zones/example.com/freeze").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.post("https://api.example.com/dns/zones/example.com/thaw").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.post("https://api.example.com/dns/zones/example.com/sync").mock(
|
|
return_value=Response(204)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.dns.create_zone(
|
|
AddZoneRequest(zone="example.com", klass="IN", extra="value")
|
|
)
|
|
assert result["zone"] == "example.com"
|
|
assert client.dns.get_zone("example.com")["type"] == "master"
|
|
client.dns.patch_zone("example.com", {"type": "slave"})
|
|
client.dns.delete_zone("example.com", confirm="example.com")
|
|
client.dns.freeze_zone("example.com")
|
|
client.dns.thaw_zone("example.com")
|
|
client.dns.sync_zone("example.com")
|
|
body = json.loads(create_route.calls[0].request.content)
|
|
assert body["zone"] == "example.com"
|
|
assert body["klass"] == "IN"
|
|
assert body["extra"] == "value"
|
|
assert json.loads(patch_route.calls[0].request.content) == {"type": "slave"}
|
|
|
|
|
|
@respx.mock
|
|
def test_batch_update_records(api_key: str) -> None:
|
|
route = respx.post("https://api.example.com/dns/zones/example.com/records:batchUpdate").mock(
|
|
return_value=Response(204)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
client.dns.batch_update_records(
|
|
"example.com",
|
|
BatchUpdateRecordsRequest(
|
|
updates=[
|
|
BatchRecordUpdate(
|
|
op="add", name="www", ttl=300, rtype="A", rdata="1.2.3.4"
|
|
)
|
|
]
|
|
),
|
|
)
|
|
assert json.loads(route.calls[0].request.content) == {
|
|
"updates": [
|
|
{"op": "add", "name": "www", "ttl": 300, "rtype": "A", "rdata": "1.2.3.4"}
|
|
]
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_async_server_status(api_key: str) -> None:
|
|
respx.get("https://api.example.com/dns/servers/default/status").mock(
|
|
return_value=Response(200, text="named is running")
|
|
)
|
|
async with AsyncConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = await client.dns.server_status("default")
|
|
assert result == "named is running"
|