Add online detection CLI command with configurable caching

src/aignostics/system/_cli.py

@@ -85,6 +85,22 @@ def info(
             console.print(yaml.dump(info, width=80, default_flow_style=False), end="")
 
 
+@cli.command()
+def online() -> None:
+    """Check if the system is online.
+
+    Exits with code 0 if online, 1 if not online.
+    Prints status information in green if online, red if not.
+    """
+    is_online = _service.is_online()
+    if is_online:
+        console.print("System is online", style="success")
+        sys.exit(0)
+    else:
+        console.print("System is not online", style="error")
+        sys.exit(1)
+
+
 if find_spec("nicegui"):
     from ..utils import gui_run  # noqa: TID252
 

src/aignostics/system/_service.py

@@ -6,6 +6,7 @@
 import re
 import ssl
 import sys
+import time
 import typing as t
 from http import HTTPStatus
 from pathlib import Path
@@ -72,6 +73,7 @@ class Service(BaseService):
     """System service."""
 
     _settings: Settings
+    _online_cache: tuple[bool, float] | None = None  # (is_online, timestamp)
 
     def __init__(self) -> None:
         """Initialize service."""
@@ -161,6 +163,35 @@ def is_token_valid(self, token: str) -> bool:
             return False
         return token == self._settings.token.get_secret_value()
 
+    def is_online(self) -> bool:
+        """Check if the system is online.
+
+        Uses cached result if available and not expired. Otherwise, performs a network health check
+        and caches the result for the configured TTL.
+
+        Returns:
+            bool: True if the system is online, False otherwise.
+        """
+        current_time = time.time()
+
+        # Check if cache is valid
+        if self._online_cache is not None:
+            cached_status, cached_time = self._online_cache
+            if current_time - cached_time < self._settings.online_cache_ttl_seconds:
+                logger.debug("Returning cached online status: %s", cached_status)
+                return cached_status
+
+        # Perform network health check
+        logger.debug("Performing network health check to determine online status")
+        health = self._determine_network_health()
+        is_online = health.status == Health.Code.UP
+
+        # Update cache
+        self._online_cache = (is_online, current_time)
+        logger.debug("Online status: %s (cached for %d seconds)", is_online, self._settings.online_cache_ttl_seconds)
+
+        return is_online
+
     @staticmethod
     def _get_public_ipv4(timeout: int = NETWORK_TIMEOUT) -> str | None:
         """Get the public IPv4 address of the system.

src/aignostics/system/_settings.py

@@ -29,3 +29,12 @@ class Settings(OpaqueSettings):
             default=None,
         ),
     ]
+
+    online_cache_ttl_seconds: Annotated[
+        int,
+        Field(
+            description="Time-to-live in seconds for online status cache",
+            default=60,
+            ge=0,
+        ),
+    ] = 60

tests/aignostics/system/cli_test.py

@@ -358,3 +358,19 @@ def test_cli_http_proxy(runner: CliRunner, silent_logging, tmp_path: Path) -> No
         result = runner.invoke(cli, ["system", "config", "get", "CURL_CA_BUNDLE"])
         assert result.exit_code == 0
         assert "None" in result.output
+
+
+def test_cli_online_when_system_is_online(runner: CliRunner) -> None:
+    """Test online command when system is online."""
+    with patch("aignostics.system._service.Service.is_online", return_value=True):
+        result = runner.invoke(cli, ["system", "online"])
+        assert result.exit_code == 0
+        assert "System is online" in result.output
+
+
+def test_cli_online_when_system_is_offline(runner: CliRunner) -> None:
+    """Test online command when system is offline."""
+    with patch("aignostics.system._service.Service.is_online", return_value=False):
+        result = runner.invoke(cli, ["system", "online"])
+        assert result.exit_code == 1
+        assert "System is not online" in result.output

tests/aignostics/system/service_test.py

@@ -1,11 +1,13 @@
 """Tests of the system service."""
 
 import os
+import time
 from unittest import mock
 
 import pytest
 
 from aignostics.system._service import Service
+from aignostics.utils import Health
 
 
 @pytest.mark.timeout(15)
@@ -339,3 +341,107 @@ def test_is_secret_key_real_world_examples() -> None:
 
     for key in non_secret_examples:
         assert not Service._is_secret_key(key), f"Expected '{key}' to NOT be identified as a secret key"
+
+
+def test_is_online_returns_true_when_network_is_healthy() -> None:
+    """Test that is_online returns True when network health check passes."""
+    service = Service()
+
+    with mock.patch.object(Service, "_determine_network_health") as mock_health:
+        mock_health.return_value = Health(status=Health.Code.UP)
+
+        assert service.is_online() is True
+        mock_health.assert_called_once()
+
+
+def test_is_online_returns_false_when_network_is_unhealthy() -> None:
+    """Test that is_online returns False when network health check fails."""
+    service = Service()
+
+    with mock.patch.object(Service, "_determine_network_health") as mock_health:
+        mock_health.return_value = Health(status=Health.Code.DOWN, reason="Network unreachable")
+
+        assert service.is_online() is False
+        mock_health.assert_called_once()
+
+
+def test_is_online_uses_cache_within_ttl() -> None:
+    """Test that is_online uses cached result within TTL."""
+    service = Service()
+
+    with mock.patch.object(Service, "_determine_network_health") as mock_health:
+        mock_health.return_value = Health(status=Health.Code.UP)
+
+        # First call should check network
+        result1 = service.is_online()
+        assert result1 is True
+        assert mock_health.call_count == 1
+
+        # Second call within TTL should use cache
+        result2 = service.is_online()
+        assert result2 is True
+        assert mock_health.call_count == 1  # Should not be called again
+
+
+def test_is_online_cache_expires_after_ttl() -> None:
+    """Test that is_online cache expires after TTL."""
+    # Create service with short TTL
+    with mock.patch.dict(os.environ, {"AIGNOSTICS_SYSTEM_ONLINE_CACHE_TTL_SECONDS": "1"}):
+        service = Service()
+
+        with mock.patch.object(Service, "_determine_network_health") as mock_health:
+            mock_health.return_value = Health(status=Health.Code.UP)
+
+            # First call
+            result1 = service.is_online()
+            assert result1 is True
+            assert mock_health.call_count == 1
+
+            # Wait for cache to expire
+            time.sleep(1.1)
+
+            # Second call after TTL should check network again
+            result2 = service.is_online()
+            assert result2 is True
+            assert mock_health.call_count == 2
+
+
+def test_is_online_cache_respects_ttl_setting() -> None:
+    """Test that is_online respects the configured TTL setting."""
+    # Set TTL to 2 seconds
+    with mock.patch.dict(os.environ, {"AIGNOSTICS_SYSTEM_ONLINE_CACHE_TTL_SECONDS": "2"}):
+        service = Service()
+
+        with mock.patch.object(Service, "_determine_network_health") as mock_health:
+            mock_health.return_value = Health(status=Health.Code.UP)
+
+            # First call
+            service.is_online()
+            assert mock_health.call_count == 1
+
+            # Call after 1 second - should still use cache
+            time.sleep(1.0)
+            service.is_online()
+            assert mock_health.call_count == 1
+
+            # Call after 2+ seconds - should refresh
+            time.sleep(1.1)
+            service.is_online()
+            assert mock_health.call_count == 2
+
+
+def test_is_online_cache_updates_on_status_change() -> None:
+    """Test that cache updates correctly when status changes."""
+    service = Service()
+
+    with mock.patch.object(Service, "_determine_network_health") as mock_health:
+        # First call - online
+        mock_health.return_value = Health(status=Health.Code.UP)
+        assert service.is_online() is True
+
+        # Wait for cache to expire
+        with mock.patch.dict(os.environ, {"AIGNOSTICS_SYSTEM_ONLINE_CACHE_TTL_SECONDS": "0"}):
+            service = Service()
+            # Second call - offline
+            mock_health.return_value = Health(status=Health.Code.DOWN, reason="Network down")
+            assert service.is_online() is False