imacat/pop-fem-audit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Make every subcommand input an explicit CLI argument or option

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
imacat
2026-08-04 15:13:31 +08:00
co-authored by Claude Fable 5
parent e0eb343ce0
commit e9027472f4
12 changed files with 285 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/src/pop_fem_audit_tools/build_db.py
+35 -29
View File
@@ -5,8 +5,13 @@
"""The builder of the SQLite working store.
Rebuilds the working store from scratch out of the committed
inputs: the year-end chart CSV, the lyrics cache, the Wikidata
artist snapshot, and the manual artist overrides. Missing tables
inputs: the year-end chart CSV, given as the positional
command-line argument, and the optional capture inputs, each
given as an option: the lyrics cache directory, the Wikidata
artist snapshot CSV, and the manual artist overrides CSV. An
omitted option leaves its capture layer unloaded; a given
option whose path does not exist fails the build. Missing
tables
are created on a fresh store; existing tables are never altered,
as the schema lifecycle belongs to the migrations. Every rebuild
deletes all the rows, loads the data, and validates it in one
@@ -18,9 +23,6 @@ The rebuild is deterministic: the builder assigns the song and
artist IDs itself, as 1, 2, 3, ... in the first-occurrence file
order, so the IDs are reproducible across rebuilds on every
database engine, given the frozen input file.
Run from the repository root; the input paths are relative to the
current working directory.
"""
import argparse
import csv
@@ -43,14 +45,6 @@ from .models import (
SongArtist,
)
CHART_CSV: Path = Path("data/yearend_hot100_2016_2025.csv")
"""The year-end chart CSV file."""
LYRICS_DIR: Path = Path("data/lyrics")
"""The lyrics cache directory."""
WIKIDATA_CSV: Path = Path("data/artists_wikidata.csv")
"""The Wikidata artist snapshot CSV file."""
OVERRIDES_CSV: Path = Path("data/artists_overrides.csv")
"""The manual artist override CSV file."""
YEARS: Sequence[int] = range(2016, 2026)
"""The expected chart years."""
RANKS_PER_YEAR: int = 100
@@ -85,6 +79,18 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
parser: argparse.ArgumentParser = argparse.ArgumentParser(
description="Rebuild the SQLite working store from the"
" committed inputs.")
parser.add_argument(
"chart_csv", type=Path,
help="the year-end chart CSV file")
parser.add_argument(
"--lyrics-dir", type=Path, default=None,
help="the lyrics cache directory to load")
parser.add_argument(
"--wikidata-csv", type=Path, default=None,
help="the Wikidata artist snapshot CSV file to apply")
parser.add_argument(
"--overrides-csv", type=Path, default=None,
help="the manual artist override CSV file to apply")
return parser.parse_args(argv)
@@ -189,18 +195,15 @@ def load_chart(session: Session, path: Path) -> None:
def load_lyrics(session: Session, directory: Path) -> None:
"""Load the cached lyrics files into the matching songs.
A missing directory is skipped. A file whose stem is not an
existing song ID is skipped with a warning to the standard
error.
A file whose stem is not an existing song ID is skipped with
a warning to the standard error.
:param session: The database session, with the songs flushed.
:param directory: The lyrics cache directory with one
``<song_id>.txt`` file per song.
:param directory: The existing lyrics cache directory with
one ``<song_id>.txt`` file per song.
:return: None.
:raises OSError: When a lyrics file cannot be read.
"""
if not directory.is_dir():
return
for path in sorted(directory.glob("*.txt")):
song: Song | None = None
if path.stem.isdigit():
@@ -217,8 +220,7 @@ def apply_artist_csv(session: Session, path: Path) -> None:
Artists match by exact name. Only the non-empty cells are
applied, so a later CSV overrides an earlier one field by
field. The note column is ignored. A missing file is
skipped.
field. The note column is ignored.
:param session: The database session, with the artists
flushed.
@@ -228,8 +230,6 @@ def apply_artist_csv(session: Session, path: Path) -> None:
:raises BuildError: When a name matches no artist.
:raises OSError: When the file cannot be read.
"""
if not path.exists():
return
with open(path, encoding="utf-8", newline="") as file:
row: dict[str, str]
for row in csv.DictReader(file):
@@ -372,7 +372,7 @@ def main(argv: list[str] | None = None) -> int:
``sys.argv``.
:return: The exit status: 0 on success, non-zero on failure.
"""
parse_args(argv)
args: argparse.Namespace = parse_args(argv)
engine: sa.Engine = ds.engine
prepare_engine(engine)
Base.metadata.create_all(engine)
@@ -380,11 +380,17 @@ def main(argv: list[str] | None = None) -> int:
counts: StoreCounts
try:
reset_store(session)
load_chart(session, CHART_CSV)
load_chart(session, args.chart_csv)
session.flush()
load_lyrics(session, LYRICS_DIR)
apply_artist_csv(session, WIKIDATA_CSV)
apply_artist_csv(session, OVERRIDES_CSV)
if args.lyrics_dir is not None:
if not args.lyrics_dir.is_dir():
raise BuildError(
f"{args.lyrics_dir}: no such directory")
load_lyrics(session, args.lyrics_dir)
if args.wikidata_csv is not None:
apply_artist_csv(session, args.wikidata_csv)
if args.overrides_csv is not None:
apply_artist_csv(session, args.overrides_csv)
session.flush()
violations: list[str] = find_violations(
session, YEARS, RANKS_PER_YEAR)
tools/src/pop_fem_audit_tools/fetch_artists.py
+18 -18
View File
@@ -6,17 +6,14 @@
Fetches the metadata of the artists without a snapshot row from
Wikidata into the capture layer: the Wikidata artist snapshot
CSV. The working store is only read, never written; the
``build-db`` subcommand assembles the captured files into the
store on the next rebuild.
CSV, given as the positional command-line argument. The working
store is only read, never written; the ``build-db`` subcommand
assembles the captured files into the store on the next rebuild.
Every fetched row is meant for later human verification: the
description of the search hit is recorded in the note column so
that a bad match can be spotted. A search miss or an error on
one artist is noted on its row and does not fail the run.
Run from the repository root; the data paths are relative to the
current working directory.
"""
import argparse
import csv
@@ -37,8 +34,6 @@ from sqlalchemy.orm import Session
from .database import ds
from .models import Artist
WIKIDATA_CSV: Path = Path("data/artists_wikidata.csv")
"""The Wikidata artist snapshot CSV file."""
API_URL: str = "https://www.wikidata.org/w/api.php"
"""The URL of the Wikidata API endpoint."""
USER_AGENT: str = ("pop-fem-audit-tools"
@@ -130,6 +125,9 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
parser: argparse.ArgumentParser = argparse.ArgumentParser(
description="Fetch the artist metadata from Wikidata"
" into the capture layer.")
parser.add_argument(
"wikidata_csv", type=Path,
help="the Wikidata artist snapshot CSV file")
return parser.parse_args(argv)
@@ -341,34 +339,36 @@ class ArtistFetcher:
return json.load(response)
def read_snapshot_names() -> set[str]:
def read_snapshot_names(path: Path) -> set[str]:
"""Read the artist names already in the snapshot CSV file.
:param path: The Wikidata artist snapshot CSV file.
:return: The artist names, or an empty set when the file is
missing.
:raises OSError: When the file cannot be read.
"""
if not WIKIDATA_CSV.exists():
if not path.exists():
return set()
with open(WIKIDATA_CSV, encoding="utf-8",
with open(path, encoding="utf-8",
newline="") as file:
reader: csv.DictReader[str] = csv.DictReader(file)
return {x["name"] for x in reader}
def append_row(snapshot: ArtistSnapshot) -> None:
def append_row(path: Path, snapshot: ArtistSnapshot) -> None:
"""Append a snapshot row to the snapshot CSV file.
The CSV file is created with the header row when missing; the
existing rows are preserved.
:param path: The Wikidata artist snapshot CSV file.
:param snapshot: The snapshot of an artist.
:return: None.
:raises OSError: When the file cannot be written.
"""
is_new: bool = not WIKIDATA_CSV.exists()
WIKIDATA_CSV.parent.mkdir(parents=True, exist_ok=True)
with open(WIKIDATA_CSV, "a", encoding="utf-8",
is_new: bool = not path.exists()
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "a", encoding="utf-8",
newline="") as file:
writer: csv.DictWriter[str] = csv.DictWriter(
file, SNAPSHOT_FIELDS)
@@ -385,7 +385,7 @@ def main(argv: list[str] | None = None) -> int:
:return: The exit status: 0 on success, misses and errors
included, non-zero on a setup error.
"""
parse_args(argv)
args: argparse.Namespace = parse_args(argv)
fetcher: ArtistFetcher = ArtistFetcher()
fetched: int = 0
not_found: int = 0
@@ -393,7 +393,7 @@ def main(argv: list[str] | None = None) -> int:
skipped: int = 0
session: Session = ds.get_db()
try:
done: set[str] = read_snapshot_names()
done: set[str] = read_snapshot_names(args.wikidata_csv)
name: str
for name in session.scalars(
sa.select(Artist.name).order_by(Artist.id)):
@@ -401,7 +401,7 @@ def main(argv: list[str] | None = None) -> int:
skipped += 1
continue
snapshot: ArtistSnapshot = fetcher.fetch(name)
append_row(snapshot)
append_row(args.wikidata_csv, snapshot)
status: str = snapshot.qid
if snapshot.note == NOTE_NOT_FOUND:
not_found += 1
tools/src/pop_fem_audit_tools/fetch_lyrics.py
+38 -30
View File
@@ -6,17 +6,15 @@
Fetches the lyrics of the songs without a cache file from the
public lyrics APIs, Lyrics.ovh and LRCLIB, into the capture
layer: the lyrics cache directory and the provenance CSV. The
working store is only read, never written; the ``build-db``
subcommand assembles the captured files into the store on the
next rebuild.
layer: the lyrics cache directory and the provenance CSV, each
given as a positional command-line argument. The working store
is only read, never written; the ``build-db`` subcommand
assembles the captured files into the store on the next rebuild.
A song that every API misses is reported in the missing lyrics
CSV, which is rewritten on every run to reflect the current
status. Misses are expected and do not fail the run.
Run from the repository root; the data paths are relative to the
current working directory.
CSV, also given as a positional command-line argument, which is
rewritten on every run to reflect the current status. Misses
are expected and do not fail the run.
"""
import argparse
import csv
@@ -42,12 +40,6 @@ from .models import (
SongArtist,
)
LYRICS_DIR: Path = Path("data/lyrics")
"""The lyrics cache directory."""
PROVENANCE_CSV: Path = Path("data/lyrics_provenance.csv")
"""The lyrics provenance CSV file."""
MISSING_CSV: Path = Path("data/lyrics_missing.csv")
"""The missing lyrics report CSV file."""
PROVENANCE_FIELDS: Sequence[str] = (
"song_id", "source", "method", "acquired_at", "note")
"""The header columns of the lyrics provenance CSV file."""
@@ -95,6 +87,15 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
parser: argparse.ArgumentParser = argparse.ArgumentParser(
description="Fetch the missing song lyrics from the"
" public lyrics APIs into the capture layer.")
parser.add_argument(
"lyrics_dir", type=Path,
help="the lyrics cache directory")
parser.add_argument(
"provenance_csv", type=Path,
help="the lyrics provenance CSV file")
parser.add_argument(
"missing_csv", type=Path,
help="the missing lyrics report CSV file")
return parser.parse_args(argv)
@@ -203,34 +204,38 @@ def query_artist(session: Session, song_id: int) -> str:
return name
def save_lyrics(song_id: int, lyrics: str) -> None:
def save_lyrics(lyrics_dir: Path, song_id: int,
lyrics: str) -> None:
"""Write the lyrics of a song into the cache directory.
The cache directory is created when missing.
:param lyrics_dir: The lyrics cache directory.
:param song_id: The song ID.
:param lyrics: The lyrics text.
:return: None.
:raises OSError: When the file cannot be written.
"""
LYRICS_DIR.mkdir(parents=True, exist_ok=True)
(LYRICS_DIR / f"{song_id}.txt").write_text(
lyrics_dir.mkdir(parents=True, exist_ok=True)
(lyrics_dir / f"{song_id}.txt").write_text(
lyrics, encoding="utf-8")
def append_provenance(song_id: int, source: str) -> None:
def append_provenance(path: Path, song_id: int,
source: str) -> None:
"""Append a provenance row for a fetched lyrics file.
The CSV file is created with the header row when missing.
:param path: The lyrics provenance CSV file.
:param song_id: The song ID.
:param source: The source name of the fetched lyrics.
:return: None.
:raises OSError: When the file cannot be written.
"""
is_new: bool = not PROVENANCE_CSV.exists()
PROVENANCE_CSV.parent.mkdir(parents=True, exist_ok=True)
with open(PROVENANCE_CSV, "a", encoding="utf-8",
is_new: bool = not path.exists()
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "a", encoding="utf-8",
newline="") as file:
writer: Any = csv.writer(file)
if is_new:
@@ -239,19 +244,21 @@ def append_provenance(song_id: int, source: str) -> None:
datetime.date.today().isoformat(), ""])
def write_missing(misses: Sequence[MissingLyrics]) -> None:
def write_missing(path: Path,
misses: Sequence[MissingLyrics]) -> None:
"""Rewrite the missing lyrics report CSV file.
The previous content is replaced, so the file reflects the
current misses only.
:param path: The missing lyrics report CSV file.
:param misses: The report entries of the songs still without
lyrics.
:return: None.
:raises OSError: When the file cannot be written.
"""
MISSING_CSV.parent.mkdir(parents=True, exist_ok=True)
with open(MISSING_CSV, "w", encoding="utf-8",
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8",
newline="") as file:
writer: Any = csv.writer(file)
writer.writerow(MISSING_FIELDS)
@@ -266,7 +273,7 @@ def main(argv: list[str] | None = None) -> int:
:return: The exit status: 0 on success, misses included,
non-zero on a setup error.
"""
parse_args(argv)
args: argparse.Namespace = parse_args(argv)
fetcher: LyricsFetcher = LyricsFetcher()
fetched: int = 0
misses: list[MissingLyrics] = []
@@ -275,7 +282,7 @@ def main(argv: list[str] | None = None) -> int:
song: Song
for song in session.scalars(
sa.select(Song).order_by(Song.id)):
if (LYRICS_DIR / f"{song.id}.txt").exists():
if (args.lyrics_dir / f"{song.id}.txt").exists():
continue
artist: str = query_artist(session, song.id)
result: tuple[str, str] | None = fetcher.fetch(
@@ -291,12 +298,13 @@ def main(argv: list[str] | None = None) -> int:
lyrics: str
source: str
lyrics, source = result
save_lyrics(song.id, lyrics)
append_provenance(song.id, source)
save_lyrics(args.lyrics_dir, song.id, lyrics)
append_provenance(args.provenance_csv, song.id,
source)
fetched += 1
print(f"song {song.id} \"{song.title}\": {source}",
file=sys.stderr)
write_missing(misses)
write_missing(args.missing_csv, misses)
except (OSError, sa.exc.SQLAlchemyError) as error:
print(f"error: {error}", file=sys.stderr)
return 1
tools/src/pop_fem_audit_tools/run_llm.py
+8 -2
View File
@@ -9,7 +9,9 @@
Sends every input item to the Anthropic Messages Batch API twice with
the same system prompt, reconciles the disagreeing items with a third
arbitration batch, and archives every artifact self-contained under
``runs/<phase>/<YYYYMMDD-HHMM>-<prompt-stem>/``.
``<runs_dir>/<phase>/<YYYYMMDD-HHMM>-<prompt-stem>/``, where the
base directory of the run archives is given as the positional
command-line argument.
"""
import argparse
import enum
@@ -173,6 +175,9 @@ def parse_args(argv: list[str] | None) -> argparse.Namespace:
"""
parser: argparse.ArgumentParser = argparse.ArgumentParser(
description="Run one LLM step: 2 runs + 1 arbitration.")
parser.add_argument(
"runs_dir", type=Path,
help="the base directory of the run archives")
parser.add_argument(
"--prompt", required=True, type=Path,
help="the prompt definition file, used as the system prompt")
@@ -567,7 +572,8 @@ def main(argv: list[str] | None = None) -> int:
return 1
try:
run_dir: Path = create_archive_dir(
Path("runs"), args.phase, args.prompt, datetime.now())
args.runs_dir, args.phase, args.prompt,
datetime.now())
except FileExistsError as error:
print(f"error: {error}", file=sys.stderr)
return 1
tools/tests/test_build_db.py
+85 -22
View File
@@ -4,7 +4,6 @@
# imacat@mail.imacat.idv.tw (imacat), 2026/7/31
"""Unit tests for the working store builder module."""
import io
import os
import tempfile
import unittest
from contextlib import redirect_stderr
@@ -104,15 +103,17 @@ class TestBuildDB(unittest.TestCase):
"""The default chart CSV fixture: 2 years with 2 ranks each."""
def setUp(self) -> None:
"""Create a temporary working directory with the fixtures."""
"""Create a temporary data directory with the fixtures."""
tmp: tempfile.TemporaryDirectory[str] \
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.__dir: Path = Path(tmp.name)
old_cwd: str = os.getcwd()
self.addCleanup(os.chdir, old_cwd)
os.chdir(self.__dir)
Path("data").mkdir()
self.__chart: Path = self.__dir / "chart.csv"
self.__lyrics: Path = self.__dir / "lyrics"
self.__wikidata: Path = \
self.__dir / "artists_wikidata.csv"
self.__overrides: Path = \
self.__dir / "artists_overrides.csv"
self.__write_chart(self.CHART_CSV)
url: str = f"sqlite:///{self.__dir}/store.sqlite3"
config.set_settings(config.Settings(
@@ -127,26 +128,25 @@ class TestBuildDB(unittest.TestCase):
patcher.start()
self.addCleanup(patcher.stop)
@staticmethod
def __write_chart(content: str) -> None:
def __write_chart(self, content: str) -> None:
"""Write the chart CSV fixture.
:param content: The CSV content.
:return: None.
"""
Path("data/yearend_hot100_2016_2025.csv").write_text(
content, encoding="utf-8")
self.__chart.write_text(content, encoding="utf-8")
@staticmethod
def __run_build() -> tuple[int, str]:
def __run_build(self, *options: str) -> tuple[int, str]:
"""Run the build with the standard error captured.
:param options: The additional command-line options.
:return: A tuple of the exit status and the standard
error.
"""
stderr: io.StringIO = io.StringIO()
with redirect_stderr(stderr):
status: int = build_db.main([])
status: int = build_db.main(
[str(self.__chart), *options])
return status, stderr.getvalue()
def __session(self) -> Session:
@@ -244,15 +244,17 @@ class TestBuildDB(unittest.TestCase):
def test_overrides_apply_over_wikidata(self) -> None:
"""Test that the overrides win over the Wikidata snapshot."""
Path("data/artists_wikidata.csv").write_text(
self.__wikidata.write_text(
"name,qid,gender,type,genre,country,note\n"
"Adele,Q2831,female,solo,pop,GB,\n",
encoding="utf-8")
Path("data/artists_overrides.csv").write_text(
self.__overrides.write_text(
"name,qid,gender,type,genre,country,note\n"
"Adele,,,,soul,,manually checked\n",
encoding="utf-8")
self.assertEqual(self.__run_build()[0], 0)
self.assertEqual(self.__run_build(
"--wikidata-csv", str(self.__wikidata),
"--overrides-csv", str(self.__overrides))[0], 0)
session: Session = self.__session()
artist: Artist | None = session.scalar(
sa.select(Artist).where(Artist.name == "Adele"))
@@ -264,12 +266,13 @@ class TestBuildDB(unittest.TestCase):
def test_unknown_override_name_fails(self) -> None:
"""Test that an unknown override name fails the build."""
Path("data/artists_overrides.csv").write_text(
self.__overrides.write_text(
"name,qid,gender,type,genre,country,note\n"
"Adel,,female,,,,typo\n", encoding="utf-8")
status: int
stderr: str
status, stderr = self.__run_build()
status, stderr = self.__run_build(
"--overrides-csv", str(self.__overrides))
self.assertNotEqual(status, 0)
self.assertIn("Adel", stderr)
session: Session = self.__session()
@@ -277,14 +280,15 @@ class TestBuildDB(unittest.TestCase):
def test_lyrics_loaded(self) -> None:
"""Test loading the lyrics cache into the songs."""
Path("data/lyrics").mkdir()
Path("data/lyrics/1.txt").write_text(
self.__lyrics.mkdir()
(self.__lyrics / "1.txt").write_text(
"Hello, it's me\n", encoding="utf-8")
Path("data/lyrics/999.txt").write_text(
(self.__lyrics / "999.txt").write_text(
"orphan\n", encoding="utf-8")
status: int
stderr: str
status, stderr = self.__run_build()
status, stderr = self.__run_build(
"--lyrics-dir", str(self.__lyrics))
self.assertEqual(status, 0)
self.assertIn("999", stderr)
self.assertIn("1 songs with lyrics", stderr)
@@ -293,3 +297,62 @@ class TestBuildDB(unittest.TestCase):
assert song is not None
self.assertEqual(song.title, "Hello")
self.assertEqual(song.lyrics, "Hello, it's me\n")
def test_omitted_options_skip_capture_layers(self) -> None:
"""Test that omitted options leave the layers unloaded."""
self.__lyrics.mkdir()
(self.__lyrics / "1.txt").write_text(
"Hello, it's me\n", encoding="utf-8")
self.__wikidata.write_text(
"name,qid,gender,type,genre,country,note\n"
"Adele,Q2831,female,solo,pop,GB,\n",
encoding="utf-8")
status: int
stderr: str
status, stderr = self.__run_build()
self.assertEqual(status, 0)
self.assertIn("0 songs with lyrics", stderr)
session: Session = self.__session()
song: Song | None = session.get(Song, 1)
assert song is not None
self.assertIsNone(song.lyrics)
artist: Artist | None = session.scalar(
sa.select(Artist).where(Artist.name == "Adele"))
assert artist is not None
self.assertIsNone(artist.wikidata_qid)
self.assertIsNone(artist.gender)
def test_missing_lyrics_dir_fails(self) -> None:
"""Test that a given but missing lyrics directory fails."""
status: int
stderr: str
status, stderr = self.__run_build(
"--lyrics-dir", str(self.__lyrics))
self.assertNotEqual(status, 0)
self.assertIn(f"error: {self.__lyrics}", stderr)
session: Session = self.__session()
self.assertEqual(list(session.scalars(sa.select(Song))), [])
def test_missing_wikidata_csv_fails(self) -> None:
"""Test that a given but missing snapshot CSV fails."""
status: int
stderr: str
status, stderr = self.__run_build(
"--wikidata-csv", str(self.__wikidata))
self.assertNotEqual(status, 0)
self.assertIn("error:", stderr)
self.assertIn(str(self.__wikidata), stderr)
session: Session = self.__session()
self.assertEqual(list(session.scalars(sa.select(Song))), [])
def test_missing_overrides_csv_fails(self) -> None:
"""Test that a given but missing override CSV fails."""
status: int
stderr: str
status, stderr = self.__run_build(
"--overrides-csv", str(self.__overrides))
self.assertNotEqual(status, 0)
self.assertIn("error:", stderr)
self.assertIn(str(self.__overrides), stderr)
session: Session = self.__session()
self.assertEqual(list(session.scalars(sa.select(Song))), [])
tools/tests/test_fetch_artists.py
+11 -14
View File
@@ -6,7 +6,6 @@
import csv
import io
import json
import os
import tempfile
import unittest
import urllib.error
@@ -31,15 +30,13 @@ class TestFetchArtists(unittest.TestCase):
"""The expected header row of the snapshot CSV file."""
def setUp(self) -> None:
"""Create a temporary working directory with the store."""
"""Create a temporary capture directory with the store."""
tmp: tempfile.TemporaryDirectory[str] \
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.__dir: Path = Path(tmp.name)
old_cwd: str = os.getcwd()
self.addCleanup(os.chdir, old_cwd)
os.chdir(self.__dir)
Path("data").mkdir()
self.__snapshot: Path = \
self.__dir / "artists_wikidata.csv"
url: str = f"sqlite:///{self.__dir}/store.sqlite3"
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL=url,
@@ -116,8 +113,7 @@ class TestFetchArtists(unittest.TestCase):
x: {"labels": {"en": {"value": y}}}
for x, y in labels.items()}}
@staticmethod
def __run_fetch() -> tuple[int, str]:
def __run_fetch(self) -> tuple[int, str]:
"""Run the fetcher with the standard error captured.
:return: A tuple of the exit status and the standard
@@ -125,7 +121,8 @@ class TestFetchArtists(unittest.TestCase):
"""
stderr: io.StringIO = io.StringIO()
with redirect_stderr(stderr):
status: int = fetch_artists.main([])
status: int = fetch_artists.main(
[str(self.__snapshot)])
return status, stderr.getvalue()
@staticmethod
@@ -183,7 +180,7 @@ class TestFetchArtists(unittest.TestCase):
"?action=wbgetentities&ids=Q2%7CQ5%7CQ3%7CQ4%7CQ6"
"&props=labels&languages=en&format=json")
rows: list[list[str]] = self.__read_rows(
Path("data/artists_wikidata.csv"))
self.__snapshot)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0], self.HEADER)
self.assertEqual(rows[1], [
@@ -214,7 +211,7 @@ class TestFetchArtists(unittest.TestCase):
status: int = self.__run_fetch()[0]
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_rows(
Path("data/artists_wikidata.csv"))
self.__snapshot)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[1], [
"BTS", "Q10", "", "group", "K-pop", "South Korea",
@@ -234,7 +231,7 @@ class TestFetchArtists(unittest.TestCase):
self.assertEqual(status, 0)
self.assertEqual(urlopen.call_count, 1)
rows: list[list[str]] = self.__read_rows(
Path("data/artists_wikidata.csv"))
self.__snapshot)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0], self.HEADER)
self.assertEqual(rows[1], [
@@ -255,7 +252,7 @@ class TestFetchArtists(unittest.TestCase):
status, stderr = self.__run_fetch()
self.assertEqual(status, 0)
rows: list[list[str]] = self.__read_rows(
Path("data/artists_wikidata.csv"))
self.__snapshot)
self.assertEqual(len(rows), 3)
self.assertEqual(rows[1][:2], ["Broken", ""])
self.assertTrue(rows[1][6].startswith("error: "))
@@ -268,7 +265,7 @@ class TestFetchArtists(unittest.TestCase):
def test_rerun_skips_existing(self) -> None:
"""Test that the snapshot rows are skipped and preserved."""
self.__seed(["Adele", "Nobody"])
snapshot: Path = Path("data/artists_wikidata.csv")
snapshot: Path = self.__snapshot
old_row: list[str] = [
"Adele", "Q1", "female", "solo", "pop",
"United Kingdom", "English singer"]
tools/tests/test_fetch_lyrics.py
+27 -27
View File
@@ -6,7 +6,6 @@
import csv
import io
import json
import os
import tempfile
import unittest
import urllib.error
@@ -38,15 +37,15 @@ class TestFetchLyrics(unittest.TestCase):
"""The expected header row of the missing report CSV file."""
def setUp(self) -> None:
"""Create a temporary working directory with the store."""
"""Create a temporary capture directory with the store."""
tmp: tempfile.TemporaryDirectory[str] \
= tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.__dir: Path = Path(tmp.name)
old_cwd: str = os.getcwd()
self.addCleanup(os.chdir, old_cwd)
os.chdir(self.__dir)
Path("data").mkdir()
self.__lyrics: Path = self.__dir / "lyrics"
self.__provenance: Path = \
self.__dir / "lyrics_provenance.csv"
self.__missing: Path = self.__dir / "lyrics_missing.csv"
url: str = f"sqlite:///{self.__dir}/store.sqlite3"
config.set_settings(config.Settings(
SQLALCHEMY_DATABASE_URL=url,
@@ -110,8 +109,7 @@ class TestFetchLyrics(unittest.TestCase):
return urllib.error.HTTPError(
"https://example.com/", 404, "Not Found", None, None)
@staticmethod
def __run_fetch() -> tuple[int, str]:
def __run_fetch(self) -> tuple[int, str]:
"""Run the fetcher with the standard error captured.
:return: A tuple of the exit status and the standard
@@ -119,7 +117,9 @@ class TestFetchLyrics(unittest.TestCase):
"""
stderr: io.StringIO = io.StringIO()
with redirect_stderr(stderr):
status: int = fetch_lyrics.main([])
status: int = fetch_lyrics.main(
[str(self.__lyrics), str(self.__provenance),
str(self.__missing)])
return status, stderr.getvalue()
@staticmethod
@@ -149,10 +149,11 @@ class TestFetchLyrics(unittest.TestCase):
self.assertEqual(request.get_header("User-agent"),
fetch_lyrics.USER_AGENT)
self.assertEqual(
Path("data/lyrics/1.txt").read_text(encoding="utf-8"),
(self.__lyrics / "1.txt")
.read_text(encoding="utf-8"),
"Hello, it's me\n")
rows: list[list[str]] = self.__read_rows(
Path("data/lyrics_provenance.csv"))
self.__provenance)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0], self.PROVENANCE_HEADER)
self.assertEqual(rows[1][:3],
@@ -175,10 +176,11 @@ class TestFetchLyrics(unittest.TestCase):
self.assertEqual(status, 0)
self.assertEqual(urlopen.call_count, 2)
self.assertEqual(
Path("data/lyrics/1.txt").read_text(encoding="utf-8"),
(self.__lyrics / "1.txt")
.read_text(encoding="utf-8"),
"Hello\n")
rows: list[list[str]] = self.__read_rows(
Path("data/lyrics_provenance.csv"))
self.__provenance)
self.assertEqual(rows[1][:2], ["1", "lrclib"])
def test_both_miss(self) -> None:
@@ -192,11 +194,9 @@ class TestFetchLyrics(unittest.TestCase):
stderr: str
status, stderr = self.__run_fetch()
self.assertEqual(status, 0)
self.assertFalse(Path("data/lyrics/1.txt").exists())
self.assertFalse(
Path("data/lyrics_provenance.csv").exists())
rows: list[list[str]] = self.__read_rows(
Path("data/lyrics_missing.csv"))
self.assertFalse((self.__lyrics / "1.txt").exists())
self.assertFalse(self.__provenance.exists())
rows: list[list[str]] = self.__read_rows(self.__missing)
self.assertEqual(len(rows), 2)
self.assertEqual(rows[0], self.MISSING_HEADER)
self.assertEqual(rows[1][:3], ["1", "Hello", "Adele"])
@@ -205,8 +205,8 @@ class TestFetchLyrics(unittest.TestCase):
def test_cached_song_skipped(self) -> None:
"""Test that a cached song triggers no HTTP request."""
self.__seed([("Hello", "Adele")])
Path("data/lyrics").mkdir()
Path("data/lyrics/1.txt").write_text(
self.__lyrics.mkdir()
(self.__lyrics / "1.txt").write_text(
"cached\n", encoding="utf-8")
urlopen: mock.Mock
with mock.patch("urllib.request.urlopen") as urlopen:
@@ -214,10 +214,10 @@ class TestFetchLyrics(unittest.TestCase):
self.assertEqual(status, 0)
urlopen.assert_not_called()
self.assertEqual(
Path("data/lyrics/1.txt").read_text(encoding="utf-8"),
(self.__lyrics / "1.txt")
.read_text(encoding="utf-8"),
"cached\n")
rows: list[list[str]] = self.__read_rows(
Path("data/lyrics_missing.csv"))
rows: list[list[str]] = self.__read_rows(self.__missing)
self.assertEqual(rows, [self.MISSING_HEADER])
def test_url_encoding(self) -> None:
@@ -252,17 +252,17 @@ class TestFetchLyrics(unittest.TestCase):
self.__response({"lyrics": "one\n"}),
self.__response({"lyrics": "two\n"})]):
self.assertEqual(self.__run_fetch()[0], 0)
provenance: Path = Path("data/lyrics_provenance.csv")
rows: list[list[str]] = self.__read_rows(provenance)
rows: list[list[str]] = self.__read_rows(
self.__provenance)
self.assertEqual(len(rows), 3)
self.assertEqual(rows[0], self.PROVENANCE_HEADER)
Path("data/lyrics/2.txt").unlink()
(self.__lyrics / "2.txt").unlink()
with mock.patch(
"urllib.request.urlopen",
side_effect=[
self.__response({"lyrics": "two again\n"})]):
self.assertEqual(self.__run_fetch()[0], 0)
rows = self.__read_rows(provenance)
rows = self.__read_rows(self.__provenance)
self.assertEqual(len(rows), 4)
self.assertEqual(rows[0], self.PROVENANCE_HEADER)
self.assertNotIn(self.PROVENANCE_HEADER, rows[1:])
tools/tests/test_run_llm.py
+16 -17
View File
@@ -6,7 +6,6 @@
"""Unit tests for the run_llm batch runner module."""
import io
import json
import os
import tempfile
import unittest
from contextlib import redirect_stderr, redirect_stdout
@@ -310,24 +309,24 @@ class TestMainFlow(RunLLMTestCase):
"""Test cases for the end-to-end main flow."""
def setUp(self) -> None:
"""Create a temporary working directory with input files."""
"""Create a temporary directory with the input files."""
directory: Path = self._make_temp_dir()
old_cwd: str = os.getcwd()
self.addCleanup(os.chdir, old_cwd)
os.chdir(directory)
Path("prompts").mkdir()
Path("prompts/task_v1.md").write_text(
"The task prompt.\n", encoding="utf-8")
Path("prompts/task_arbitration_v1.md").write_text(
self.__runs: Path = directory / "runs"
prompt: Path = directory / "task_v1.md"
prompt.write_text("The task prompt.\n", encoding="utf-8")
arbitration: Path = directory / "task_arbitration_v1.md"
arbitration.write_text(
"The arbitration prompt.\n", encoding="utf-8")
Path("items.jsonl").write_text(
self.__input: Path = directory / "items.jsonl"
self.__input.write_text(
'{"id": "a", "content": "first item"}\n'
'{"id": "b", "content": "second item"}\n',
encoding="utf-8")
self.__argv: list[str] = [
"--prompt", "prompts/task_v1.md",
"--arbitration-prompt", "prompts/task_arbitration_v1.md",
"--input", "items.jsonl",
str(self.__runs),
"--prompt", str(prompt),
"--arbitration-prompt", str(arbitration),
"--input", str(self.__input),
"--phase", "coding"]
self.__settings: config.Settings = config.Settings(
SQLALCHEMY_DATABASE_URL="sqlite://",
@@ -385,7 +384,8 @@ class TestMainFlow(RunLLMTestCase):
:return: The archive directory.
"""
directories: list[Path] = list(Path("runs/coding").iterdir())
directories: list[Path] = list(
(self.__runs / "coding").iterdir())
self.assertEqual(len(directories), 1)
return directories[0]
@@ -509,9 +509,8 @@ class TestMainFlow(RunLLMTestCase):
def test_invalid_input_exits_non_zero(self) -> None:
"""Test that an invalid input file aborts before archiving."""
Path("items.jsonl").write_text(
'{"id": "a"}\n', encoding="utf-8")
self.__input.write_text('{"id": "a"}\n', encoding="utf-8")
status: int = self.__run_main(
self.__argv + ["--dry-run"])[0]
self.assertEqual(status, 1)
self.assertFalse(Path("runs").exists())
self.assertFalse(self.__runs.exists())