observer add test for pub uptime alert

AGENTS.md

@@ -0,0 +1,31 @@
+# Repository Guidelines
+
+## Project Structure & Modules
+- Core code lives in `pyth_observer/`: CLI entrypoint in `cli.py`, check logic under `check/`, alert dispatchers in `dispatch.py`, event types in `event.py`, and HTTP probes in `health_server.py`.
+- Supporting assets and defaults: sample configs (`sample.config.yaml`, `sample.publishers.yaml`, `sample.coingecko.yaml`), Dockerfile for container builds, and helper scripts in `scripts/` (e.g., `build_coingecko_mapping.py`).
+- Tests are in `tests/` and mirror module names (`test_checks_price_feed.py`, `test_checks_publisher.py`).
+
+## Setup, Build & Run
+- Use Python 3.11 with Poetry 2.x. Suggested bootstrap: `poetry env use $(which python)` then `poetry install`.
+- Common Make targets: `make setup` (install deps), `make run` (devnet run), `make test`, `make cover`, `make lint`, `make clean`.
+- Direct commands: `poetry run pyth-observer --config sample.config.yaml --publishers sample.publishers.yaml --coingecko-mapping sample.coingecko.yaml` to run locally; add `-l debug` for verbose logs.
+- CoinGecko mapping: `poetry run python scripts/build_coingecko_mapping.py -o my_mapping.json` and compare with `-e sample.coingecko.yaml` before replacing defaults.
+
+## Testing Guidelines
+- Framework: `pytest`. Quick check with `poetry run pytest`; coverage report via `make cover` (writes `htmlcov/`).
+- Keep tests colocated under `tests/` with `test_*` naming. Prefer async tests for async code paths and mock network calls.
+- Add regression tests alongside new checks or dispatch paths; include sample config fragments when useful.
+
+## Coding Style & Naming
+- Auto-format with `black` and import order via `isort` (run together with `make lint`). Lint also runs `pyright` and `pyflakes`.
+- Target Python 3.11; favor type hints on public functions and dataclasses/models. Use snake_case for functions/variables, PascalCase for classes, and uppercase for constants.
+- Keep config keys consistent with existing YAML samples; avoid hard-coding secrets—read from env vars.
+
+## Commit & PR Practices
+- Follow the existing Conventional Commit style (`fix:`, `chore:`, `refactor!:`, etc.) seen in `git log`.
+- PRs should summarize behavior changes, link issues, and include reproduction or validation steps (commands run, configs used). Add screenshots only when output formatting changes.
+- Keep diffs small and focused; update sample config or docs when user-facing options change.
+
+## Configuration & Security Notes
+- Sensitive values (API keys, tokens) must be supplied via environment variables; never commit them. Use `.env` locally and document new keys in `README.md`.
+- For deployments, wire liveness/readiness probes to `GET /live` and `GET /ready` on port 8080.

tests/test_checks_publisher.py

@@ -1,6 +1,8 @@
 import random
 import time
+from datetime import datetime
 from unittest.mock import patch
+from zoneinfo import ZoneInfo
 
 import pytest
 from pythclient.market_schedule import MarketSchedule
@@ -10,6 +12,7 @@
 from pyth_observer.check.publisher import (
     PUBLISHER_CACHE,
     PriceUpdate,
+    PublisherOfflineCheck,
     PublisherPriceCheck,
     PublisherStalledCheck,
     PublisherState,
@@ -195,3 +198,220 @@ def test_redemption_rate_passes_check(self):
 
         # Should pass even after long period without changes
         self.run_check(check, 3600, True)  # 1 hour
+
+
+class TestPublisherOfflineCheck:
+    """Test suite for PublisherOfflineCheck covering various scenarios."""
+
+    def make_state(
+        self,
+        publisher_slot: int,
+        latest_block_slot: int,
+        schedule: MarketSchedule | None = None,
+        publisher_name: str = "test_publisher",
+        symbol: str = "Crypto.BTC/USD",
+    ) -> PublisherState:
+        """Helper to create PublisherState for testing."""
+        if schedule is None:
+            schedule = MarketSchedule("America/New_York;O,O,O,O,O,O,O;")
+        return PublisherState(
+            publisher_name=publisher_name,
+            symbol=symbol,
+            asset_type="Crypto",
+            schedule=schedule,
+            public_key=SolanaPublicKey("2hgu6Umyokvo8FfSDdMa9nDKhcdv9Q4VvGNhRCeSWeD3"),
+            status=PythPriceStatus.TRADING,
+            aggregate_status=PythPriceStatus.TRADING,
+            slot=publisher_slot,
+            aggregate_slot=latest_block_slot - 5,
+            latest_block_slot=latest_block_slot,
+            price=100.0,
+            price_aggregate=100.0,
+            confidence_interval=1.0,
+            confidence_interval_aggregate=1.0,
+        )
+
+    def make_check(
+        self,
+        state: PublisherState,
+        max_slot_distance: int = 10,
+        abandoned_slot_distance: int = 100,
+    ) -> PublisherOfflineCheck:
+        """Helper to create PublisherOfflineCheck with config."""
+        return PublisherOfflineCheck(
+            state,
+            {
+                "max_slot_distance": max_slot_distance,
+                "abandoned_slot_distance": abandoned_slot_distance,
+            },
+        )
+
+    def run_check_with_datetime(
+        self,
+        check: PublisherOfflineCheck,
+        check_datetime: datetime,
+        expected: bool | None = None,
+    ) -> bool:
+        """Run check with mocked datetime and optionally assert result."""
+        with patch("pyth_observer.check.publisher.datetime") as mock_datetime:
+            mock_datetime.now.return_value = check_datetime
+            result = check.run()
+            if expected is not None:
+                assert result is expected
+            return result
+
+    def test_market_closed_passes_check(self):
+        """Test that check passes when market is closed."""
+        # Market schedule that's always closed (C = closed)
+        closed_schedule = MarketSchedule("America/New_York;C,C,C,C,C,C,C;")
+        state = self.make_state(
+            publisher_slot=100,
+            latest_block_slot=200,
+            schedule=closed_schedule,
+        )
+        check = self.make_check(state, max_slot_distance=10, abandoned_slot_distance=50)
+
+        # Should pass regardless of slot distance when market is closed
+        assert check.run() is True
+
+    def test_market_open_within_max_distance_passes(self):
+        """Test that check passes when slot distance is within max_slot_distance."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=105)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        assert check.run() is True
+
+    def test_market_open_exceeds_max_distance_fails(self):
+        """Test that check fails when slot distance exceeds max_slot_distance but not abandoned."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=120)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        assert check.run() is False
+
+    def test_market_open_exceeds_abandoned_distance_passes(self):
+        """Test that check passes when slot distance exceeds abandoned_slot_distance."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=250)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        assert check.run() is True
+
+    def test_boundary_at_max_slot_distance(self):
+        """Test boundary condition at max_slot_distance."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=110)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        assert check.run() is False
+
+    def test_boundary_below_max_slot_distance(self):
+        """Test boundary condition just below max_slot_distance."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=109)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        assert check.run() is True
+
+    def test_boundary_at_abandoned_slot_distance(self):
+        """Test boundary condition at abandoned_slot_distance."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=200)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        # Distance is exactly 100, which is not > 100, so should fail
+        assert check.run() is False
+
+    def test_boundary_above_abandoned_slot_distance(self):
+        """Test boundary condition just above abandoned_slot_distance."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=201)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        # Distance is 101, which is > 100, so should pass (abandoned)
+        assert check.run() is True
+
+    def test_different_configurations(self):
+        """Test with different configuration values."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=150)
+
+        # Test with larger max_slot_distance - distance is 50, which is < 60, so should pass
+        check1 = self.make_check(
+            state, max_slot_distance=60, abandoned_slot_distance=200
+        )
+        assert check1.run() is True
+
+        # Test with smaller abandoned_slot_distance - distance is 50, which is > 40, so should pass (abandoned)
+        check2 = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=40
+        )
+        assert check2.run() is True
+
+    def test_zero_distance_passes(self):
+        """Test that zero slot distance passes the check."""
+        state = self.make_state(publisher_slot=100, latest_block_slot=100)
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        assert check.run() is True
+
+    def test_market_schedule_variations(self):
+        """Test with different market schedule patterns."""
+        # Test with weekday-only schedule (Mon-Fri open)
+        weekday_schedule = MarketSchedule("America/New_York;O,O,O,O,O,C,C;")
+        state = self.make_state(
+            publisher_slot=100,
+            latest_block_slot=120,
+            schedule=weekday_schedule,
+        )
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        # Test on a Monday (market open) - should fail because market is open and distance exceeds max
+        monday_open = datetime(
+            2024, 1, 15, 14, 0, 0, tzinfo=ZoneInfo("America/New_York")
+        )
+        self.run_check_with_datetime(check, monday_open, expected=False)
+
+        # Test on a Sunday (market closed) - should pass because market is closed
+        sunday_closed = datetime(
+            2024, 1, 14, 14, 0, 0, tzinfo=ZoneInfo("America/New_York")
+        )
+        self.run_check_with_datetime(check, sunday_closed, expected=True)
+
+    def test_market_opening_detects_offline_publisher(self):
+        """Test that when market opens, an offline publisher triggers the check."""
+        # Use a weekday-only schedule (Mon-Fri open, weekends closed)
+        weekday_schedule = MarketSchedule("America/New_York;O,O,O,O,O,C,C;")
+        # Create a state where publisher is offline (slot distance exceeds max)
+        state = self.make_state(
+            publisher_slot=100,
+            latest_block_slot=120,
+            schedule=weekday_schedule,
+        )
+        check = self.make_check(
+            state, max_slot_distance=10, abandoned_slot_distance=100
+        )
+
+        # First, verify market closed - should pass even with offline publisher
+        market_closed_time = datetime(
+            2024, 1, 14, 23, 59, 59, tzinfo=ZoneInfo("America/New_York")
+        )  # Sunday night (market closed)
+        self.run_check_with_datetime(check, market_closed_time, expected=True)
+
+        # Now market opens - check should fire because publisher is offline
+        # Distance is 20, which exceeds max_slot_distance of 10
+        market_open_time = datetime(
+            2024, 1, 15, 0, 0, 0, tzinfo=ZoneInfo("America/New_York")
+        )  # Monday morning (market open)
+        self.run_check_with_datetime(check, market_open_time, expected=False)