""" Tests for the person storage layer. """ from __future__ import annotations import asyncio import os import tempfile from pathlib import Path import pytest from src.astro_mcp import storage @pytest.fixture def tmp_db(tmp_path, monkeypatch): """Use a temporary database for each test.""" db_path = tmp_path / "test_astro.sqlite3" monkeypatch.setattr(storage, "DB_PATH", db_path) storage._initialized = False # Reset init state yield db_path storage._initialized = False # Cleanup class TestAddPerson: def test_add_person(self, tmp_db): result = asyncio.run(storage.add_person( name="Test Person", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) assert result["name"] == "Test Person" assert result["birth_datetime"] == "2000-01-01T12:00:00Z" assert result["latitude"] == 47.0 assert result["longitude"] == 8.0 assert result["elevation"] == 0.0 assert "id" in result assert "created_at" in result def test_add_person_with_nickname(self, tmp_db): result = asyncio.run(storage.add_person( name="Test Person", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, nickname="testy", )) assert result["nickname"] == "testy" def test_add_person_with_elevation(self, tmp_db): result = asyncio.run(storage.add_person( name="Test Person", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, elevation=500.0, )) assert result["elevation"] == 500.0 class TestGetPerson: def test_get_by_id(self, tmp_db): added = asyncio.run(storage.add_person( name="Test Person", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) fetched = asyncio.run(storage.get_person(person_id=added["id"])) assert fetched is not None assert fetched["id"] == added["id"] assert fetched["name"] == "Test Person" def test_get_by_nickname(self, tmp_db): asyncio.run(storage.add_person( name="Test Person", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, nickname="testy", )) fetched = asyncio.run(storage.get_person(nickname="testy")) assert fetched is not None assert fetched["nickname"] == "testy" def test_get_not_found(self, tmp_db): result = asyncio.run(storage.get_person(person_id="nonexistent")) assert result is None def test_get_no_params(self, tmp_db): result = asyncio.run(storage.get_person()) assert result is None class TestListPersons: def test_empty_list(self, tmp_db): result = asyncio.run(storage.list_persons()) assert result == [] def test_list_persons(self, tmp_db): for i in range(3): asyncio.run(storage.add_person( name=f"Person {i}", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) result = asyncio.run(storage.list_persons()) assert len(result) == 3 def test_list_ordered_newest_first(self, tmp_db): for i in range(3): asyncio.run(storage.add_person( name=f"Person {i}", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) result = asyncio.run(storage.list_persons()) # Newest first assert result[0]["name"] == "Person 2" assert result[-1]["name"] == "Person 0" class TestUpdatePerson: def test_update_name(self, tmp_db): added = asyncio.run(storage.add_person( name="Original", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) updated = asyncio.run(storage.update_person( person_id=added["id"], name="Updated", )) assert updated is not None assert updated["name"] == "Updated" assert updated["updated_at"] is not None def test_update_birth_data(self, tmp_db): added = asyncio.run(storage.add_person( name="Test", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) updated = asyncio.run(storage.update_person( person_id=added["id"], birth_datetime="1995-06-15T08:30:00Z", latitude=52.0, longitude=13.0, )) assert updated["birth_datetime"] == "1995-06-15T08:30:00Z" assert updated["latitude"] == 52.0 assert updated["longitude"] == 13.0 def test_update_not_found(self, tmp_db): result = asyncio.run(storage.update_person( person_id="nonexistent", name="Updated", )) assert result is None def test_update_no_fields_returns_unchanged(self, tmp_db): added = asyncio.run(storage.add_person( name="Test", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) updated = asyncio.run(storage.update_person(person_id=added["id"])) assert updated is not None assert updated["name"] == "Test" # Unchanged class TestDeletePerson: def test_delete_person(self, tmp_db): added = asyncio.run(storage.add_person( name="Test", birth_datetime="2000-01-01T12:00:00Z", latitude=47.0, longitude=8.0, )) deleted = asyncio.run(storage.delete_person(added["id"])) assert deleted is True # Verify it's gone fetched = asyncio.run(storage.get_person(person_id=added["id"])) assert fetched is None def test_delete_not_found(self, tmp_db): result = asyncio.run(storage.delete_person("nonexistent")) assert result is False