astro-mcp/tests/test_live_charts.py

"""
End-to-end integration tests: call the LIVE astro-mcp server via MCP SSE
and compare results against reference data.

Reference values were obtained by querying the same Swiss Ephemeris that
the ephemeris-mcp server uses, so these tests verify end-to-end correctness
of the full pipeline: MCP client -> astro-mcp -> ephemeris-mcp -> Swiss Ephemeris.

These tests require both servers to be running:
  - ephemeris-mcp  (port 7015)
  - astro-mcp      (port 7016)

Run with: pytest tests/test_live_charts.py -v
Skip with: SKIP_LIVE_TESTS=1 pytest tests/
"""
from __future__ import annotations

import asyncio
import os
from datetime import timedelta
from typing import Any

import pytest
from mcp import ClientSession
from mcp.client.sse import sse_client

pytestmark = [
    pytest.mark.skipif(
        os.environ.get("SKIP_LIVE_TESTS", "0") == "1",
        reason="SKIP_LIVE_TESTS=1",
    ),
    pytest.mark.live,
]

ASTRO_MCP_URL = os.environ.get(
    "ASTRO_MCP_URL", "http://192.168.0.249:7016/mcp/sse"
)


async def call_astro_tool(
    tool_name: str, arguments: dict[str, Any], timeout: float = 30.0
) -> dict[str, Any]:
    """Call a tool on the live astro-mcp server via MCP SSE."""
    url = ASTRO_MCP_URL
    if not url.endswith("/mcp/sse"):
        url = url.rstrip("/") + "/mcp/sse"

    async with sse_client(url, timeout=timeout, sse_read_timeout=timeout) as streams:
        async with ClientSession(
            *streams, read_timeout_seconds=timedelta(seconds=timeout)
        ) as session:
            await session.initialize()
            result = await session.call_tool(tool_name, arguments)

            payload = getattr(result, "structuredContent", None)
            if isinstance(payload, dict):
                return payload

            content_items = getattr(result, "content", []) or []
            for item in content_items:
                text = getattr(item, "text", None)
                if isinstance(text, str) and text.strip():
                    import json
                    try:
                        decoded = json.loads(text)
                    except Exception:
                        continue
                    if isinstance(decoded, dict):
                        return decoded

            error_texts = []
            for item in content_items:
                text = getattr(item, "text", None)
                if isinstance(text, str) and text.strip():
                    error_texts.append(text.strip())
            if error_texts:
                return {"error": error_texts[0]}
            return {}


# ── Reference data ────────────────────────────────────────────────────
# Values captured from the live server (Swiss Ephemeris via ephemeris-mcp).
# These are the ground truth for our specific ephemeris version.

# Albert Einstein: 1879-03-14 11:30 AM LMT (UTC+0:53), Ulm, Germany
# Rodden AA (birth certificate)
EINSTEIN = {
    "name": "Albert Einstein",
    "datetime": "1879-03-14T11:30:00+00:53",
    "lat": 48.39841,
    "lon": 9.99155,
    "planets": {
        # body: (sign, degree_in_sign, retrograde)
        "sun":       ("Pisces", 23.50, False),
        "moon":      ("Sagittarius", 14.40, False),
        "mercury":   ("Aries", 3.13, False),
        "venus":     ("Aries", 16.97, False),
        "mars":      ("Capricornus", 26.91, False),
        "jupiter":   ("Aquarius", 27.48, False),
        "saturn":    ("Aries", 4.19, False),
        "uranus":    ("Virgo", 1.29, True),
        "neptune":   ("Taurus", 7.87, False),
        "pluto":     ("Taurus", 24.73, False),
        "chiron":    ("Aries", 0.00, False),
        "true_node": ("Aquarius", 2.73, True),
        "mean_node": ("Aquarius", 1.48, True),
    },
    "angles": {
        "ascendant": ("Cancer", 8.90),
        "midheaven": ("Pisces", 12.39),
    },
}

# Chaka Khan: 1953-03-23 9:05 PM CST (UTC-6), Chicago, IL
# Rodden AA
CHAKA_KHAN = {
    "name": "Chaka Khan",
    "datetime": "1953-03-23T21:05:00-06:00",
    "lat": 41.87811,
    "lon": -87.62980,
    "planets": {
        "sun":       ("Aries", 3.19, False),
        "moon":      ("Cancer", 23.41, False),
        "mercury":   ("Pisces", 22.83, True),
        "venus":     ("Taurus", 1.32, True),
        "mars":      ("Taurus", 2.80, False),
        "jupiter":   ("Taurus", 19.80, False),
        "saturn":    ("Libra", 25.53, True),
        "uranus":    ("Cancer", 14.43, False),
        "neptune":   ("Libra", 23.05, True),
        "pluto":     ("Leo", 21.15, True),
        "chiron":    ("Aries", 0.00, False),
        "true_node": ("Aquarius", 11.14, False),
        "mean_node": ("Aquarius", 9.73, True),
    },
    "angles": {
        "ascendant": ("Scorpius", 8.93),
        "midheaven": ("Leo", 22.33),
    },
}


def _check_planets(result: dict, ref: dict, tol: float = 0.5) -> list[str]:
    """Check planetary positions against reference. Returns list of errors."""
    errors = []
    planets = {p["body"]: p for p in result.get("planets", [])}

    for name, (ref_sign, ref_deg, ref_retro) in ref["planets"].items():
        if name not in planets:
            errors.append(f"{name}: missing from result")
            continue

        p = planets[name]
        ref_abs = SIGN_START[ref_sign] + ref_deg

        if p["sign"] != ref_sign:
            errors.append(f"{name}: sign {p['sign']} != {ref_sign}")
        if abs(p["degree_within_sign"] - ref_deg) > tol:
            errors.append(
                f"{name}: deg {p['degree_within_sign']:.2f} != {ref_deg}°"
            )
        if abs(p["absolute_lon"] - ref_abs) > tol:
            errors.append(
                f"{name}: abs_lon {p['absolute_lon']:.2f} != {ref_abs:.2f}"
            )
        if p["retrograde"] != ref_retro:
            errors.append(
                f"{name}: retrograde {p['retrograde']} != {ref_retro}"
            )

    return errors


def _check_angles(result: dict, ref: dict, tol: float = 0.5) -> list[str]:
    """Check angles against reference. Returns list of errors."""
    errors = []
    angles = result.get("angles", {})

    for angle_name, (ref_sign, ref_deg) in ref["angles"].items():
        if angle_name not in angles:
            errors.append(f"{angle_name}: missing from result")
            continue

        a = angles[angle_name]
        ref_abs = SIGN_START[ref_sign] + ref_deg

        if a["sign"] != ref_sign:
            errors.append(f"{angle_name}: sign {a['sign']} != {ref_sign}")
        if abs(a["absolute_lon"] - ref_abs) > tol:
            errors.append(
                f"{angle_name}: abs_lon {a['absolute_lon']:.2f} != {ref_abs:.2f}"
            )

    return errors


SIGN_START = {
    "Aries": 0, "Taurus": 30, "Gemini": 60, "Cancer": 90,
    "Leo": 120, "Virgo": 150, "Libra": 180, "Scorpius": 210,
    "Sagittarius": 240, "Capricornus": 270, "Aquarius": 300, "Pisces": 330,
}


# ── Tests: Einstein ───────────────────────────────────────────────────

class TestLiveEinstein:
    """Call live astro-mcp server for Einstein's chart and verify."""

    @pytest.mark.asyncio
    async def test_natal_chart_planets(self):
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": EINSTEIN["datetime"],
                "latitude": EINSTEIN["lat"],
                "longitude": EINSTEIN["lon"],
                "house_system": "placidus",
            },
        )
        assert "error" not in result, f"Server error: {result.get('error')}"
        errors = _check_planets(result, EINSTEIN)
        assert not errors, "Planet mismatches:\n" + "\n".join(errors)

    @pytest.mark.asyncio
    async def test_natal_chart_angles(self):
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": EINSTEIN["datetime"],
                "latitude": EINSTEIN["lat"],
                "longitude": EINSTEIN["lon"],
                "house_system": "placidus",
            },
        )
        assert "error" not in result
        errors = _check_angles(result, EINSTEIN)
        assert not errors, "Angle mismatches:\n" + "\n".join(errors)

    @pytest.mark.asyncio
    async def test_natal_chart_houses(self):
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": EINSTEIN["datetime"],
                "latitude": EINSTEIN["lat"],
                "longitude": EINSTEIN["lon"],
                "house_system": "placidus",
            },
        )
        assert "error" not in result
        assert "houses" in result
        assert len(result["houses"]) == 12
        house_numbers = sorted(h["house"] for h in result["houses"])
        assert house_numbers == list(range(1, 13))
        for h in result["houses"]:
            assert "sign" in h
            assert "degree" in h
            assert "absolute_lon" in h

    @pytest.mark.asyncio
    async def test_natal_chart_aspects(self):
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": EINSTEIN["datetime"],
                "latitude": EINSTEIN["lat"],
                "longitude": EINSTEIN["lon"],
                "house_system": "placidus",
            },
        )
        assert "error" not in result
        assert "aspects" in result
        assert len(result["aspects"]) > 0
        for asp in result["aspects"]:
            assert "body1" in asp
            assert "body2" in asp
            assert "aspect" in asp
            assert "orb" in asp


# ── Tests: Chaka Khan ─────────────────────────────────────────────────

class TestLiveChakaKhan:
    """Call live astro-mcp server for Chaka Khan's chart and verify."""

    @pytest.mark.asyncio
    async def test_natal_chart_planets(self):
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": CHAKA_KHAN["datetime"],
                "latitude": CHAKA_KHAN["lat"],
                "longitude": CHAKA_KHAN["lon"],
                "house_system": "placidus",
            },
        )
        assert "error" not in result, f"Server error: {result.get('error')}"
        errors = _check_planets(result, CHAKA_KHAN)
        assert not errors, "Planet mismatches:\n" + "\n".join(errors)

    @pytest.mark.asyncio
    async def test_natal_chart_angles(self):
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": CHAKA_KHAN["datetime"],
                "latitude": CHAKA_KHAN["lat"],
                "longitude": CHAKA_KHAN["lon"],
                "house_system": "placidus",
            },
        )
        assert "error" not in result
        errors = _check_angles(result, CHAKA_KHAN)
        assert not errors, "Angle mismatches:\n" + "\n".join(errors)

    @pytest.mark.asyncio
    async def test_retrograde_planets(self):
        """Chaka Khan has 6 retrograde planets."""
        result = await call_astro_tool(
            "calculate_natal_chart",
            {
                "birth_datetime": CHAKA_KHAN["datetime"],
                "latitude": CHAKA_KHAN["lat"],
                "longitude": CHAKA_KHAN["lon"],
                "house_system": "placidus",
                "include_overview": True,
            },
        )
        assert "error" not in result
        assert "overview" in result

        retro = result["overview"]["retrograde_planets"]
        retro_names = {p["body"] for p in retro}
        expected = {"mercury", "venus", "saturn", "neptune", "pluto", "mean_node"}
        assert retro_names == expected, \
            f"Retrograde mismatch: expected {expected}, got {retro_names}"


# ── Tests: _byId tools with DB celebrities ────────────────────────────

class TestLiveByNickname:
    """Test _byId tools using celebrities in the persons DB."""

    @pytest.mark.asyncio
    async def test_einstein_by_nickname(self):
        result = await call_astro_tool(
            "calculate_natal_chart_by_id",
            {"person_id": "einstein", "house_system": "placidus"},
        )
        assert "error" not in result, f"Server error: {result.get('error')}"
        assert result.get("chart_type") == "natal"
        assert "planets" in result
        assert "houses" in result
        assert "angles" in result
        errors = _check_planets(result, EINSTEIN)
        assert not errors, "Planet mismatches:\n" + "\n".join(errors)

    @pytest.mark.asyncio
    async def test_chaka_by_nickname(self):
        result = await call_astro_tool(
            "calculate_natal_chart_by_id",
            {"person_id": "chaka", "house_system": "placidus"},
        )
        assert "error" not in result
        assert result.get("chart_type") == "natal"
        errors = _check_planets(result, CHAKA_KHAN)
        assert not errors, "Planet mismatches:\n" + "\n".join(errors)