tests/test_financials_normalizer_public_equity_contract.py

"""Public-equity contract for financials-normalizer."""

from __future__ import annotations

import csv
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def read(relative_path: str) -> str:
    return (ROOT / relative_path).read_text(encoding="utf-8")


class FinancialsNormalizerPublicEquityContractTests(unittest.TestCase):
    def test_skill_reframes_to_public_equity_inputs(self) -> None:
        text = read("skills/financials-normalizer/SKILL.md")
        for phrase in [
            "public-company source financials",
            "consensus/guidance inputs",
            "share-count support",
            "net-debt and capital-allocation support",
            "ETF/index constituent support",
        ]:
            self.assertIn(phrase, text)
        self.assertNotIn("Classify job. Public " + "markets, Credit Markets", text)

    def test_integration_guide_routes_credit_security_work_out(self) -> None:
        text = read("skills/financials-normalizer/references/integration-guide.md")
        self.assertIn("Credit Markets handoff / equity-risk debt and liquidity context", text)
        self.assertIn("Route covenant packages, recovery waterfalls", text)
        self.assertNotIn("## Credit Markets and distressed", text)

    def test_schema_uses_equity_risk_debt_context_not_debt_schedule(self) -> None:
        text = read("skills/financials-normalizer/references/normalization-schema.md")
        self.assertIn("equity_risk_debt_liquidity_context", text)
        self.assertIn("Credit-security schedules", text)
        self.assertNotIn(
            "`statement`: `income_statement`, `balance_sheet`, `cash_flow`, `kpi_schedule`, `segment`, `debt_schedule`",
            text,
        )

    def test_standalone_model_ready_package_requires_comparability_and_validation(self) -> None:
        skill = read("skills/financials-normalizer/SKILL.md")
        schema = read("skills/financials-normalizer/references/normalization-schema.md")
        qa = read("skills/financials-normalizer/references/qa-rules.md")
        for phrase in [
            "model-loading package",
            "disclosure/comparability bridge",
            "Validation_Checks",
            "Do not infer fiscal period-end dates",
            "reserve `consensus_estimate`",
            "`comparable_rounded`",
        ]:
            self.assertIn(phrase, skill)
        for phrase in [
            "Standalone model-ready package",
            "Disclosure_Comparability_Bridge",
            "Validation_Checks",
            "company guidance as consensus",
            "`comparable_rounded`",
        ]:
            self.assertIn(phrase, schema)
        self.assertIn("Do not report a number of passed checks", qa)
        self.assertIn("changed cash-and-securities presentation", qa)

    def test_helper_enforces_fiscal_dates_guidance_and_source_typing(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_dir:
            tmp = Path(tmp_dir)
            input_csv = tmp / "financials.csv"
            fieldnames = [
                "entity",
                "source_id",
                "statement",
                "line_item_original",
                "line_item_standard",
                "line_item_id",
                "period_label",
                "source_value",
                "source_location",
                "evidence_label",
                "source_name",
                "source_type",
            ]
            rows = [
                {
                    "entity": "NVDA",
                    "source_id": "SRC-001",
                    "statement": "income_statement",
                    "line_item_original": "Revenue",
                    "line_item_standard": "Revenue",
                    "line_item_id": "revenue",
                    "period_label": "Q1 FY2027",
                    "source_value": "100",
                    "source_location": "p. 1",
                    "evidence_label": "fact_source_reported",
                },
                {
                    "entity": "NVDA",
                    "source_id": "SRC-001",
                    "statement": "consensus_estimate",
                    "line_item_original": "Revenue outlook",
                    "line_item_standard": "Revenue guidance midpoint",
                    "line_item_id": "guidance_revenue_midpoint",
                    "period_label": "Q2 FY2027 outlook",
                    "source_value": "110",
                    "source_location": "p. 2",
                    "evidence_label": "issuer_management_claim",
                },
                {
                    "entity": "NVDA",
                    "source_id": "SRC-001",
                    "statement": "consensus_estimate",
                    "line_item_original": "Revenue consensus",
                    "line_item_standard": "Revenue consensus",
                    "line_item_id": "consensus_revenue",
                    "period_label": "Q2 FY2027 estimate",
                    "source_value": "109",
                    "source_location": "consensus p. 1",
                },
                {
                    "entity": "NVDA",
                    "source_id": "SRC-001",
                    "statement": "debt_schedule",
                    "line_item_original": "Debt due after five years",
                    "line_item_standard": "Debt due after five years",
                    "line_item_id": "debt_due_after_five_years",
                    "period_label": "Q1 FY2027",
                    "source_value": "50",
                    "source_location": "p. 3",
                    "evidence_label": "fact_source_reported",
                },
                {
                    "entity": "NVDA",
                    "source_id": "SRC-002",
                    "statement": "kpi_schedule",
                    "line_item_original": "Provenance marker",
                    "line_item_standard": "Provenance marker",
                    "line_item_id": "provenance_marker",
                    "period_label": "Q1 FY2027",
                    "source_value": "1",
                    "source_location": "source_note.md",
                    "evidence_label": "fact_source_reported",
                    "source_type": "user_prompt",
                },
            ]
            with input_csv.open("w", newline="", encoding="utf-8") as handle:
                writer = csv.DictWriter(handle, fieldnames=fieldnames)
                writer.writeheader()
                writer.writerows(rows)
            out_dir = tmp / "out"
            script = (
                ROOT
                / "skills"
                / "financials-normalizer"
                / "scripts"
                / "normalize_extracted_financials.py"
            )
            result = subprocess.run(
                [sys.executable, str(script), str(input_csv), "--output-dir", str(out_dir)],
                cwd=str(ROOT),
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(0, result.returncode, result.stderr + result.stdout)
            with (out_dir / "Normalized_Financials_Long.csv").open(
                newline="", encoding="utf-8"
            ) as handle:
                normalized = list(csv.DictReader(handle))
            self.assertEqual("", normalized[0]["period_end"])
            self.assertEqual("forecast", normalized[1]["period_type"])
            self.assertEqual("kpi_schedule", normalized[1]["statement"])
            self.assertIn("separately from external consensus", normalized[1]["normalization_note"])
            self.assertEqual("consensus_estimate", normalized[2]["statement"])
            self.assertEqual("estimate_consensus", normalized[2]["evidence_label"])
            self.assertIn("verify external source", normalized[2]["normalization_note"])
            self.assertEqual("equity_risk_debt_liquidity_context", normalized[3]["statement"])
            self.assertIn("legacy debt_schedule migrated", normalized[3]["normalization_note"])
            with (out_dir / "Source_Index.csv").open(newline="", encoding="utf-8") as handle:
                sources = {row["source_id"]: row for row in csv.DictReader(handle)}
            self.assertEqual("uploaded_file", sources["SRC-002"]["source_type"])
            issues = (out_dir / "Normalization_Issues.csv").read_text(encoding="utf-8")
            self.assertIn("missing_period_end", issues)

            validator = (
                ROOT
                / "skills"
                / "financials-normalizer"
                / "scripts"
                / "validate_normalized_financials.py"
            )
            valid = subprocess.run(
                [sys.executable, str(validator), str(out_dir / "Normalized_Financials_Long.csv")],
                cwd=str(ROOT),
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(0, valid.returncode, valid.stderr + valid.stdout)

            normalized[1]["statement"] = "consensus_estimate"
            with (out_dir / "Normalized_Financials_Long.csv").open(
                "w", newline="", encoding="utf-8"
            ) as handle:
                writer = csv.DictWriter(handle, fieldnames=normalized[0].keys())
                writer.writeheader()
                writer.writerows(normalized)
            invalid = subprocess.run(
                [sys.executable, str(validator), str(out_dir / "Normalized_Financials_Long.csv")],
                cwd=str(ROOT),
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(1, invalid.returncode)
            self.assertIn("consensus_estimate is reserved", invalid.stdout)


if __name__ == "__main__":
    unittest.main()