from __future__ import annotations import tempfile import unittest from pathlib import Path from unittest.mock import patch from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from ecospecies_api import repository SAMPLE_PAYLOAD = [ { "slug": "test-shad", "source_file": "Test Shad.txt", "title": "Test Shad (Alosa testus)", "common_name": "Test Shad", "scientific_name": "Alosa testus", "flelmr_code": "9999", "summary": "", "section_count": 2, "diagnostics": [ { "level": "warning", "code": "missing_summary", "message": "Summary/Abstract section is missing.", }, { "level": "warning", "code": "missing_citations", "message": "References section not found.", }, ], "sections": [ {"heading": "HEADER", "content": "Header content"}, {"heading": "HABITAT", "content": "Habitat content"}, ], } ] UPDATED_PAYLOAD = [ { "slug": "test-shad", "source_file": "Test Shad v2.txt", "title": "Test Shad Revised (Alosa testus)", "common_name": "Test Shad", "scientific_name": "Alosa testus revised", "flelmr_code": "1000", "summary": "Imported replacement summary.", "section_count": 2, "diagnostics": [ { "level": "warning", "code": "missing_flelmr_code", "message": "Replacement diagnostic.", } ], "sections": [ {"heading": "HEADER", "content": "Replacement header content"}, {"heading": "HABITAT", "content": "Replacement habitat content"}, ], } ] DIFFERENT_PAYLOAD = [ { "slug": "other-fish", "source_file": "Other Fish.txt", "title": "Other Fish (Pisces otherus)", "common_name": "Other Fish", "scientific_name": "Pisces otherus", "flelmr_code": "2000", "summary": "Other fish summary.", "section_count": 1, "diagnostics": [], "sections": [ {"heading": "HEADER", "content": "Other fish header"}, ], } ] class RepositoryWorkflowTests(unittest.TestCase): def setUp(self) -> None: self.tempdir = tempfile.TemporaryDirectory() db_path = Path(self.tempdir.name) / "test.db" self.engine = create_engine(f"sqlite:///{db_path}", future=True) self.session_local = sessionmaker( bind=self.engine, autoflush=False, autocommit=False, future=True, ) self.engine_patch = patch.object(repository, "create_db_engine", return_value=self.engine) self.session_patch = patch.object(repository, "SessionLocal", self.session_local) self.engine_patch.start() self.session_patch.start() repository.import_species_payload(SAMPLE_PAYLOAD) def tearDown(self) -> None: self.session_patch.stop() self.engine_patch.stop() self.engine.dispose() self.tempdir.cleanup() def test_import_filters_missing_summary_diagnostic_from_accepted_dataset(self) -> None: detail = repository.get_species_by_slug("test-shad") self.assertIsNotNone(detail) self.assertEqual(detail["section_count"], 2) self.assertEqual([section["position"] for section in detail["sections"]], [1, 2]) self.assertEqual([item["code"] for item in detail["diagnostics"]], ["missing_citations"]) def test_editorial_update_changes_publication_visibility_and_creates_audit(self) -> None: result = repository.update_species_editorial( slug="test-shad", publication_status="draft", summary="Editor-authored summary.", editor_notes="Needs another review pass.", is_archived=None, username="bob", ) self.assertIsNotNone(result) self.assertEqual(result["publication_status"], "draft") self.assertEqual(result["summary"], "Editor-authored summary.") self.assertEqual(result["last_modified_by"], "bob") self.assertEqual(repository.get_species_by_slug("test-shad"), None) editor_detail = repository.get_editor_species_detail("test-shad") audit = repository.list_species_audit("test-shad") self.assertIsNotNone(editor_detail) self.assertEqual(editor_detail["publication_status"], "draft") self.assertEqual(editor_detail["summary"], "Editor-authored summary.") self.assertEqual(editor_detail["editor_notes"], "Needs another review pass.") self.assertIsNotNone(audit) self.assertEqual(audit[0]["action"], "editorial_update") self.assertEqual(audit[0]["changed_by"], "bob") self.assertIn("summary", audit[0]["details"]) self.assertIn("publication_status", audit[0]["details"]) def test_section_update_records_section_audit_metadata(self) -> None: result = repository.update_species_section( slug="test-shad", section_position=2, content="Updated habitat content.", username="carol", ) self.assertIsNotNone(result) self.assertEqual(result["section"]["position"], 2) self.assertEqual(result["section"]["content"], "Updated habitat content.") self.assertEqual(result["last_modified_by"], "carol") self.assertEqual(sorted(result["changed_fields"].keys()), ["section_content"]) editor_detail = repository.get_editor_species_detail("test-shad") audit = repository.list_species_audit("test-shad") self.assertIsNotNone(editor_detail) self.assertEqual(editor_detail["sections"][1]["content"], "Updated habitat content.") self.assertIsNotNone(audit) self.assertEqual(audit[0]["action"], "section_update") self.assertEqual(audit[0]["changed_by"], "carol") self.assertEqual(audit[0]["details"]["section_position"], 2) self.assertEqual(audit[0]["details"]["section_heading"], "HABITAT") self.assertEqual( audit[0]["details"]["section_content"], {"from": "Habitat content", "to": "Updated habitat content."}, ) def test_reimport_preserves_editorial_state_and_audit_history(self) -> None: repository.update_species_editorial( slug="test-shad", publication_status="draft", summary="Editor-authored summary.", editor_notes="Needs another review pass.", is_archived=None, username="bob", ) repository.update_species_section( slug="test-shad", section_position=2, content="Updated habitat content.", username="carol", ) repository.import_species_payload(UPDATED_PAYLOAD) editor_detail = repository.get_editor_species_detail("test-shad") audit = repository.list_species_audit("test-shad") self.assertIsNotNone(editor_detail) self.assertEqual(editor_detail["source_file"], "Test Shad v2.txt") self.assertEqual(editor_detail["title"], "Test Shad Revised (Alosa testus)") self.assertEqual(editor_detail["scientific_name"], "Alosa testus revised") self.assertEqual(editor_detail["flelmr_code"], "1000") self.assertEqual(editor_detail["publication_status"], "draft") self.assertEqual(editor_detail["summary"], "Editor-authored summary.") self.assertEqual(editor_detail["editor_notes"], "Needs another review pass.") self.assertEqual(editor_detail["sections"][0]["content"], "Replacement header content") self.assertEqual(editor_detail["sections"][1]["content"], "Updated habitat content.") self.assertEqual([item["code"] for item in editor_detail["diagnostics"]], ["missing_flelmr_code"]) self.assertIsNotNone(audit) self.assertEqual(len(audit), 2) self.assertEqual([entry["action"] for entry in audit], ["section_update", "editorial_update"]) def test_reimport_updates_summary_when_no_editorial_override_exists(self) -> None: repository.import_species_payload(UPDATED_PAYLOAD) detail = repository.get_species_by_slug("test-shad") self.assertIsNotNone(detail) self.assertEqual(detail["summary"], "Imported replacement summary.") self.assertEqual(detail["sections"][0]["content"], "Replacement header content") def test_editor_can_archive_species_explicitly(self) -> None: result = repository.update_species_editorial( slug="test-shad", publication_status=None, summary=None, editor_notes=None, is_archived=True, username="dana", ) public_detail = repository.get_species_by_slug("test-shad") editor_detail = repository.get_editor_species_detail("test-shad") audit = repository.list_species_audit("test-shad") self.assertIsNotNone(result) self.assertTrue(result["is_archived"]) self.assertEqual(result["last_modified_by"], "dana") self.assertIsNone(public_detail) self.assertIsNotNone(editor_detail) self.assertTrue(editor_detail["is_archived"]) self.assertIsNotNone(audit) self.assertEqual(audit[0]["action"], "editorial_update") self.assertEqual(audit[0]["details"]["is_archived"], {"from": False, "to": True}) def test_editor_can_unarchive_species_explicitly(self) -> None: repository.update_species_editorial( slug="test-shad", publication_status=None, summary=None, editor_notes=None, is_archived=True, username="dana", ) result = repository.update_species_editorial( slug="test-shad", publication_status=None, summary=None, editor_notes=None, is_archived=False, username="erin", ) public_detail = repository.get_species_by_slug("test-shad") audit = repository.list_species_audit("test-shad") self.assertIsNotNone(result) self.assertFalse(result["is_archived"]) self.assertEqual(result["last_modified_by"], "erin") self.assertIsNotNone(public_detail) self.assertIsNotNone(audit) self.assertEqual(audit[0]["details"]["is_archived"], {"from": True, "to": False}) def test_missing_species_is_archived_instead_of_deleted(self) -> None: repository.import_species_payload(DIFFERENT_PAYLOAD) public_detail = repository.get_species_by_slug("test-shad") editor_detail = repository.get_editor_species_detail("test-shad") editor_items = repository.get_editor_species_list() audit = repository.list_species_audit("test-shad") self.assertIsNone(public_detail) self.assertIsNotNone(editor_detail) self.assertTrue(editor_detail["is_archived"]) self.assertEqual([item["slug"] for item in repository.list_species()], ["other-fish"]) self.assertEqual([item["slug"] for item in editor_items], ["other-fish", "test-shad"]) self.assertIsNotNone(audit) self.assertEqual(audit[0]["action"], "import_archive") self.assertEqual(audit[0]["details"]["is_archived"], {"from": False, "to": True}) def test_archived_species_is_restored_when_it_reappears(self) -> None: repository.import_species_payload(DIFFERENT_PAYLOAD) repository.import_species_payload(UPDATED_PAYLOAD) public_detail = repository.get_species_by_slug("test-shad") editor_detail = repository.get_editor_species_detail("test-shad") audit = repository.list_species_audit("test-shad") self.assertIsNotNone(public_detail) self.assertIsNotNone(editor_detail) self.assertFalse(editor_detail["is_archived"]) self.assertEqual(public_detail["summary"], "Imported replacement summary.") self.assertIsNotNone(audit) self.assertEqual(audit[0]["action"], "import_restore") self.assertEqual(audit[0]["details"]["is_archived"], {"from": True, "to": False}) if __name__ == "__main__": unittest.main()