juliaandClaude Sonnet 4.6
e8d48d4fcb
scripts/import_publications.py imports all osw=1 publications from
03-data/osws.xml — upserts pub rows, loads cover images, scans
03-data/osdb/{pubId}/ for fruit images and PDFs, links matched fruits
by osdb_number. Idempotent re-runs. 17 unit tests. Makefile updated
to include import_publications_test in make test. Added
scripts/README_import.md with run order. Added spec-first rule to
.claude/CLAUDE.md.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
237 lines
8.4 KiB
Python
237 lines
8.4 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
from import_publications import parse_publications, scan_pub_dir, import_publications
|
|
|
|
MINIMAL_XML_STR = """<?xml version="1.0" encoding="iso-8859-1" ?>
|
|
<root>
|
|
<obj>
|
|
<id>skip_me</id>
|
|
<name>Not a publication</name>
|
|
<author>Nobody</author>
|
|
<osw>0</osw>
|
|
</obj>
|
|
<obj>
|
|
<id>ber</id>
|
|
<name>Bernisches Stammregister</name>
|
|
<author>.</author>
|
|
<img>osdb/ber/ber_s0.jpg</img>
|
|
<osw>1</osw>
|
|
</obj>
|
|
<obj>
|
|
<id>cal</id>
|
|
<name>Obst- und Beeren</name>
|
|
<author>Calwer</author>
|
|
<img>osdb/cal/cal_s0.jpg</img>
|
|
<osw>1</osw>
|
|
</obj>
|
|
<obj>
|
|
<id>noc</id>
|
|
<name>No Cover</name>
|
|
<author>Someone</author>
|
|
<osw>1</osw>
|
|
</obj>
|
|
</root>
|
|
"""
|
|
MINIMAL_XML = MINIMAL_XML_STR.encode("iso-8859-1")
|
|
|
|
|
|
def _write_xml(tmp_dir, content=MINIMAL_XML):
|
|
p = Path(tmp_dir) / "osws.xml"
|
|
p.write_bytes(content)
|
|
return str(p)
|
|
|
|
|
|
class TestParsePublications(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp()
|
|
self.xml_path = _write_xml(self.tmp)
|
|
|
|
def test_excludes_osw_not_1(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
ids = [p["id"] for p in pubs]
|
|
self.assertNotIn("skip_me", ids)
|
|
|
|
def test_includes_osw_1(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
ids = [p["id"] for p in pubs]
|
|
self.assertIn("ber", ids)
|
|
self.assertIn("cal", ids)
|
|
|
|
def test_author_dot_becomes_none(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
ber = next(p for p in pubs if p["id"] == "ber")
|
|
self.assertIsNone(ber["author"])
|
|
|
|
def test_author_string_preserved(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
cal = next(p for p in pubs if p["id"] == "cal")
|
|
self.assertEqual(cal["author"], "Calwer")
|
|
|
|
def test_img_path_present(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
cal = next(p for p in pubs if p["id"] == "cal")
|
|
self.assertEqual(cal["img_path"], "osdb/cal/cal_s0.jpg")
|
|
|
|
def test_img_path_absent_is_none(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
noc = next(p for p in pubs if p["id"] == "noc")
|
|
self.assertIsNone(noc["img_path"])
|
|
|
|
def test_title_mapped(self):
|
|
pubs = parse_publications(self.xml_path)
|
|
cal = next(p for p in pubs if p["id"] == "cal")
|
|
self.assertEqual(cal["title"], "Obst- und Beeren")
|
|
|
|
|
|
class TestScanPubDir(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp()
|
|
|
|
def test_returns_empty_for_missing_directory(self):
|
|
result = scan_pub_dir(os.path.join(self.tmp, "nonexistent"), "xyz")
|
|
self.assertEqual(result, {})
|
|
|
|
def test_extracts_osdb_number_from_image(self):
|
|
d = Path(self.tmp)
|
|
(d / "apfel_ber_s0.jpg").write_bytes(b"img")
|
|
result = scan_pub_dir(str(d), "ber")
|
|
self.assertIn("apfel", result)
|
|
self.assertEqual(result["apfel"]["img_path"], str(d / "apfel_ber_s0.jpg"))
|
|
|
|
def test_extracts_osdb_number_from_pdf(self):
|
|
d = Path(self.tmp)
|
|
(d / "birne_cal.pdf").write_bytes(b"pdf")
|
|
result = scan_pub_dir(str(d), "cal")
|
|
self.assertIn("birne", result)
|
|
self.assertEqual(result["birne"]["pdf_path"], str(d / "birne_cal.pdf"))
|
|
|
|
def test_combines_img_and_pdf_for_same_fruit(self):
|
|
d = Path(self.tmp)
|
|
(d / "apfel_ber_s0.jpg").write_bytes(b"img")
|
|
(d / "apfel_ber.pdf").write_bytes(b"pdf")
|
|
result = scan_pub_dir(str(d), "ber")
|
|
self.assertIn("img_path", result["apfel"])
|
|
self.assertIn("pdf_path", result["apfel"])
|
|
|
|
def test_ignores_thumbnail_files(self):
|
|
d = Path(self.tmp)
|
|
(d / "apfel_ber_tn.jpg").write_bytes(b"tn")
|
|
result = scan_pub_dir(str(d), "ber")
|
|
self.assertEqual(result, {})
|
|
|
|
def test_multi_segment_osdb_number(self):
|
|
d = Path(self.tmp)
|
|
(d / "berner_grauechapfel_ber_s0.jpg").write_bytes(b"img")
|
|
result = scan_pub_dir(str(d), "ber")
|
|
self.assertIn("berner_grauechapfel", result)
|
|
|
|
|
|
class TestImportPublications(unittest.TestCase):
|
|
def _make_data_dir(self, pubs):
|
|
"""Build a temp data dir with osws.xml and osdb/<id>/ dirs."""
|
|
tmp = tempfile.mkdtemp()
|
|
lines = ['<?xml version="1.0" encoding="iso-8859-1" ?>', "<root>"]
|
|
for p in pubs:
|
|
lines.append(" <obj>")
|
|
lines.append(f" <id>{p['id']}</id>")
|
|
lines.append(f" <name>{p['name']}</name>")
|
|
author = p.get("author", "Test")
|
|
lines.append(f" <author>{author}</author>")
|
|
if p.get("img"):
|
|
lines.append(f" <img>{p['img']}</img>")
|
|
lines.append(" <osw>1</osw>")
|
|
lines.append(" </obj>")
|
|
lines.append("</root>")
|
|
Path(tmp, "osws.xml").write_bytes("\n".join(lines).encode("iso-8859-1"))
|
|
|
|
for p in pubs:
|
|
pub_dir = Path(tmp, "osdb", p["id"])
|
|
pub_dir.mkdir(parents=True, exist_ok=True)
|
|
for fruit in p.get("fruits", []):
|
|
if fruit.get("img"):
|
|
(pub_dir / f"{fruit['osdb_number']}_{p['id']}_s0.jpg").write_bytes(b"\x89PNG")
|
|
if fruit.get("pdf"):
|
|
(pub_dir / f"{fruit['osdb_number']}_{p['id']}.pdf").write_bytes(b"%PDF")
|
|
if p.get("cover_bytes"):
|
|
cover_rel = p["img"]
|
|
cover_path = Path(tmp, cover_rel)
|
|
cover_path.parent.mkdir(parents=True, exist_ok=True)
|
|
cover_path.write_bytes(p["cover_bytes"])
|
|
return tmp
|
|
|
|
def _make_conn(self, pub_db_id=42, fruit_db_ids=None):
|
|
"""Return a mock psycopg2 connection."""
|
|
conn = MagicMock()
|
|
cur = MagicMock()
|
|
conn.cursor.return_value.__enter__.return_value = cur
|
|
|
|
# fetchone side_effect: first call = pub id; subsequent = fruit lookups.
|
|
# None in fruit_db_ids → bare None (simulates no row found); int → (int,) tuple.
|
|
fruit_db_ids = fruit_db_ids or []
|
|
side_effects = [(pub_db_id,)] + [((fid,) if fid is not None else None) for fid in fruit_db_ids]
|
|
cur.fetchone.side_effect = side_effects
|
|
return conn, cur
|
|
|
|
def test_missing_cover_image_does_not_abort_import(self):
|
|
data_dir = self._make_data_dir([
|
|
{"id": "ber", "name": "Test", "img": "osdb/ber/ber_s0.jpg"},
|
|
])
|
|
# cover file NOT created → missing
|
|
conn, cur = self._make_conn(pub_db_id=1)
|
|
counts = import_publications(conn, data_dir)
|
|
self.assertEqual(counts["publications"], 1)
|
|
self.assertEqual(counts["covers"], 0)
|
|
|
|
def test_fruit_not_in_db_is_skipped_with_warning(self):
|
|
data_dir = self._make_data_dir([
|
|
{"id": "cal", "name": "Cal", "fruits": [
|
|
{"osdb_number": "unknown_fruit", "img": True},
|
|
]},
|
|
])
|
|
conn, cur = self._make_conn(pub_db_id=5, fruit_db_ids=[None])
|
|
counts = import_publications(conn, data_dir)
|
|
self.assertEqual(counts["skipped_fruits"], 1)
|
|
self.assertEqual(counts["fruits_linked"], 0)
|
|
|
|
def test_happy_path_upserts_pub_links_fruits_imports_files(self):
|
|
data_dir = self._make_data_dir([
|
|
{
|
|
"id": "pom",
|
|
"name": "Pomologie",
|
|
"author": "Diel",
|
|
"img": "osdb/pom/pom_s0.jpg",
|
|
"cover_bytes": b"\x89PNG",
|
|
"fruits": [
|
|
{"osdb_number": "apfelsorte_alpha", "img": True, "pdf": True},
|
|
],
|
|
}
|
|
])
|
|
conn, cur = self._make_conn(pub_db_id=7, fruit_db_ids=[99])
|
|
counts = import_publications(conn, data_dir)
|
|
self.assertEqual(counts["publications"], 1)
|
|
self.assertEqual(counts["covers"], 1)
|
|
self.assertEqual(counts["fruits_linked"], 1)
|
|
self.assertEqual(counts["pdfs"], 1)
|
|
self.assertEqual(counts["images"], 1)
|
|
self.assertEqual(counts["skipped_fruits"], 0)
|
|
|
|
def test_osw_not_1_never_imported(self):
|
|
tmp = tempfile.mkdtemp()
|
|
xml = b"""<?xml version="1.0" encoding="iso-8859-1"?>
|
|
<root>
|
|
<obj><id>skip</id><name>Skip</name><author>X</author><osw>0</osw></obj>
|
|
</root>"""
|
|
Path(tmp, "osws.xml").write_bytes(xml)
|
|
conn, cur = self._make_conn()
|
|
counts = import_publications(conn, tmp)
|
|
self.assertEqual(counts["publications"], 0)
|
|
cur.execute.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|