diff --git a/pyproject.toml b/pyproject.toml index ea5f9731..ce8ff98d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,10 +82,6 @@ source = ["tailscale"] [tool.coverage.report] show_missing = true fail_under = 95 -omit = [ - "src/tailscale/_cli.py", - "src/tailscale/cli/*", -] [tool.ty.environment] python-version = "3.11" diff --git a/src/tailscale/cli/async_typer.py b/src/tailscale/cli/async_typer.py index 11022ffe..a82cdf36 100644 --- a/src/tailscale/cli/async_typer.py +++ b/src/tailscale/cli/async_typer.py @@ -28,7 +28,6 @@ from typing import ( TYPE_CHECKING, Any, - ParamSpec, TypeVar, ) @@ -36,12 +35,10 @@ from typer import Typer as SyncTyper if TYPE_CHECKING: - from collections.abc import Coroutine - from typer.core import TyperCommand, TyperGroup -_P = ParamSpec("_P") -_R = TypeVar("_R") +# A callback or command, which can be a coroutine function or a plain one. +_F = TypeVar("_F", bound=Callable[..., Any]) HandleErrorFunc = Callable[[Any], None] @@ -71,10 +68,7 @@ def callback( # type: ignore[override] # ty: ignore[invalid-method-override] hidden: bool = False, deprecated: bool = False, rich_help_panel: str | None = None, - ) -> Callable[ - [Callable[_P, Coroutine[Any, Any, _R]]], - Callable[_P, Coroutine[Any, Any, _R]], - ]: + ) -> Callable[[_F], _F]: """Create a new typer callback.""" super_callback = super().callback( cls=cls, @@ -95,12 +89,12 @@ def callback( # type: ignore[override] # ty: ignore[invalid-method-override] ) def decorator( - func: Callable[_P, Coroutine[Any, Any, _R]], - ) -> Callable[_P, Coroutine[Any, Any, _R]]: + func: _F, + ) -> _F: if inspect.iscoroutinefunction(func): @wraps(func) - def sync_func(*_args: _P.args, **_kwargs: _P.kwargs) -> _R: + def sync_func(*_args: Any, **_kwargs: Any) -> Any: return asyncio.run(func(*_args, **_kwargs)) super_callback(sync_func) @@ -129,10 +123,7 @@ def command( # type: ignore[override] # ty: ignore[invalid-method-override] deprecated: bool = False, # Rich settings rich_help_panel: str | None = None, - ) -> Callable[ - [Callable[_P, Coroutine[Any, Any, _R]]], - Callable[_P, Coroutine[Any, Any, _R]], - ]: + ) -> Callable[[_F], _F]: """Create a new typer command.""" super_command = super().command( name, @@ -150,12 +141,12 @@ def command( # type: ignore[override] # ty: ignore[invalid-method-override] ) def decorator( - func: Callable[_P, Coroutine[Any, Any, _R]], - ) -> Callable[_P, Coroutine[Any, Any, _R]]: + func: _F, + ) -> _F: if inspect.iscoroutinefunction(func): @wraps(func) - def sync_func(*_args: _P.args, **_kwargs: _P.kwargs) -> _R: + def sync_func(*_args: Any, **_kwargs: Any) -> Any: return asyncio.run(func(*_args, **_kwargs)) super_command(sync_func) diff --git a/src/tailscale/tailscale.py b/src/tailscale/tailscale.py index 5dcf491a..a3cf396e 100644 --- a/src/tailscale/tailscale.py +++ b/src/tailscale/tailscale.py @@ -75,7 +75,7 @@ class Tailscale: oauth_client_id: str | None = None oauth_client_secret: str | None = field(default=None, repr=False) - request_timeout: int = 8 + request_timeout: float = 8 session: ClientSession | None = None token_storage: TokenStorage | None = field(default=None, repr=False) diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index 84c41468..005c58bc 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -10,11 +10,14 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest +from typer import Exit from typer.core import TyperGroup from typer.main import get_command from typer.testing import CliRunner +from tailscale._cli import main from tailscale.cli import cli +from tailscale.cli.async_typer import AsyncTyper from tailscale.exceptions import ( TailscaleAuthenticationError, TailscaleConnectionError, @@ -1165,6 +1168,105 @@ def test_missing_auth( # --- error handlers --- +@pytest.mark.parametrize( + ("error", "title"), + [ + (TailscalePermissionError(403, "not on this plan"), "Permission denied"), + (TailscaleAuthenticationError("bad key"), "Authentication error"), + (TailscaleConnectionError("unreachable"), "Connection error"), + (TailscaleError("something broke"), "Tailscale API error"), + ], +) +def test_errors_through_the_entry_point( + capsys: pytest.CaptureFixture[str], + error: TailscaleError, + title: str, +) -> None: + """Errors of a command end up at their handler, like when run for real.""" + mock_client = _mock_tailscale() + mock_client._request.side_effect = error # pylint: disable=protected-access + with ( + patch("tailscale.cli._build_client", return_value=mock_client), + pytest.raises(SystemExit) as exc_info, + ): + cli(["dump", "settings", "--api-key", "tskey-api-test"]) + + assert exc_info.value.code == 1 + assert title in capsys.readouterr().out + + +def test_unhandled_errors_through_the_entry_point() -> None: + """Errors without a handler are raised as they are.""" + mock_client = _mock_tailscale() + mock_client._request.side_effect = RuntimeError("boom") # pylint: disable=protected-access + with ( + patch("tailscale.cli._build_client", return_value=mock_client), + pytest.raises(RuntimeError, match="boom"), + ): + cli(["dump", "settings", "--api-key", "tskey-api-test"]) + + +def test_console_script_runs_the_cli() -> None: + """The console script runs the CLI.""" + with patch("tailscale.cli.cli") as mock_cli: + main() + mock_cli.assert_called_once_with() + + +def test_async_callback(runner: CliRunner) -> None: + """An async callback of an AsyncTyper app runs, like an async command.""" + app = AsyncTyper() + called: list[str] = [] + + @app.callback(invoke_without_command=True) + async def callback() -> None: + called.append("callback") + + @app.command() + async def hello() -> None: + called.append("hello") + + result = runner.invoke(app, ["hello"]) + assert result.exit_code == 0, result.output + assert called == ["callback", "hello"] + + +def test_sync_callback_and_command(runner: CliRunner) -> None: + """Plain functions keep working as callback and command.""" + app = AsyncTyper() + called: list[str] = [] + + @app.callback() + def callback() -> None: + called.append("callback") + + @app.command() + def hello() -> None: + called.append("hello") + + result = runner.invoke(app, ["hello"]) + assert result.exit_code == 0, result.output + assert called == ["callback", "hello"] + + +def test_exit_through_the_entry_point() -> None: + """A typer.Exit is not turned into an error.""" + app = AsyncTyper() + app.error_handler(Exception)(lambda _: pytest.fail("handler called")) + + @app.command() + def stop() -> None: + raise Exit(code=3) + + @app.command() + def other() -> None: + """Make this a group, so commands are called by name.""" + + with pytest.raises(SystemExit) as exc_info: + app(["stop"]) + assert exc_info.value.code == 3 + + def test_authentication_error_handler( capsys: pytest.CaptureFixture[str], snapshot: SnapshotAssertion, diff --git a/tests/test_real_fixtures.py b/tests/test_real_fixtures.py index 3c770582..becff23a 100644 --- a/tests/test_real_fixtures.py +++ b/tests/test_real_fixtures.py @@ -22,75 +22,195 @@ END = datetime(2026, 10, 10, tzinfo=UTC) +def _check_devices(devices: Any) -> None: + assert len(devices) == 18 + # Devices are keyed by their legacy ID, not their node ID. + (device,) = (device for device in devices.values() if device.node_id == DEVICE_ID) + assert devices[device.device_id] is device + assert device.os == "macOS" + assert device.last_seen == datetime(2019, 6, 19, 17, 7, 49, tzinfo=UTC) + + +def _check_device(device: Any) -> None: + assert device.node_id == DEVICE_ID + assert device.client_version == "1.20.4-t8c6cb1cea-g2accf24a5" + + +def _check_routes(routes: Any) -> None: + assert routes.advertised_routes == [] + assert routes.enabled_routes == [] + + +def _check_posture_attributes(posture: Any) -> None: + assert posture.attributes["node:osVersion"] == "11.5.2" + assert posture.attributes["node:tsAutoUpdate"] is False + assert posture.expiries == {} + + +def _check_users(users: Any) -> None: + (user,) = users + assert (user.role, user.status, user.device_count) == ("owner", "active", 18) + + +def _check_keys(keys: Any) -> None: + (key,) = keys + assert key.key_type == "api" + assert key.scopes == ["all", "all:read"] + assert key.expiry_seconds == 7776000 + + +def _check_settings(settings: Any) -> None: + assert settings.devices_key_duration_days == 180 + assert settings.https_enabled is True + assert settings.route_selection == "active-passive-failover" + + +def _check_dns_configuration(configuration: Any) -> None: + assert [resolver.address for resolver in configuration.nameservers] == [ + "203.0.113.50" + ] + assert configuration.preferences.magic_dns is True + assert configuration.split_dns == {} + + +def _check_contacts(contacts: Any) -> None: + # The API sends an empty email for a contact that is not set. + assert contacts.account is not None + assert contacts.account.email is None + assert contacts.account.needs_verification is False + + +def _check_organization_tailnets(page: Any) -> None: + assert page.total_count == 1 + assert page.cursor is None + assert [tailnet.display_name for tailnet in page.tailnets] == ["example1.com"] + + +def _check_audit_logs(logs: Any) -> None: + assert len(logs) == 29 + # Nanoseconds from the API end up as the microseconds Python has. + assert logs[0].event_time == datetime(2024, 1, 13, 18, 39, 35, 416440, tzinfo=UTC) + assert logs[0].action == "LOGIN" + assert sum(log.event_group_id is None for log in logs) == 7 + + +def _check_empty(result: Any) -> None: + assert result == [] + + @pytest.mark.parametrize( - ("fixture", "path", "call"), + ("fixture", "path", "call", "check"), [ - ("devices.json", "tailnet/frenck/devices?fields=all", lambda ts: ts.devices()), + ( + "devices.json", + "tailnet/frenck/devices?fields=all", + lambda ts: ts.devices(), + _check_devices, + ), ( "device.json", f"device/{DEVICE_ID}?fields=all", lambda ts: ts.device(DEVICE_ID), + _check_device, ), ( "routes.json", f"device/{DEVICE_ID}/routes", lambda ts: ts.device_routes(DEVICE_ID), + _check_routes, ), ( "device-posture-attributes.json", f"device/{DEVICE_ID}/attributes", lambda ts: ts.device_posture_attributes(DEVICE_ID), + _check_posture_attributes, ), - ("users.json", "tailnet/frenck/users", lambda ts: ts.users()), + ("users.json", "tailnet/frenck/users", lambda ts: ts.users(), _check_users), ( "user-invites.json", "tailnet/frenck/user-invites", lambda ts: ts.user_invites(), + _check_empty, + ), + ( + "keys.json", + "tailnet/frenck/keys?all=true", + lambda ts: ts.keys(), + _check_keys, + ), + ( + "settings.json", + "tailnet/frenck/settings", + lambda ts: ts.tailnet_settings(), + _check_settings, ), - ("keys.json", "tailnet/frenck/keys?all=true", lambda ts: ts.keys()), - ("settings.json", "tailnet/frenck/settings", lambda ts: ts.tailnet_settings()), ( "dns-configuration.json", "tailnet/frenck/dns/configuration", lambda ts: ts.dns_configuration(), + _check_dns_configuration, + ), + ( + "webhooks.json", + "tailnet/frenck/webhooks", + lambda ts: ts.webhooks(), + _check_empty, + ), + ( + "services.json", + "tailnet/frenck/services", + lambda ts: ts.services(), + _check_empty, ), - ("webhooks.json", "tailnet/frenck/webhooks", lambda ts: ts.webhooks()), - ("services.json", "tailnet/frenck/services", lambda ts: ts.services()), ( "posture-integrations.json", "tailnet/frenck/posture/integrations", lambda ts: ts.posture_integrations(), + _check_empty, + ), + ( + "oauth-apps.json", + "tailnet/frenck/oauth-apps", + lambda ts: ts.oauth_apps(), + _check_empty, + ), + ( + "contacts.json", + "tailnet/frenck/contacts", + lambda ts: ts.contacts(), + _check_contacts, ), - ("oauth-apps.json", "tailnet/frenck/oauth-apps", lambda ts: ts.oauth_apps()), - ("contacts.json", "tailnet/frenck/contacts", lambda ts: ts.contacts()), ( "organization-tailnets.json", "organizations/-/tailnets", lambda ts: ts.organization_tailnets(), + _check_organization_tailnets, ), ( "audit-logs.json", "tailnet/frenck/logging/configuration" "?start=2026-10-08T00:00:00%2B00:00&end=2026-10-10T00:00:00%2B00:00", lambda ts: ts.configuration_audit_logs(start=START, end=END), + _check_audit_logs, ), ], ) -async def test_real_fixture( +async def test_real_fixture( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-many-positional-arguments responses: aioresponses, tailscale_client: Tailscale, fixture: str, path: str, call: Callable[[Tailscale], Awaitable[Any]], + check: Callable[[Any], None], ) -> None: - """Test a real, anonymized response of the API parses.""" + """Test a real, anonymized response of the API parses into what it holds.""" responses.get( f"{URL}/{path}", status=200, body=load_fixture(f"real/{fixture}"), content_type="application/json", ) - assert await call(tailscale_client) is not None + check(await call(tailscale_client)) async def test_real_devices( diff --git a/tests/test_tailscale.py b/tests/test_tailscale.py index 2dfa3755..e91cb1c3 100644 --- a/tests/test_tailscale.py +++ b/tests/test_tailscale.py @@ -134,6 +134,26 @@ async def test_timeout( await tailscale_client._request("test") +async def test_timeout_stops_a_hanging_request( + responses: aioresponses, + tailscale_client: Tailscale, +) -> None: + """Test a request that does not finish in time is stopped.""" + + async def hang(*_args: Any, **_kwargs: Any) -> CallbackResult: + await asyncio.sleep(10) + return CallbackResult(status=200, body="{}") + + responses.get(f"{URL}/test", callback=hang) + tailscale_client.request_timeout = 0.05 + + loop = asyncio.get_running_loop() + started = loop.time() + with pytest.raises(TailscaleConnectionError, match="Timeout"): + await tailscale_client._request("test") + assert loop.time() - started < 1 + + async def test_http_error404( responses: aioresponses, tailscale_client: Tailscale, @@ -606,20 +626,17 @@ async def test_set_device_routes( responses: aioresponses, tailscale_client: Tailscale, ) -> None: - """Test setting device routes.""" + """Test setting device routes replaces the enabled routes.""" responses.post( f"{URL}/device/12345/routes", status=200, - body=load_fixture("device_routes.json"), + body='{"advertisedRoutes": ["10.200.0.0/16", "192.168.50.0/24"],' + '"enabledRoutes": ["10.200.0.0/16"]}', content_type="application/json", ) routes = await tailscale_client.set_device_routes("12345", routes=["10.200.0.0/16"]) - assert routes.advertised_routes == [ - "10.200.0.0/16", - "192.168.50.0/24", - "172.16.0.0/12", - ] - assert routes.enabled_routes == ["10.200.0.0/16", "192.168.50.0/24"] + assert routes.advertised_routes == ["10.200.0.0/16", "192.168.50.0/24"] + assert routes.enabled_routes == ["10.200.0.0/16"] async def test_set_device_ipv4_address( @@ -636,6 +653,132 @@ async def test_set_device_ipv4_address( await tailscale_client.set_device_ipv4_address("12345", ipv4_address="100.64.0.1") +@pytest.mark.parametrize( + ("method", "args", "kwargs", "http_method", "path", "payload", "response"), + [ + ("delete_device", ["n1"], {}, "DELETE", "device/n1", None, ""), + ( + "authorize_device", + ["n1"], + {"authorized": True}, + "POST", + "device/n1/authorized", + {"authorized": True}, + "", + ), + ( + "authorize_device", + ["n1"], + {"authorized": False}, + "POST", + "device/n1/authorized", + {"authorized": False}, + "", + ), + ("expire_device_key", ["n1"], {}, "POST", "device/n1/expire", None, ""), + ( + "set_device_key_expiry", + ["n1"], + {"key_expiry_disabled": False}, + "POST", + "device/n1/key", + {"keyExpiryDisabled": False}, + "", + ), + ( + "rename_device", + ["n1"], + {"name": "server"}, + "POST", + "device/n1/name", + {"name": "server"}, + "", + ), + ( + "set_device_tags", + ["n1"], + {"tags": []}, + "POST", + "device/n1/tags", + {"tags": []}, + "", + ), + ( + "set_device_routes", + ["n1"], + {"routes": []}, + "POST", + "device/n1/routes", + {"routes": []}, + '{"advertisedRoutes": [], "enabledRoutes": []}', + ), + ( + "set_device_ipv4_address", + ["n1"], + {"ipv4_address": "100.64.0.1"}, + "POST", + "device/n1/ip", + {"ipv4": "100.64.0.1"}, + "", + ), + ( + "set_dns_nameservers", + [], + {"dns": []}, + "POST", + "tailnet/frenck/dns/nameservers", + {"dns": []}, + '{"dns": []}', + ), + ( + "set_dns_preferences", + [], + {"magic_dns": False}, + "POST", + "tailnet/frenck/dns/preferences", + {"magicDNS": False}, + '{"magicDNS": false}', + ), + ( + "set_dns_search_paths", + [], + {"search_paths": ["corp.example.com"]}, + "POST", + "tailnet/frenck/dns/searchpaths", + {"searchPaths": ["corp.example.com"]}, + '{"searchPaths": ["corp.example.com"]}', + ), + ], +) +async def test_request_payloads( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-many-positional-arguments + responses: aioresponses, + tailscale_client: Tailscale, + method: str, + args: list[Any], + kwargs: dict[str, Any], + http_method: str, + path: str, + payload: Any, + response: str, +) -> None: + """Test changes send exactly one request, with exactly the given payload.""" + responses.add( + f"{URL}/{path}", + method=http_method, + status=200, + body=response, + content_type="application/json", + ) + await getattr(tailscale_client, method)(*args, **kwargs) + + assert responses.requests + ((request_method, request_url), requests), *other = responses.requests.items() + assert not other + assert (request_method, str(request_url)) == (http_method, f"{URL}/{path}") + (request,) = requests + assert request.kwargs["json"] == payload + + # --- DNS tests ---