""" Tests for journal_reference_db.py — the read-only SQLite accessor. Tests use the actual bundled journal_quality.db file for integration testing against real data. """ import pytest from local_deep_research.journal_quality.db import ( get_journal_reference_db, ) # The ref_db fixture (skip-if-missing pattern) lives in conftest.py and # is shared with test_db_accessors.py. # --------------------------------------------------------------------------- # lookup_source # --------------------------------------------------------------------------- class TestLookupSource: """Source lookup by ID, ISSN, and name.""" def test_lookup_by_name(self, ref_db): result = ref_db.lookup_source(name="Nature") assert result is not None assert result["name"] == "Nature" assert result["h_index"] > 1000 def test_lookup_by_name_case_insensitive(self, ref_db): result = ref_db.lookup_source(name="NATURE") assert result is not None def test_lookup_by_name_the_prefix_added(self, ref_db): """'Astrophysical Journal Letters' should find 'The Astrophysical Journal Letters'.""" result = ref_db.lookup_source(name="Astrophysical Journal Letters") if result: assert "astrophysical" in result["name"].lower() def test_lookup_by_name_the_prefix_stripped(self, ref_db): """'The Lancet' should also find 'Lancet' if stored without 'the'.""" result = ref_db.lookup_source(name="The Lancet") # Either found directly or via prefix stripping if result: assert "lancet" in result["name"].lower() def test_lookup_by_issn(self, ref_db): # Nature ISSN-L result = ref_db.lookup_source(issn="0028-0836") assert result is not None assert "nature" in result["name"].lower() def test_lookup_by_openalex_id(self, ref_db): # Look up a known journal by name first, get its ID nature = ref_db.lookup_source(name="Nature") if nature and nature.get("openalex_source_id"): result = ref_db.lookup_source( source_id=nature["openalex_source_id"] ) assert result is not None def test_lookup_not_found(self, ref_db): result = ref_db.lookup_source(name="ZZZ Nonexistent Journal 12345") assert result is None def test_lookup_none_args(self, ref_db): result = ref_db.lookup_source() assert result is None # --------------------------------------------------------------------------- # is_predatory # --------------------------------------------------------------------------- class TestIsPredatory: """Predatory check against the reference DB.""" def test_legitimate_journal_not_predatory(self, ref_db): is_pred, _ = ref_db.is_predatory(journal_name="Nature") assert is_pred is False def test_no_args_returns_false(self, ref_db): is_pred, _ = ref_db.is_predatory() assert is_pred is False # --------------------------------------------------------------------------- # is_whitelisted # --------------------------------------------------------------------------- class TestIsWhitelisted: """Whitelist check (DOAJ or high h-index).""" def test_high_h_index_is_whitelisted(self, ref_db): # Nature has h_index > 10 assert ref_db.is_whitelisted(name="Nature") is True def test_unknown_not_whitelisted(self, ref_db): assert ref_db.is_whitelisted(name="ZZZ Unknown 12345") is False # --------------------------------------------------------------------------- # Dashboard queries # --------------------------------------------------------------------------- class TestGetSummary: """Summary statistics.""" def test_returns_dict_with_expected_keys(self, ref_db): summary = ref_db.get_summary() assert "total" in summary assert summary["total"] > 200000 assert "avg_quality" in summary assert "predatory_count" in summary assert "doaj_count" in summary def test_avg_quality_in_range(self, ref_db): summary = ref_db.get_summary() assert 1 <= summary["avg_quality"] <= 10 class TestGetQualityDistribution: """Quality histogram.""" def test_returns_dict(self, ref_db): dist = ref_db.get_quality_distribution() assert isinstance(dist, dict) assert len(dist) > 0 def test_values_are_positive(self, ref_db): dist = ref_db.get_quality_distribution() for v in dist.values(): assert v > 0 class TestGetSourceDistribution: """Score source breakdown.""" def test_returns_dict(self, ref_db): dist = ref_db.get_source_distribution() assert isinstance(dist, dict) assert "openalex" in dist class TestGetJournalsPage: """Paginated journal list.""" def test_first_page(self, ref_db): journals, total = ref_db.get_journals_page(page=1, per_page=10) assert len(journals) == 10 assert total > 200000 def test_search_filter(self, ref_db): journals, total = ref_db.get_journals_page(search="nature", per_page=10) assert total > 0 assert all("nature" in j["name"].lower() for j in journals) def test_tier_filter(self, ref_db): journals, total = ref_db.get_journals_page(tier="elite", per_page=10) assert all(j["quality"] >= 9 for j in journals) def test_sort_by_h_index(self, ref_db): journals, _ = ref_db.get_journals_page( sort="h_index", order="desc", per_page=5 ) h_values = [j["h_index"] for j in journals if j["h_index"] is not None] assert h_values == sorted(h_values, reverse=True) def test_invalid_sort_column_defaults_to_quality(self, ref_db): # Should not crash journals, _ = ref_db.get_journals_page( sort="'; DROP TABLE sources; --", per_page=5 ) assert len(journals) == 5 def test_pagination_offset(self, ref_db): page1, _ = ref_db.get_journals_page(page=1, per_page=5) page2, _ = ref_db.get_journals_page(page=2, per_page=5) # Different pages should have different journals names1 = {j["name"] for j in page1} names2 = {j["name"] for j in page2} assert names1 != names2 class TestGetInstitutionsPage: """Tests for JournalQualityDB.get_institutions_page.""" def test_invalid_order_defaults_to_desc(self, ref_db): """A tainted ``order`` string must not crash or reach SQL. Regression guard for the allowlist added in fix(db): validate order param in get_institutions_page. The DB layer treats anything other than "asc" / "desc" as "desc", so the two calls below must return identical institution lists. """ bad, _ = ref_db.get_institutions_page( order="'; DROP TABLE institutions; --", per_page=5, ) desc, _ = ref_db.get_institutions_page( order="desc", per_page=5, ) assert [i["openalex_id"] for i in bad] == [ i["openalex_id"] for i in desc ] # --------------------------------------------------------------------------- # Build function # --------------------------------------------------------------------------- class TestBuildReferenceDb: """Test the build_reference_db function.""" def test_build_creates_db(self, tmp_path): """Build a DB into a temp directory.""" from local_deep_research.config.paths import ( get_journal_data_directory, ) from local_deep_research.journal_quality.db import build_db data_dir = get_journal_data_directory() if not (data_dir / "openalex_sources.json.gz").exists(): pytest.skip("OpenAlex data file not found") output = tmp_path / "test_quality.db" build_db(data_dir=data_dir, output_path=output) assert output.exists() assert output.stat().st_size > 1_000_000 # at least 1MB def test_built_db_is_queryable(self, tmp_path): """Built DB should be queryable via sqlite3.""" import sqlite3 from local_deep_research.config.paths import ( get_journal_data_directory, ) from local_deep_research.journal_quality.db import build_db data_dir = get_journal_data_directory() if not (data_dir / "openalex_sources.json.gz").exists(): pytest.skip("OpenAlex data file not found") output = tmp_path / "test_quality.db" build_db(data_dir=data_dir, output_path=output) # Open read-only via URI so chmod 0o444 doesn't matter conn = sqlite3.connect(f"file:{output}?mode=ro", uri=True) count = conn.execute("SELECT COUNT(*) FROM sources").fetchone()[0] assert count > 200000 conn.close() def test_built_db_indexes_score_source(self): """The dashboard's /api/journals endpoint filters by score_source via equality — the column needs an index or each filtered page does a full scan of the ~217K-row table. Asserted against an in-memory build so the test runs without the 350 MB OpenAlex snapshot. """ from sqlalchemy import create_engine, inspect from local_deep_research.journal_quality.models import ( JournalQualityBase, ) engine = create_engine("sqlite:///:memory:") try: JournalQualityBase.metadata.create_all(engine) indexed_cols = { col for idx in inspect(engine).get_indexes("sources") for col in idx["column_names"] } assert "score_source" in indexed_cols, ( "sources.score_source must be indexed — it's a dashboard " "filter predicate; without the index the query does a " "full-table scan of ~217K rows." ) finally: engine.dispose() def test_score_source_check_constraint_rejects_invalid(self): """CHECK constraint on ``sources.score_source`` must reject values outside the allowlist — defense-in-depth against any future writer (refactor, migration, import script) that accidentally bypasses the API layer's validation and tries to insert garbage. """ from sqlalchemy import create_engine, text from sqlalchemy.exc import IntegrityError from local_deep_research.journal_quality.models import ( JournalQualityBase, ) engine = create_engine("sqlite:///:memory:") # SQLite only enforces CHECK constraints when foreign-key/CHECK # handling is explicitly on for the connection. create_engine's # default SQLite connection does enforce CHECK — verify via a # round-trip. JournalQualityBase.metadata.create_all(engine) try: # Minimal columns needed for a valid row — the NOT NULL flags # come from the model, not this test. is_in_doaj / # is_predatory both default to False but the insert still has # to satisfy NOT NULL. cols = "name, name_lower, is_in_doaj, is_predatory, score_source" with engine.begin() as conn: # Happy path: known-valid value goes through. conn.execute( text( f"INSERT INTO sources ({cols}) " "VALUES ('Test Journal', 'test journal', " "0, 0, 'openalex')" ) ) with engine.begin() as conn: with pytest.raises( IntegrityError, match=r"(?i)CHECK|score_source" ): conn.execute( text( f"INSERT INTO sources ({cols}) " "VALUES ('Bad Journal', 'bad journal', " "0, 0, 'garbage')" ) ) finally: engine.dispose() def test_built_db_stamps_schema_version(self, tmp_path): """``build_db`` must stamp ``PRAGMA user_version`` to the current ``JOURNAL_QUALITY_SCHEMA_VERSION`` constant. The startup-time ``_validate_existing_db`` check reads this pragma to detect schema drift and force a rebuild — if the stamp goes missing silently, stale DBs survive forever past a schema change and users see mysterious query failures. Reads the constant dynamically rather than hardcoding a number so this test stays green across future version bumps without needing a mechanical update. """ import sqlite3 from local_deep_research.config.paths import ( get_journal_data_directory, ) from local_deep_research.journal_quality.db import ( JOURNAL_QUALITY_SCHEMA_VERSION, build_db, ) data_dir = get_journal_data_directory() if not (data_dir / "openalex_sources.json.gz").exists(): pytest.skip("OpenAlex data file not found") output = tmp_path / "test_quality.db" build_db(data_dir=data_dir, output_path=output) conn = sqlite3.connect(f"file:{output}?mode=ro", uri=True) try: stamped = conn.execute("PRAGMA user_version").fetchone()[0] finally: conn.close() assert stamped == JOURNAL_QUALITY_SCHEMA_VERSION, ( f"PRAGMA user_version={stamped}, expected " f"{JOURNAL_QUALITY_SCHEMA_VERSION}. If this fails, " "_validate_existing_db won't detect schema drift and " "stale DBs will never rebuild after the next bump." ) # --------------------------------------------------------------------------- # _populate_sources unit tests (hijacked + DOAJ-only second pass) # --------------------------------------------------------------------------- class TestPopulateSources: """Cover the two correctness fixes in _populate_sources directly, without requiring the 50 MB on-disk DB.""" def _build_in_memory(self, sources, doaj, pred): from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.db import _populate_sources from local_deep_research.journal_quality.models import ( JournalQualityBase, ) engine = create_engine("sqlite:///:memory:") JournalQualityBase.metadata.create_all(engine) Session = sessionmaker(bind=engine) with Session() as s: _populate_sources(s, sources, doaj, pred) s.commit() return engine def test_hijacked_journal_flagged(self): from sqlalchemy import select from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.models import Source sources = {"S1": {"n": "Fake Cloned Journal", "i": "1234-5678", "h": 5}} pred = { "journals": set(), "publishers": set(), "hijacked": {"fake cloned journal"}, "long_pubs": [], } engine = self._build_in_memory(sources, {}, pred) try: with sessionmaker(bind=engine)() as s: row = s.scalars( select(Source).where( Source.name_lower == "fake cloned journal" ) ).first() assert row is not None assert row.is_predatory is True assert row.predatory_source == "stop-predatory-hijacked" finally: engine.dispose() def test_doaj_only_journal_inserted(self): from sqlalchemy import select from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.models import Source doaj = { "9999-0001": { "name": "Some Small OA Journal", "publisher": "Small Press", } } pred = { "journals": set(), "publishers": set(), "hijacked": set(), "long_pubs": [], } engine = self._build_in_memory({}, doaj, pred) try: with sessionmaker(bind=engine)() as s: row = s.scalars( select(Source).where( Source.name_lower == "some small oa journal" ) ).first() assert row is not None assert row.score_source == "doaj" assert row.is_in_doaj is True assert row.openalex_source_id is None finally: engine.dispose() def test_doaj_crossref_flags_existing_openalex_source(self): """First-pass DOAJ cross-reference, both directions. Against a single non-empty DOAJ dump, an OpenAlex source whose normalized ISSN IS in the dump must be flagged ``is_in_doaj=True``, and one whose ISSN is NOT must stay ``is_in_doaj=False`` — both keeping ``score_source='openalex'`` (they came from OpenAlex; DOAJ only adds the flag). Testing both sides against the same dump pins the actual per-ISSN match: a regression that flagged every source whenever the dump is non-empty (``is_in_doaj = bool(doaj_data)``) passes a positive-only test but fails the negative assertion here. Partner to ``test_doaj_only_journal_inserted`` (second pass, DOAJ-only venues). Guards the ``doaj_data.get(issn)`` cross-ref at ``db.py::_populate_sources``. """ from sqlalchemy import select from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.models import Source from local_deep_research.utilities.citation_normalizer import ( normalize_issn, ) # S1's ISSN is hyphenated while the DOAJ dump is keyed by the # normalized (no-dash) form, so the match exercises normalize_issn on # both sides. S2 is a real OpenAlex source whose ISSN is absent from # the dump — the negative control. sources = { "S1": {"n": "PLoS ONE", "i": "1932-6203", "h": 200, "p": "PLOS"}, "S2": {"n": "Obscure Closed Journal", "i": "0000-0019", "h": 40}, } doaj = { normalize_issn("1932-6203"): { "name": "PLoS ONE", "publisher": "PLOS", } } pred = { "journals": set(), "publishers": set(), "hijacked": set(), "long_pubs": [], } engine = self._build_in_memory(sources, doaj, pred) try: with sessionmaker(bind=engine)() as s: hit = s.scalars( select(Source).where( Source.issn == normalize_issn("1932-6203") ) ).first() assert hit is not None assert hit.is_in_doaj is True # Came from OpenAlex, so the score source stays "openalex"; # DOAJ only contributes the is_in_doaj flag on this pass. assert hit.score_source == "openalex" assert hit.openalex_source_id == "S1" # Negative control: in OpenAlex, not in the (non-empty) dump. miss = s.scalars( select(Source).where( Source.issn == normalize_issn("0000-0019") ) ).first() assert miss is not None assert miss.is_in_doaj is False assert miss.score_source == "openalex" assert miss.openalex_source_id == "S2" finally: engine.dispose() def test_quartile_derivation_global_per_type(self): """Sources are bucketed Q1–Q4 by cited_by_count percentile within each source_type. Display-only signal — quality stays h-index-driven. """ from sqlalchemy import select from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.models import Source # 8 journals with cited_by_count 100..800. With 8 records the # percentile splits land at: ranks 6,7 → Q1, ranks 4,5 → Q2, # ranks 2,3 → Q3, ranks 0,1 → Q4. Plus a 9th journal with NULL # cited_by_count to confirm it gets NULL quartile. sources = { f"S{i}": { "n": f"Journal {i}", "i": f"0000-{i:04d}", "h": 5, "cb": (i + 1) * 100, } for i in range(8) } sources["S99"] = {"n": "Citationless Journal", "i": "9999-9999", "h": 5} pred = { "journals": set(), "publishers": set(), "hijacked": set(), "long_pubs": [], } engine = self._build_in_memory(sources, {}, pred) try: with sessionmaker(bind=engine)() as s: rows = s.scalars( select(Source).order_by(Source.cited_by_count) ).all() quartiled = [r for r in rows if r.cited_by_count is not None] assert len(quartiled) == 8 # Highest two should be Q1, lowest two should be Q4. assert quartiled[-1].quartile == "Q1" assert quartiled[-2].quartile == "Q1" assert quartiled[0].quartile == "Q4" assert quartiled[1].quartile == "Q4" # Every quartile bucket is represented. assert {r.quartile for r in quartiled} == { "Q1", "Q2", "Q3", "Q4", } # NULL cited_by_count → NULL quartile. citationless = s.scalars( select(Source).where(Source.name == "Citationless Journal") ).first() assert citationless is not None assert citationless.quartile is None # Quartile now feeds into quality (fixed in Round 5 review): # Q1 → 8 (STRONG), Q2 → 7, Q3 → 6, Q4 → 5. h_index=5 is below # the quartile-bump threshold so Q1 does not promote to 10. quartile_to_quality = {"Q1": 8, "Q2": 7, "Q3": 6, "Q4": 5} for r in quartiled: assert r.quality == quartile_to_quality[r.quartile], ( f"Journal with quartile={r.quartile} should score " f"{quartile_to_quality[r.quartile]} not {r.quality}" ) finally: engine.dispose() def test_quartile_separates_journals_from_conferences(self): """Journal pool and conference pool are percentile-binned independently — a conference shouldn't push a journal out of Q1. """ from sqlalchemy import select from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.models import Source sources = { # Single journal — alone in its pool, lands at the top → Q1 "J1": { "n": "Solo Journal", "t": "j", "i": "0001-0001", "h": 5, "cb": 10, }, # Single conference — alone in its pool, also Q1 "C1": { "n": "Solo Conference", "t": "c", "i": "0002-0002", "h": 5, "cb": 10000, }, } pred = { "journals": set(), "publishers": set(), "hijacked": set(), "long_pubs": [], } engine = self._build_in_memory(sources, {}, pred) try: with sessionmaker(bind=engine)() as s: rows = s.scalars(select(Source)).all() by_name = {r.name: r for r in rows} # Pool size 1 → percentile 0/1=0.0 → Q4 (lowest bucket). # The point of the test is that they're processed in # separate pools, not merged: the conference's huge cb does # NOT affect the journal's quartile. assert by_name["Solo Journal"].quartile is not None assert by_name["Solo Conference"].quartile is not None finally: engine.dispose() def test_print_and_electronic_issn_both_survive(self): from sqlalchemy import select from sqlalchemy.orm import sessionmaker from local_deep_research.journal_quality.models import Source sources = { "S1": {"n": "Journal of X", "i": "1111-1111", "h": 50}, "S2": {"n": "Journal of X", "i": "2222-2222", "h": 50}, } pred = { "journals": set(), "publishers": set(), "hijacked": set(), "long_pubs": [], } engine = self._build_in_memory(sources, {}, pred) try: with sessionmaker(bind=engine)() as s: rows = s.scalars( select(Source).where(Source.name_lower == "journal of x") ).all() # ISSNs are stored in canonical 8-char no-dash form (normalize_issn) issns = {r.issn for r in rows} assert issns == {"11111111", "22222222"} finally: engine.dispose() # --------------------------------------------------------------------------- # Singleton # --------------------------------------------------------------------------- class TestSingleton: """Singleton pattern.""" def test_returns_same_instance(self): import local_deep_research.journal_quality.db as mod mod._db = None a = get_journal_reference_db() b = get_journal_reference_db() assert a is b mod._db = None # --------------------------------------------------------------------------- # Read-only enforcement # --------------------------------------------------------------------------- class TestLookupSourcesBatch: """Batch name lookup used by the user-research dashboard endpoint.""" def test_empty_names_returns_empty_dict(self, ref_db): assert ref_db.lookup_sources_batch([]) == {} def test_none_and_blank_names_filtered(self, ref_db): # None and "" entries are dropped before the SQL query — the # result for an input that has nothing real is an empty dict, # not a SQL error. assert ref_db.lookup_sources_batch([None, "", " "]) == {} def test_returns_matches_keyed_by_name_lower(self, ref_db): result = ref_db.lookup_sources_batch(["Nature", "Science"]) # Both are near-certainly in the ref DB; the keys are the # normalized name_lower values, not the input casing. assert "nature" in result assert "science" in result assert result["nature"]["h_index"] > 1000 assert result["nature"]["quality"] is not None def test_unknown_name_simply_absent(self, ref_db): result = ref_db.lookup_sources_batch( ["Nature", "ZZZ-definitely-not-a-real-journal-XYZ"] ) assert "nature" in result assert "zzz-definitely-not-a-real-journal-xyz" not in result def test_deduplicates_input(self, ref_db): # Repeated names should not cause duplicate key errors or # multiply SQL parameter counts — the input is de-duplicated. result = ref_db.lookup_sources_batch(["Nature"] * 5 + ["Science"]) assert set(result.keys()) >= {"nature", "science"} class TestStaleDataVersionWarning: """_ensure_engine must log a WARNING when the on-disk version.json is behind the bundled JOURNAL_DATA_VERSION so admins see the mismatch in server logs without having to visit the dashboard. """ @pytest.fixture def loguru_sink(self): """Capture loguru log records — caplog doesn't see loguru output. The package's __init__ calls ``logger.disable("local_deep_research")`` to avoid interfering with downstream users' log setup, so the sink alone isn't enough — we also re-enable the package for the duration of the test. """ from loguru import logger as loguru_logger records = [] def _sink(message): records.append(message.record) # diagnose=False keeps the captured sink consistent with the # production policy (#4185 / #4384) — exceptions logged through this # fixture would otherwise carry frame-local repr() into recorded # messages and pytest output. handler_id = loguru_logger.add(_sink, level="WARNING", diagnose=False) loguru_logger.enable("local_deep_research") yield records loguru_logger.disable("local_deep_research") loguru_logger.remove(handler_id) def test_warn_on_stale_version(self, tmp_path, monkeypatch, loguru_sink): """A mismatched version.json triggers one WARNING per engine. The scenario we're guarding against: an older build is on disk, code has been upgraded to JOURNAL_DATA_VERSION="v4", but nothing has invalidated the cached sources. Without this warning, the filter silently serves stale scores until an admin manually visits /metrics/journals. """ import json from local_deep_research.journal_quality import db as db_mod from local_deep_research.journal_quality import downloader instance = db_mod.JournalQualityDB() version_file = tmp_path / "version.json" version_file.write_text(json.dumps({"version": "v3"})) monkeypatch.setattr(downloader, "JOURNAL_DATA_VERSION", "v4") instance._warn_on_stale_data_version(tmp_path) assert any( "stale" in r["message"] and "v3" in r["message"] for r in loguru_sink ) assert instance._stale_version_warned is True # Second call must not re-emit — one warning per engine lifetime. loguru_sink.clear() instance._warn_on_stale_data_version(tmp_path) assert not loguru_sink def test_no_warn_on_matching_version( self, tmp_path, monkeypatch, loguru_sink ): """When version.json matches JOURNAL_DATA_VERSION, stay silent.""" import json from local_deep_research.journal_quality import db as db_mod from local_deep_research.journal_quality import downloader instance = db_mod.JournalQualityDB() version_file = tmp_path / "version.json" version_file.write_text(json.dumps({"version": "v4"})) monkeypatch.setattr(downloader, "JOURNAL_DATA_VERSION", "v4") instance._warn_on_stale_data_version(tmp_path) assert not loguru_sink assert instance._stale_version_warned is False def test_no_warn_on_missing_version_file(self, tmp_path, loguru_sink): """Fresh install (no version.json yet) → silent; the dashboard banner handles first-run messaging, no server-log spam.""" from local_deep_research.journal_quality import db as db_mod instance = db_mod.JournalQualityDB() instance._warn_on_stale_data_version(tmp_path) assert not loguru_sink class TestReadOnlyEnforcement: """The runtime accessor must physically refuse writes.""" def test_write_attempt_raises_operational_error(self, ref_db): """sqlalchemy.exc.OperationalError on any write attempt. This is the safety net for the SQLite URI mode=ro flag — if someone removes it from `_ensure_engine`, this test catches it. """ from sqlalchemy.exc import OperationalError from local_deep_research.journal_quality.models import Source with ref_db.session() as s: s.add( Source( name="hack", name_lower="hack", score_source="test", ) ) with pytest.raises(OperationalError, match="readonly"): s.commit() s.rollback() # --------------------------------------------------------------------------- # Cleanup-path logging # --------------------------------------------------------------------------- class TestUnlinkUnusableDbLogs: """``_unlink_unusable_db`` does best-effort corruption-recovery cleanup. Previously swallowed OSError silently; now logs so a chmod/unlink failure surfaces in the ops log instead of quietly masking the real underlying problem (read-only mount, Windows file-in-use, permissions). """ def test_chmod_failure_logs_warning(self, tmp_path, loguru_caplog): from local_deep_research.journal_quality.db import JournalQualityDB dead_path = tmp_path / "nonexistent.db" # chmod will raise FileNotFound # 30 == WARNING in the stdlib level numbering that loguru_caplog # maps from (see tests/conftest.py::loguru_caplog fixture). with loguru_caplog.at_level(30): JournalQualityDB._unlink_unusable_db(dead_path) # Both chmod and unlink fail on a missing file — we expect a # warning mentioning the path, not silent success. assert any( "chmod" in rec.message.lower() and str(dead_path) in rec.message for rec in loguru_caplog.records ), "chmod failure on cleanup must be logged, not silenced" assert any( "unlink" in rec.message.lower() and str(dead_path) in rec.message for rec in loguru_caplog.records ), "unlink failure on cleanup must be logged, not silenced" def test_success_path_is_silent(self, tmp_path, loguru_caplog): """Happy path: both chmod and unlink succeed → nothing logged above WARNING level. Otherwise we'd pollute logs on every schema-drift rebuild. """ from local_deep_research.journal_quality.db import JournalQualityDB path = tmp_path / "doomed.db" path.write_text("x") path.chmod(0o444) # simulate the real post-build chmod with loguru_caplog.at_level(30): # WARNING JournalQualityDB._unlink_unusable_db(path) assert not path.exists() # No WARNING records for chmod / unlink in success case. assert not any( "unlink" in rec.message.lower() or "chmod" in rec.message.lower() for rec in loguru_caplog.records if rec.levelname == "WARNING" )