Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/adguardhome/adguardhome.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import aiohttp
from yarl import URL

from .client import AdGuardHomeClients
from .exceptions import AdGuardHomeConnectionError, AdGuardHomeError
from .filtering import AdGuardHomeFiltering
from .parental import AdGuardHomeParental
Expand Down Expand Up @@ -74,6 +75,7 @@ def __init__( # noqa: PLR0913
if self.base_path[-1] != "/":
self.base_path += "/"

self.clients = AdGuardHomeClients(self)
self.filtering = AdGuardHomeFiltering(self)
self.parental = AdGuardHomeParental(self)
self.querylog = AdGuardHomeQueryLog(self)
Expand Down
120 changes: 120 additions & 0 deletions src/adguardhome/client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
"""Interacting with AdGuardHome clients."""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
from adguardhome.adguardhome import AdGuardHome


@dataclass
class WhoisInfo:
"""Not described in the OpenAPI docs."""

type: str # noqa: A003


@dataclass
class AutoClient:
"""Automatically discovered AdGuardHome client."""

ip: str # pylint: disable=C0103
name: str
source: str
whois_info: WhoisInfo | None


@dataclass
class Client: # pylint: disable=R0902
"""Administratively managed AdGuardHome client."""

name: str
ids: list[str]
use_global_settings: bool
filtering_enabled: bool
parental_enabled: bool
safebrowsing_enabled: bool
safesearch_enabled: bool
use_global_blocked_services: bool
blocked_services: list[str] | None
upstreams: list[str]
tags: list[str]


@dataclass
class AdGuardHomeClients:
"""A resource facade for the /clients API on AdGuardHome."""

adguard: AdGuardHome

async def get_auto_clients(self) -> list[AutoClient]:
"""List the AutoClients detected by the AdGuardHome instance.

Returns
-------
A List of `AutoClient` objects corresponding to the /clients['auto_clients']
API response.

"""

def _make_auto_client(raw: dict[str, Any]) -> AutoClient:
whois_info = (
WhoisInfo(type=raw["whois_info"]["type"])
if "whois_info" in raw and raw["whois_info"]
else None
)
return AutoClient(
ip=raw["ip"],
name=raw["name"],
source=raw["source"],
whois_info=whois_info,
)

raw_auto_clients = (await self.adguard.request("clients", method="GET"))[
"auto_clients"
]
return [_make_auto_client(a) for a in raw_auto_clients]

async def get_clients(self) -> list[Client]:
"""List the Clients configured on the AdGuardHome instance.

These clients are mutable and can be updated by this API.

Returns
-------
A List of `Client` objects corresponding to the /clients['clients']
API response.

"""

def _make_client(raw: dict[str, Any]) -> Client:
return Client(
name=raw["name"],
ids=raw["ids"],
use_global_settings=raw["use_global_settings"],
filtering_enabled=raw["filtering_enabled"],
parental_enabled=raw["parental_enabled"],
safebrowsing_enabled=raw["safebrowsing_enabled"],
safesearch_enabled=raw["safesearch_enabled"],
use_global_blocked_services=raw["use_global_blocked_services"],
blocked_services=raw["blocked_services"],
upstreams=raw["upstreams"],
tags=raw["tags"],
)

return [
_make_client(c)
for c in (await self.adguard.request("clients", method="GET"))["clients"]
]

async def get_supported_tags(self) -> list[str]:
"""List supported tags for Clients.

Returns
-------
The supported tags for clients.

"""
return (await self.adguard.request("clients", method="GET"))["supported_tags"]
131 changes: 131 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
"""Tests for `adguardhome.client`"""

import json

import aiohttp
import pytest
from aresponses import ResponsesMockServer

from adguardhome import AdGuardHome
from adguardhome.client import AutoClient, Client, WhoisInfo


@pytest.mark.parametrize(
"method", ["get_auto_clients", "get_clients", "get_supported_tags"]
)
async def test_empty_list(aresponses: ResponsesMockServer, method: str) -> None:
"""Test listing clients."""
aresponses.add(
"example.com:3000",
"/control/clients",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text=json.dumps({"auto_clients": [], "clients": [], "supported_tags": []}),
),
)

async with aiohttp.ClientSession() as session:
adguard = AdGuardHome("example.com", session=session)
call = getattr(adguard.clients, method)
assert await call() == []


async def test_get_auto_clients(aresponses: ResponsesMockServer) -> None:
"""Test listing auto clients."""
aresponses.add(
"example.com:3000",
"/control/clients",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text=json.dumps(
{
"auto_clients": [
{"ip": "192.0.2.1", "name": "otto", "source": "test"},
{
"ip": "192.0.2.3",
"name": "otto",
"source": "test",
"whois_info": {"type": "who knows?"},
},
],
"clients": [],
"supported_tags": [],
}
),
),
)

async with aiohttp.ClientSession() as session:
adguard = AdGuardHome("example.com", session=session)
output = await adguard.clients.get_auto_clients()
assert (
AutoClient(ip="192.0.2.1", name="otto", source="test", whois_info=None)
in output
)
assert (
AutoClient(
ip="192.0.2.3",
name="otto",
source="test",
whois_info=WhoisInfo(type="who knows?"),
)
in output
)


async def test_get_clients(aresponses: ResponsesMockServer) -> None:
"""Test getting configured clients from AdGuard Home"""
aresponses.add(
"example.com:3000",
"/control/clients",
"GET",
aresponses.Response(
status=200,
headers={"Content-Type": "application/json"},
text=json.dumps(
{
"auto_clients": [],
"clients": [
{
"blocked_services": None,
"filtering_enabled": False,
"ids": ["192.0.2.1", "192.0.2.2"],
"name": "test",
"parental_enabled": True,
"safebrowsing_enabled": True,
"safesearch_enabled": True,
"tags": ["some tag"],
"upstreams": ["some upstream"],
"use_global_blocked_services": False,
"use_global_settings": False,
},
],
"supported_tags": [],
}
),
),
)

async with aiohttp.ClientSession() as session:
adguard = AdGuardHome("example.com", session=session)
output = await adguard.clients.get_clients()
assert (
Client(
name="test",
ids=["192.0.2.1", "192.0.2.2"],
filtering_enabled=False,
parental_enabled=True,
safebrowsing_enabled=True,
safesearch_enabled=True,
use_global_blocked_services=False,
use_global_settings=False,
blocked_services=None,
tags=["some tag"],
upstreams=["some upstream"],
)
in output
)
Loading