Class TestApplyDecisions

tests/extraction/test_prefilter.py:306–365

class TestApplyDecisions(unittest.TestCase):
    def test_buckets_each_decision_type(self) -> None:
        decisions = [
            Work("/p/work.gz", "u/26.04/1/work.gz"),
            SizeSkip("/p/big.gz", "u/26.04/1/big.gz", 9999, 2048, "max"),
            AlreadyStored("/p/old.gz", "u/26.04/1/old.gz"),
            FilterSkip("/p/fs.gz", "u/26.04/1/fs.gz", "llm", "openai/other"),
            ContentDup("/p/dup.gz", "u/26.04/1/dup.gz", "u/26.04/1/canon.gz"),
        ]
        s = _FakeStore()
        out = apply_decisions(decisions, s, filter_db=("llm:openai/old",))  # type: ignore[arg-type]
        self.assertEqual(out.work_files, ["/p/work.gz"])
        self.assertEqual(
            out.content_dups, [("/p/dup.gz", "u/26.04/1/dup.gz", "u/26.04/1/canon.gz")]
        )
        self.assertEqual(out.size_filtered, 1)
        self.assertEqual(out.already_stored, 1)
        # prefilter_skipped includes size + already-stored + filter-skip.
        self.assertEqual(out.prefilter_skipped, 3)
        self.assertEqual(s.deleted, [])

    def test_stale_symlink_deletes_db_row(self) -> None:
        decisions = [
            Symlink(
                gz_path="/p/sym.gz",
                short_path="u/26.04/1/sym.gz",
                canonical_source="u/26.04/1/canon.gz",
                stale_in_db=True,
                canonical_in_inputs=False,
            ),
        ]
        s = _FakeStore(sources_in_db={"u/26.04/1/sym.gz"})
        out = apply_decisions(decisions, s, filter_db=())  # type: ignore[arg-type]
        self.assertEqual(s.deleted, ["u/26.04/1/sym.gz"])
        self.assertEqual(
            out.symlinks,
            [("/p/sym.gz", "u/26.04/1/sym.gz", "u/26.04/1/canon.gz")],
        )

    def test_fresh_symlink_no_delete(self) -> None:
        decisions = [
            Symlink(
                gz_path="/p/sym.gz",
                short_path="u/26.04/1/sym.gz",
                canonical_source="u/26.04/1/canon.gz",
                stale_in_db=False,
                canonical_in_inputs=True,
            ),
        ]
        s = _FakeStore()
        apply_decisions(decisions, s, filter_db=())  # type: ignore[arg-type]
        self.assertEqual(s.deleted, [])

    def test_classified_defaults(self) -> None:
        out = apply_decisions([], _FakeStore(), filter_db=())  # type: ignore[arg-type]
        self.assertIsInstance(out, Classified)
        self.assertEqual(out.work_files, [])
        self.assertEqual(out.symlinks, [])