astro-mcp/tests/test_tools.py

"""
Integration tests for MCP tools with mocked ephemeris client.
"""

from __future__ import annotations

import asyncio
import json
from unittest.mock import AsyncMock, patch

import pytest
from fastapi.testclient import TestClient

from src.astro_mcp.server import create_app

# ── Mock ephemeris get_sky_state response ───────────────────────────

def make_mock_sky_state(
    bodies: list[dict] | None = None,
    lst: float = 6.0,
) -> dict:
    """Create a realistic mock get_sky_state response."""
    if bodies is None:
        bodies = [
            {"body": "sun", "ecliptic_lon": 25.0, "ecliptic_lat": 0.0,
             "distance_au": 1.0, "speed_lon": 1.0, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 1.5, "dec": 10.0},
            {"body": "moon", "ecliptic_lon": 120.5, "ecliptic_lat": 2.0,
             "distance_au": 0.00257, "speed_lon": 13.0, "speed_lat": 0.5,
             "speed_dist": 0.0, "ra": 8.2, "dec": 15.0},
            {"body": "mercury", "ecliptic_lon": 30.0, "ecliptic_lat": 1.0,
             "distance_au": 0.5, "speed_lon": 1.5, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 2.0, "dec": 12.0},
            {"body": "venus", "ecliptic_lon": 95.0, "ecliptic_lat": -1.0,
             "distance_au": 0.7, "speed_lon": 0.8, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 6.5, "dec": 20.0},
            {"body": "mars", "ecliptic_lon": 200.0, "ecliptic_lat": 3.0,
             "distance_au": 1.5, "speed_lon": 0.6, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 13.5, "dec": -5.0},
            {"body": "jupiter", "ecliptic_lon": 250.0, "ecliptic_lat": 1.0,
             "distance_au": 5.2, "speed_lon": 0.1, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 16.8, "dec": -15.0},
            {"body": "saturn", "ecliptic_lon": 300.0, "ecliptic_lat": -2.0,
             "distance_au": 9.5, "speed_lon": 0.05, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 20.2, "dec": -20.0},
            {"body": "uranus", "ecliptic_lon": 330.0, "ecliptic_lat": 0.5,
             "distance_au": 19.2, "speed_lon": 0.02, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 22.5, "dec": -10.0},
            {"body": "neptune", "ecliptic_lon": 310.0, "ecliptic_lat": -1.0,
             "distance_au": 30.0, "speed_lon": 0.01, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 21.0, "dec": -18.0},
            {"body": "pluto", "ecliptic_lon": 260.0, "ecliptic_lat": 2.0,
             "distance_au": 39.5, "speed_lon": -0.01, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 17.5, "dec": -22.0},
            {"body": "chiron", "ecliptic_lon": 15.0, "ecliptic_lat": 0.0,
             "distance_au": 10.0, "speed_lon": 0.03, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 1.0, "dec": 8.0},
            {"body": "true_node", "ecliptic_lon": 45.0, "ecliptic_lat": 0.0,
             "distance_au": 0.0, "speed_lon": -0.05, "speed_lat": 0.0,
             "speed_dist": 0.0, "ra": 3.0, "dec": 15.0},
        ]

    return {
        "input": {"datetime": "2026-06-02T12:00:00Z", "lat": 47.0, "lon": 8.0,
                  "elevation": 0.0, "geocentric": True},
        "timestamp_utc": "2026-06-02T12:00:00Z",
        "julian_day": 2461172.0,
        "planetary_positions": {
            "input": {"datetime": "2026-06-02T12:00:00Z", "lat": 47.0, "lon": 8.0,
                      "elevation": 0.0, "geocentric": True},
            "timestamp_utc": "2026-06-02T12:00:00Z",
            "julian_day": 2461172.0,
            "bodies": bodies,
        },
        "solar_events": {"date": "2026-06-02", "events_jd": {}},
        "lunar_state": {"phase_name": "Waxing Gibbous", "illumination_fraction": 0.75},
        "sidereal_time": {
            "greenwich_sidereal_time": lst,
            "local_sidereal_time": lst,
            "obliquity_of_ecliptic": 23.4367,
        },
        "houses": {
            "system": "PLACIDUS",
            "cusps": [
                {"house": 1, "absolute_lon": 98.0, "sign": "Cancer", "abbreviation": "Cn", "degree": 8.0},
                {"house": 2, "absolute_lon": 128.0, "sign": "Leo", "abbreviation": "Le", "degree": 8.0},
                {"house": 3, "absolute_lon": 158.0, "sign": "Virgo", "abbreviation": "Vi", "degree": 8.0},
                {"house": 4, "absolute_lon": 188.0, "sign": "Libra", "abbreviation": "Li", "degree": 8.0},
                {"house": 5, "absolute_lon": 218.0, "sign": "Scorpius", "abbreviation": "Sc", "degree": 8.0},
                {"house": 6, "absolute_lon": 248.0, "sign": "Sagittarius", "abbreviation": "Sg", "degree": 8.0},
                {"house": 7, "absolute_lon": 278.0, "sign": "Capricornus", "abbreviation": "Cp", "degree": 8.0},
                {"house": 8, "absolute_lon": 308.0, "sign": "Aquarius", "abbreviation": "Aq", "degree": 8.0},
                {"house": 9, "absolute_lon": 338.0, "sign": "Pisces", "abbreviation": "Pi", "degree": 8.0},
                {"house": 10, "absolute_lon": 8.0, "sign": "Aries", "abbreviation": "Ar", "degree": 8.0},
                {"house": 11, "absolute_lon": 38.0, "sign": "Taurus", "abbreviation": "Ta", "degree": 8.0},
                {"house": 12, "absolute_lon": 68.0, "sign": "Gemini", "abbreviation": "Ge", "degree": 8.0},
            ],
            "ascendant": 98.0,
            "midheaven": 8.0,
            "descendant": 278.0,
            "imum_coeli": 188.0,
        },
    }


# ── Fixtures ────────────────────────────────────────────────────────

@pytest.fixture
def mock_sky_state():
    return make_mock_sky_state()


@pytest.fixture
def app_with_mock(mock_sky_state):
    """Create app with mocked ephemeris client."""
    with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
        mock.return_value = mock_sky_state
        app = create_app()
        yield app, mock


# ── Tests: get_planetary_positions ──────────────────────────────────

class TestGetPlanetaryPositions:
    def test_returns_enhanced_bodies(self, app_with_mock):
        app, mock = app_with_mock
        client = TestClient(app)
        res = client.get("/health")
        assert res.status_code == 200

    def test_tool_call_via_mock(self, mock_sky_state):
        """Test the tool function directly with mocked ephemeris client."""
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.get_planetary_positions(datetime="2026-06-02T12:00:00Z", lat=47.0, lon=8.0)
            )

        assert "bodies" in result
        assert len(result["bodies"]) == 12

        # Check sun is in Aries (25°)
        sun = [b for b in result["bodies"] if b["body"] == "sun"][0]
        assert sun["sign"] == "Aries"
        assert abs(sun["degree_within_sign"] - 25.0) < 0.01
        assert sun["retrograde"] is False

        # Check Pluto is retrograde (speed < 0)
        pluto = [b for b in result["bodies"] if b["body"] == "pluto"][0]
        assert pluto["retrograde"] is True

    def test_bodies_filter(self, mock_sky_state):
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.get_planetary_positions(
                    datetime="2026-06-02T12:00:00Z",
                    lat=47.0, lon=8.0,
                    bodies=["sun", "moon"],
                )
            )

        assert len(result["bodies"]) == 2
        names = {b["body"] for b in result["bodies"]}
        assert names == {"sun", "moon"}

    def test_error_handling(self):
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = {"error": "connection refused"}
            result = asyncio.run(
                tools.get_planetary_positions(datetime="2026-06-02T12:00:00Z")
            )

        assert "error" in result


# ── Tests: calculate_natal_chart ────────────────────────────────────

class TestCalculateNatalChart:
    def test_full_natal_chart(self, mock_sky_state):
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.calculate_natal_chart(
                    birth_datetime="2026-06-02T12:00:00Z",
                    latitude=47.0,
                    longitude=8.0,
                )
            )

        assert result["chart_type"] == "natal"
        assert "planets" in result
        assert "houses" in result
        assert "aspects" in result
        assert "angles" in result

        # 12 planets
        assert len(result["planets"]) == 12

        # 12 houses
        assert len(result["houses"]) == 12

        # Each planet has required fields
        for p in result["planets"]:
            assert "body" in p
            assert "sign" in p
            assert "degree_within_sign" in p
            assert "house" in p
            assert "retrograde" in p
            assert 1 <= p["house"] <= 12

        # Angles
        for key in ("ascendant", "midheaven", "descendant", "imum_coeli"):
            assert key in result["angles"]
            assert "sign" in result["angles"][key]

        # Aspects have required fields
        for asp in result["aspects"]:
            assert "body1" in asp
            assert "body2" in asp
            assert "aspect" in asp
            assert "orb" in asp

    def test_custom_house_system(self, mock_sky_state):
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.calculate_natal_chart(
                    birth_datetime="2026-06-02T12:00:00Z",
                    latitude=47.0,
                    longitude=8.0,
                    house_system="equal",
                )
            )

        assert result["input"]["house_system"] == "equal"
        assert len(result["houses"]) == 12


# ── Tests: calculate_transit_chart ──────────────────────────────────

class TestCalculateTransitChart:
    def test_transit_chart(self, mock_sky_state):
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.calculate_transit_chart(
                    birth_datetime="2000-01-01T12:00:00Z",
                    transit_datetime="2026-06-02T12:00:00Z",
                    latitude=47.0,
                    longitude=8.0,
                )
            )

        assert result["chart_type"] == "transit"
        assert "natal_planets" in result
        assert "transiting_planets" in result
        assert "aspects" in result
        assert "houses" in result

        # Transit planets should have natal_house
        for tp in result["transiting_planets"]:
            assert "natal_house" in tp
            assert 1 <= tp["natal_house"] <= 12


# ── Tests: calculate_synastry_chart ─────────────────────────────────

class TestCalculateSynastryChart:
    def test_synastry_chart(self, mock_sky_state):
        from src.astro_mcp import tools

        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.calculate_synastry_chart(
                    person1_datetime="2000-01-01T12:00:00Z",
                    person1_latitude=47.0,
                    person1_longitude=8.0,
                    person2_datetime="1995-06-15T08:00:00Z",
                    person2_latitude=52.0,
                    person2_longitude=13.0,
                )
            )

        assert result["chart_type"] == "synastry"
        assert "chart1_natal" in result
        assert "chart2_natal" in result
        assert "interaspects" in result
        assert "house_overlays" in result
        assert "composite_chart" in result
        assert "davison_chart" in result

        # House overlays
        assert "person2_in_person1_houses" in result["house_overlays"]
        assert "person1_in_person2_houses" in result["house_overlays"]

        # Composite chart has planets
        assert "planets" in result["composite_chart"]


# ── Tests: calculate_composite_chart ─────────────────────────────────

class TestCalculateCompositeChart:
    def test_composite_chart(self):
        from src.astro_mcp import tools
        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = make_mock_sky_state()
            import asyncio
            result = asyncio.run(
                tools.calculate_composite_chart(
                    person1_datetime="2000-01-01T12:00:00Z",
                    person1_latitude=47.0,
                    person1_longitude=8.0,
                    person2_datetime="1995-06-15T08:00:00Z",
                    person2_latitude=52.0,
                    person2_longitude=13.0,
                )
            )
        assert result["chart_type"] == "composite"
        assert "planets" in result
        assert "houses" in result
        assert "aspects" in result
        assert "angles" in result
        assert "composite_location" in result
        assert len(result["planets"]) > 0
        assert len(result["houses"]) == 12


# ── Tests: list_house_systems ───────────────────────────────────────

class TestListHouseSystems:
    def test_returns_systems(self):
        from src.astro_mcp import tools
        result = tools.list_house_systems()
        assert "systems" in result
        ids = [s["id"] for s in result["systems"]]
        assert "placidus" in ids
        assert "equal" in ids
        assert "whole_sign" in ids


# ── Tests: person_manage (stub) ─────────────────────────────────────
class TestPersonManage:
    def test_add_and_get(self, tmp_path, monkeypatch):
        """Test add + get round-trip with real storage."""
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio

        # Add
        add_result = asyncio.run(tools.person_manage(
            action="add",
            name="Lucky",
            birth_datetime="1990-05-15T10:30:00Z",
            latitude=47.3,
            longitude=8.5,
            nickname="lucky",
        ))
        assert "person" in add_result
        assert add_result["person"]["name"] == "Lucky"
        person_id = add_result["person"]["id"]

        # Get by ID
        get_result = asyncio.run(tools.person_manage(
            action="get", person_id=person_id,
        ))
        assert get_result["person"]["name"] == "Lucky"

        # Get by nickname
        get_result2 = asyncio.run(tools.person_manage(
            action="get", nickname="lucky",
        ))
        assert get_result2["person"]["id"] == person_id

    def test_list_persons(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        for i in range(3):
            asyncio.run(tools.person_manage(
                action="add", name=f"P{i}",
                birth_datetime="2000-01-01T12:00:00Z",
                latitude=0.0, longitude=0.0,
            ))

        result = asyncio.run(tools.person_manage(action="list"))
        assert result["count"] == 3

    def test_update_person(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        add_result = asyncio.run(tools.person_manage(
            action="add", name="Original",
            birth_datetime="2000-01-01T12:00:00Z",
            latitude=0.0, longitude=0.0,
        ))
        pid = add_result["person"]["id"]

        update_result = asyncio.run(tools.person_manage(
            action="update", person_id=pid, name="Updated",
        ))
        assert update_result["person"]["name"] == "Updated"

    def test_delete_person(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        add_result = asyncio.run(tools.person_manage(
            action="add", name="ToDelete",
            birth_datetime="2000-01-01T12:00:00Z",
            latitude=0.0, longitude=0.0,
        ))
        pid = add_result["person"]["id"]

        del_result = asyncio.run(tools.person_manage(
            action="delete", person_id=pid,
        ))
        assert del_result["deleted"] is True

        get_result = asyncio.run(tools.person_manage(
            action="get", person_id=pid,
        ))
        assert "error" in get_result

    def test_missing_required_fields(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        result = asyncio.run(tools.person_manage(
            action="add", name="Test",
        ))
        assert "error" in result

    def test_unknown_action(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        result = asyncio.run(tools.person_manage(action="bogus"))
        assert "error" in result


# ── Tests: get_transit_preview (stub) ───────────────────────────────

class TestTransitPreview:
    def test_invalid_date_format(self):
        from src.astro_mcp import tools
        result = asyncio.run(
            tools.get_transit_preview(
                birth_datetime="1980-01-01T12:00:00Z",
                latitude=47.0, longitude=8.0,
                start_date="not-a-date",
                end_date="2026-07-01",
            )
        )
        assert "error" in result

    def test_end_before_start(self):
        from src.astro_mcp import tools
        result = asyncio.run(
            tools.get_transit_preview(
                birth_datetime="1980-01-01T12:00:00Z",
                latitude=47.0, longitude=8.0,
                start_date="2026-07-01",
                end_date="2026-06-01",
            )
        )
        assert "error" in result

    def test_range_too_large(self):
        from src.astro_mcp import tools
        result = asyncio.run(
            tools.get_transit_preview(
                birth_datetime="1980-01-01T12:00:00Z",
                latitude=47.0, longitude=8.0,
                start_date="2026-01-01",
                end_date="2027-06-01",
            )
        )
        assert "error" in result

    def test_min_significance(self, mock_sky_state):
        from src.astro_mcp import tools
        with patch("src.astro_mcp.tools.call_sky_state", new_callable=AsyncMock) as mock:
            mock.return_value = mock_sky_state
            result = asyncio.run(
                tools.get_transit_preview(
                    birth_datetime="1980-01-01T12:00:00Z",
                    latitude=47.0, longitude=8.0,
                    start_date="2026-06-01",
                    end_date="2026-06-03",
                    min_significance=5.0,
                )
            )
        assert "days" in result


# ── Tests: _byId tools ───────────────────────────────────────────────

class TestByIdTools:
    def test_calculate_natal_chart_by_id(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        pid = asyncio.run(tools.person_manage(
            action="add", name="Lucky",
            birth_datetime="1980-06-15T10:30:00",
            latitude=47.0, longitude=8.0,
            nickname="lucky", birthplace="Zurich",
        ))["person"]["id"]

        result = asyncio.run(tools.calculate_natal_chart_by_id(person_id=pid))
        assert "error" not in result
        assert result["chart_type"] == "natal"
        assert len(result["planets"]) == 13

    def test_calculate_transit_chart_by_id(self, tmp_path, monkeypatch):
        from src.astro_mcp import tools, storage
        db_path = tmp_path / "test2.sqlite3"
        monkeypatch.setattr(storage, "DB_PATH", db_path)
        storage._initialized = False

        import asyncio
        pid = asyncio.run(tools.person_manage(
            action="add", name="Transit",
            birth_datetime="1990-03-20T08:00:00",
            latitude=52.0, longitude=13.0,
            nickname="transit_test", birthplace="Berlin",
        ))["person"]["id"]

        result = asyncio.run(tools.calculate_transit_chart_by_id(
            person_id=pid, transit_datetime="2026-06-02T12:00:00",
        ))
        assert "error" not in result
        assert result["chart_type"] == "transit"

    def test_by_id_not_found(self):
        from src.astro_mcp import tools
        import asyncio
        result = asyncio.run(tools.calculate_natal_chart_by_id(person_id="nonexistent"))
        assert "error" in result