""" 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)