Upgrade to mreg-api 0.5.0

mreg_cli/cli.py

@@ -289,9 +289,9 @@ def record_responses(self) -> None:
         """Record API responses for the last executed command."""
         output = OutputManager()
         client = get_client()
-        for response in client.get_client_history():
+        for response in client.requests:
             output.recording_request(response)
-        client.clear_client_history()
+        client.requests.clear()
 
     def process_command_line(self, line: str, *, interactive: bool = True) -> None:
         """Process a line containing a command."""

mreg_cli/commands/host_submodules/cname.py

@@ -7,7 +7,6 @@
 from mreg_cli.client import get_client
 from mreg_cli.commands.host import registry as command_registry
 from mreg_cli.exceptions import (
-    CreateError,
     EntityAlreadyExists,
     EntityNotFound,
     EntityOwnershipMismatch,
@@ -142,11 +141,8 @@ def cname_replace(args: argparse.Namespace) -> None:
     if not old_host:
         raise EntityNotFound(f"Could not find the host for the CNAME {cname}.")
 
-    updated_cname = client.cname.update(cname_obj, host=host)
-    if updated_cname:
-        OutputManager().add_ok(f"Moved CNAME alias {cname}: {old_host.name} -> {host.name}.")
-    else:
-        raise PatchError(f"Failed to move CNAME alias {cname}.")
+    client.cname.update(cname_obj, host=host)
+    OutputManager().add_ok(f"Moved CNAME alias {cname}: {old_host.name} -> {host.name}.")
 
 
 @command_registry.register_command(

mreg_cli/commands/host_submodules/core.py

@@ -34,7 +34,6 @@
 from mreg_cli.commands.host import registry as command_registry
 from mreg_cli.exceptions import (
     APIError,
-    CreateError,
     DeleteError,
     EntityAlreadyExists,
     EntityNotFound,
@@ -48,6 +47,7 @@
 from mreg_cli.output.history import output_host_history
 from mreg_cli.outputmanager import OutputManager
 from mreg_cli.types import Flag, QueryParams
+from mreg_cli.utilities.api import strict_limit
 from mreg_cli.utilities.resolution import resolve_host
 from mreg_cli.utilities.shared import convert_wildcard_to_regex
 
@@ -577,7 +577,9 @@ def _add_param(param: str, value: str) -> None:
         if value:
             _add_param(param, value)
 
-    hosts = client.host.list(limit=500, **params)
+    with strict_limit(client):
+        hosts = client.host.list(limit=500, **params)
+
     output_hostlist(hosts)
 
 

mreg_cli/commands/host_submodules/rr.py

@@ -49,7 +49,6 @@
 from mreg_cli.client import get_client
 from mreg_cli.commands.host import registry as command_registry
 from mreg_cli.exceptions import (
-    CreateError,
     DeleteError,
     EntityAlreadyExists,
     EntityNotFound,
@@ -934,13 +933,16 @@ def ttl_set(args: argparse.Namespace) -> None:
     if target_host is None and target_srv is None:
         raise EntityNotFound(f"No host or SRV record found for {name}")
 
+    # NOTE: do we really need to confirm that we set the TTL by refreshing?
     if target_host is not None:
-        result = client.host.update(target_host, ttl=ttl_value)
+        client.host.update(target_host, ttl=ttl_value)
+        result = client.host.refresh(target_host)
         new_ttl = result.ttl if result.ttl is not None else ttl
         OutputManager().add_ok(f"Set TTL for {target_host} to {new_ttl}.")
     else:
         assert target_srv is not None
-        result = client.srv.update(target_srv, ttl=ttl_value)
+        client.srv.update(target_srv, ttl=ttl_value)
+        result = client.srv.refresh(target_srv)
         new_ttl = result.ttl if result.ttl is not None else ttl
         OutputManager().add_ok(f"Set TTL for {target_srv} to {new_ttl}.")
 

mreg_cli/commands/network.py

@@ -40,6 +40,7 @@
 from mreg_cli.output.network import output_network_policy_attribute
 from mreg_cli.outputmanager import OutputManager
 from mreg_cli.types import Flag, QueryParams
+from mreg_cli.utilities.api import strict_limit
 from mreg_cli.utilities.resolution import resolve_host, resolve_network
 from mreg_cli.utilities.shared import convert_wildcard_to_regex, string_to_int
 from mreg_cli.utilities.validators import is_valid_category_tag, is_valid_location_tag
@@ -287,7 +288,8 @@ def find(args: argparse.Namespace) -> None:
         if not params:
             raise InputFailure("Need at least one search criteria")
 
-        networks = client.network.list(limit=500, **params)
+        with strict_limit(client):
+            networks = client.network.list(limit=500, **params)
 
     if not networks:
         raise EntityNotFound("No networks matching the query were found.")

mreg_cli/exceptions.py

@@ -135,6 +135,10 @@ class LoginFailedError(CliError):
     """Error class for login failure."""
 
 
+class TooManyResults(CliWarning):
+    """API returned too many results."""
+
+
 # FIXME: Inconsistent handling of HTTP errors in the original CLI implementation
 #        DELETE errors were considered errors, while other HTTP errors were
 #        considered warnings. They should all be considered errors.

mreg_cli/outputmanager.py

@@ -15,17 +15,19 @@
 from collections.abc import Iterable
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Generator, Literal, overload
+from typing import TYPE_CHECKING, Any, Generator, Literal, overload
 from urllib.parse import urlencode, urlparse
 
 import httpx
-from mreg_api.client import RequestRecord
 from pydantic import BaseModel
 
 from mreg_cli.errorbuilder import build_error_message
 from mreg_cli.exceptions import CliError, FileError
 from mreg_cli.types import Json, JsonMapping, RecordingEntry, TimeInfo
 
+if TYPE_CHECKING:
+    from mreg_api.requestlog import RequestRecord
+
 logger = logging.getLogger(__name__)
 
 

mreg_cli/utilities/api.py

@@ -8,18 +8,24 @@
 
 import functools
 import logging
-from typing import Callable, ParamSpec, TypeVar
+from collections.abc import Generator
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Callable, ParamSpec, TypeVar
 from urllib.parse import urljoin
 
 import httpx
 import mreg_api
+from mreg_api.events import Event, EventKind
 from prompt_toolkit import prompt
 
 from mreg_cli.client import get_client
 from mreg_cli.config import MregCliConfig
-from mreg_cli.exceptions import CliError, LoginFailedError
+from mreg_cli.exceptions import CliError, LoginFailedError, TooManyResults
 from mreg_cli.tokenfile import TokenFile
 
+if TYPE_CHECKING:
+    from mreg_api import MregClient
+
 logger = logging.getLogger(__name__)
 
 
@@ -38,6 +44,25 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
     return wrapper
 
 
+@contextmanager
+def strict_limit(client: MregClient) -> Generator[None, None, None]:
+    """Context manager that aborts requests for a query that returns too many results.
+
+    Adds a temporary mreg-api handler that listens for TRUNCATION events and
+    raises TooManyResults if such an event is received.
+    """
+
+    def fail_on_truncation(event: Event) -> None:
+        if event.kind == EventKind.TRUNCATION:
+            raise TooManyResults(f"{event.message} Refine your search.")
+
+    client.events.subscribe(fail_on_truncation)
+    try:
+        yield
+    finally:
+        client.events.unsubscribe(fail_on_truncation)
+
+
 def try_token_or_login(user: str, url: str, fail_without_token: bool = False) -> None:
     """Check for a valid token or interactively log in to MREG.
 

mreg_cli/utilities/resolution.py

@@ -61,7 +61,7 @@ def resolve_host(
 
     Resolution order: numeric id → IP → MAC → name → CNAME target.
     """
-    from mreg_cli.outputmanager import OutputManager
+    from mreg_cli.outputmanager import OutputManager  # noqa: PLC0415
 
     # We got passed an integer, assume host ID
     if isinstance(identifier, int):
@@ -145,8 +145,6 @@ def resolve_network(
 
     Replaces Network.get_by_any_means() / get_by_any_means_or_raise().
     """
-    from mreg_cli.exceptions import EntityNotFound as CliEntityNotFound
-
     network: Network | None = None
     try:
         # Try as IP first
@@ -161,7 +159,7 @@ def resolve_network(
         except Exception:
             pass
 
-    if network is None and identifier.isdigit():
+    if network is None and identifier.isdecimal():
         try:
             network = client.network.first(id=int(identifier), required=False)
             # network = client.network.get(int(identifier), required=False)

pyproject.toml

@@ -69,6 +69,7 @@ dev = [
     "setuptools-scm",
     "build",
     "pyinstaller",
+    "pytest-httpserver>=1.1.5",
 ]
 ci = [{ include-group = "dev" }, "tox-gh-actions"]
 
@@ -234,5 +235,5 @@ version_file = "mreg_cli/_version.py"
 
 # Uncomment to activate git installation for local development if PyPI version
 # is not up to date with most recent mreg-api git main commit.
-# [tool.uv.sources]
-# mreg-api = { git = "https://github.com/unioslo/mreg-api", branch = "main" }
+[tool.uv.sources]
+mreg-api = { git = "https://github.com/unioslo/mreg-api", branch = "main" }

tests/api/test_client.py

@@ -1,17 +1,26 @@
 from __future__ import annotations
 
+from typing import Any
 from unittest.mock import patch
 
+import pytest
 from mreg_api import CacheConfig, MregClient
+from mreg_api.endpoints import Endpoint
+from pytest_httpserver import HTTPServer
 
 from mreg_cli.config import MregCliConfig
+from mreg_cli.exceptions import TooManyResults
+from mreg_cli.utilities.api import strict_limit
 
 
 def test_client_cache_readonly_fs_dir() -> None:
     """Test that client caching handles read-only filesystem gracefully (with directory arg)."""
     with patch("os.makedirs") as mock_makedirs:
         mock_makedirs.side_effect = PermissionError("Read-only directory")
-        client = MregClient(url="https://mreg.example.com", cache=CacheConfig(enable=True, directory="/readonly/path"))
+        client = MregClient(
+            url="https://mreg.example.com",
+            cache=CacheConfig(enable=True, directory="/readonly/path"),
+        )
     assert not client.cache.is_enabled
     assert client.cache._cache is None  # pyright: ignore[reportPrivateUsage]
 
@@ -31,3 +40,70 @@ def test_client_cache_default_enabled() -> None:
     client = MregClient(url="https://mreg.example.com", cache=CacheConfig(enable=cliconf.cache))
     assert client.cache.is_enabled
     assert client.cache._cache is not None  # pyright: ignore[reportPrivateUsage
+
+
+@pytest.mark.parametrize("paginated_response", [True, False])
+def test_client_strict_limit(httpserver: HTTPServer, paginated_response: bool) -> None:
+    """Test the `strict_limit` context manager."""
+    client = MregClient(url=httpserver.url_for("/"), cache=False)
+
+    resp: list[dict[str, Any]] | dict[str, Any] = [
+        {
+            "id": 1,
+            "name": "host1",
+            "comment": "",
+            "ipaddresses": [],
+            "created_at": "2024-01-01T00:00:00Z",
+            "updated_at": "2024-01-01T00:00:00Z",
+        },
+        {
+            "id": 2,
+            "name": "host2",
+            "comment": "",
+            "ipaddresses": [],
+            "created_at": "2024-01-01T00:00:00Z",
+            "updated_at": "2024-01-01T00:00:00Z",
+        },
+        {
+            "id": 3,
+            "name": "host3",
+            "comment": "",
+            "ipaddresses": [],
+            "created_at": "2024-01-01T00:00:00Z",
+            "updated_at": "2024-01-01T00:00:00Z",
+        },
+        {
+            "id": 4,
+            "name": "host4",
+            "comment": "",
+            "ipaddresses": [],
+            "created_at": "2024-01-01T00:00:00Z",
+            "updated_at": "2024-01-01T00:00:00Z",
+        },
+        {
+            "id": 5,
+            "name": "host5",
+            "comment": "",
+            "ipaddresses": [],
+            "created_at": "2024-01-01T00:00:00Z",
+            "updated_at": "2024-01-01T00:00:00Z",
+        },
+    ]
+    if paginated_response:
+        resp = {
+            "results": resp,
+            "next": None,
+            "previous": None,
+            "count": 5,
+        }
+    httpserver.expect_request(Endpoint.Hosts).respond_with_json(resp)
+
+    # This fails
+    with strict_limit(client):
+        with pytest.raises(TooManyResults) as excinfo:
+            client.host.list(limit=4)
+        assert "Refine your search" in str(excinfo.value)
+
+    # This succeeds (listener is removed after the context manager exits)
+    hosts = client.host.list(limit=4)  # truncates without raising
+    assert len(hosts) == 4