platform-ci/tests/test_cli_integration.py

"""CLI integration: verify judges existing JUnit reports without rerunning, scan writes a redacted manifest, report needs all three jobs.

No test starts an application build, a scanner or the network: the native runner and the scan runner are replaced.
"""

import json
import shutil
import tempfile
import unittest
from functools import partial
from pathlib import Path
from typing import override
from unittest import mock

from platform_ci.__main__ import main
from platform_ci.packaging import manifest_digest
from platform_ci.plan import Target, to_compact_json
from platform_ci.secret_scan import scan_history
from tests.support import PLATFORM_REVISION, SOURCE_REVISION, SOURCE_TREE, event, example_configuration, first_target, plan_for
from tests.test_secret_scan import FakeGit

PASSING = '<testsuite tests="2" skipped="1" failures="0" errors="0"><testcase name="a"/></testsuite>'
SENTINEL = "SENTINEL-SECRET-BODY"
FINDING = json.dumps([{"RuleID": "generic", "File": "a.txt", "Secret": SENTINEL}])


class CliTestCase(unittest.TestCase):
   """A temporary source checkout and a maven target that declares optional JUnit and required file reports."""

   @override
   def setUp(self) -> None:
      """Create the checkout."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.root = Path(directory.name).resolve()
      self.source = self.root / "source"
      self.source.mkdir()
      self.target: Target = first_target("maven-reports.ci.yaml")

   def write(self, relative: str, content: str) -> None:
      """Write a file under the source checkout."""
      path = self.source / relative
      path.parent.mkdir(parents=True, exist_ok=True)
      path.write_text(content, encoding="utf-8")


class VerifyCommandTest(CliTestCase):
   """Native commands run once; afterwards each declared JUnit report is judged as data."""

   def _verify(self, native_succeeds: bool = True) -> tuple[int, mock.Mock]:
      runner = mock.Mock(return_value=native_succeeds)
      environment = {"CI_TARGET": to_compact_json(self.target)}
      with mock.patch("platform_ci.__main__.run_target_commands", runner):
         code = main(["verify", "--source", str(self.source)], environment)
      return code, runner

   def test_passing_and_empty_reports_pass_after_one_native_run(self) -> None:
      """Zero reports, zero tests and all-skipped stay allowed, as in the legacy gate."""
      self.write("target/surefire-reports/TEST-a.xml", PASSING)
      self.write("target/surefire-reports/TEST-b.xml", '<testsuite tests="1" skipped="1"><testcase name="b"><skipped/></testcase></testsuite>')
      code, runner = self._verify()
      self.assertEqual(code, 0)
      runner.assert_called_once()

   def test_no_matching_report_is_not_judged_by_verify(self) -> None:
      """Whether a required report exists is packaging's decision, so verify passes with none."""
      self.assertEqual(self._verify()[0], 0)

   def test_failing_malformed_and_declaring_reports_fail_the_build_result(self) -> None:
      """Failures, errors, malformed XML and DTDs fail without a second native run."""
      bad_reports = {
         "failure": '<testsuite tests="1" failures="1"><testcase name="a"><failure/></testcase></testsuite>',
         "error": '<testsuite tests="1" errors="1"/>',
         "malformed": "<testsuite",
         "doctype": '<!DOCTYPE testsuite [<!ENTITY x "y">]><testsuite tests="0"/>',
      }
      for name, content in bad_reports.items():
         with self.subTest(report=name):
            self.write("target/surefire-reports/TEST-a.xml", PASSING)
            self.write("target/failsafe-reports/TEST-bad.xml", content)
            code, runner = self._verify()
            self.assertEqual(code, 1)
            runner.assert_called_once()

   def test_failed_native_run_fails_without_reading_reports(self) -> None:
      """A failed build result is final; reports are left for packaging to retain."""
      self.write("target/surefire-reports/TEST-a.xml", PASSING)
      with mock.patch("platform_ci.__main__.check_junit_reports") as judged:
         self.assertEqual(self._verify(native_succeeds=False)[0], 1)
      judged.assert_not_called()

   def test_file_reports_and_excluded_directories_are_not_judged(self) -> None:
      """Only `junit` reports are parsed, and matches inside `.git` or `.platform-ci` are skipped like packaging does."""
      self.write("target/site/jacoco/jacoco.xml", "not xml at all <")
      self.write(".platform-ci/target/surefire-reports/TEST-x.xml", "<broken")
      self.write(".git/target/surefire-reports/TEST-y.xml", "<broken")
      self.assertEqual(self._verify()[0], 0)

   def test_a_report_symlinked_outside_the_workspace_fails(self) -> None:
      """A link to a file outside the checkout is an error, never silently read or skipped."""
      outside = self.root / "outside.xml"
      outside.write_text(PASSING, encoding="utf-8")
      link = self.source / "target" / "surefire-reports" / "TEST-link.xml"
      link.parent.mkdir(parents=True)
      try:
         link.symlink_to(outside)
      except OSError:
         self.skipTest("symbolic links are unavailable")
      self.assertEqual(self._verify()[0], 1)

   def test_missing_required_output_still_fails_packaging(self) -> None:
      """A required file report that was never produced fails packaging after a successful build."""
      environment = {
         "CI_TARGET": to_compact_json(self.target),
         "CI_BUILD_OUTCOME": "success",
         "CI_SOURCE_REVISION": SOURCE_REVISION,
         "CI_SOURCE_TREE": SOURCE_TREE,
         "CI_PLATFORM_REVISION": PLATFORM_REVISION,
         "CI_RUN_ID": "1",
         "CI_RUN_ATTEMPT": "1",
      }
      with self.assertLogs("platform_ci", level="ERROR"):
         self.assertEqual(main(["package", "--source", str(self.source), "--output", str(self.root / "out")], environment), 1)


class ScanCommandTest(CliTestCase):
   """The scan command writes a source, platform and run bound manifest that never contains findings or output."""

   def _scan(self, fake: FakeGit, environment: dict[str, str] | None = None, output: Path | None = None) -> int:
      full = {"CI_SOURCE_REVISION": SOURCE_REVISION, "CI_PLATFORM_REVISION": PLATFORM_REVISION, "CI_RUN_ID": "77", "CI_RUN_ATTEMPT": "2"}
      full.update(environment or {})
      scan = partial(scan_history, run_command=fake, parent_environment={"PATH": "/usr/bin"})
      arguments = ["scan", "--source", str(self.source), "--report", ".platform-security/gitleaks.json", "--output", str(output or self.root / "security-output")]
      with mock.patch("platform_ci.__main__.scan_history", scan), self.assertLogs("platform_ci", level="INFO"):
         return main(arguments, full)

   def _manifest(self) -> dict[str, object]:
      loaded: dict[str, object] = json.loads((self.root / "security-output" / "manifest.json").read_text(encoding="utf-8"))
      return loaded

   def test_clean_scan_writes_a_bound_manifest_and_exits_zero(self) -> None:
      """The manifest binds the exact source, platform and run, and its digest verifies."""
      self.assertEqual(self._scan(FakeGit()), 0)
      manifest = self._manifest()
      self.assertEqual(
         {key: manifest[key] for key in ("kind", "source_revision", "platform_revision", "run_id", "run_attempt", "passed", "findings", "failure")},
         {"kind": "secret-scan", "source_revision": SOURCE_REVISION, "platform_revision": PLATFORM_REVISION, "run_id": "77", "run_attempt": "2", "passed": True, "findings": 0, "failure": None},
      )
      self.assertEqual(manifest["digest"], manifest_digest(manifest))
      self.assertEqual(sorted(path.name for path in (self.root / "security-output").iterdir()), ["manifest.json"])

   def test_findings_fail_and_never_reach_the_manifest_or_logs(self) -> None:
      """A finding exits 1; only counts and the raw report's digest are recorded, never its body or tool output."""
      fake = FakeGit(scan_exit_code=2, report=FINDING)
      self.assertEqual(self._scan(fake), 1)
      manifest = self._manifest()
      self.assertEqual((manifest["passed"], manifest["findings"]), (False, 1))
      text = (self.root / "security-output" / "manifest.json").read_text(encoding="utf-8")
      for forbidden in (SENTINEL, "RAW-MATCHED-SECRET-IN-OUTPUT", "generic", "a.txt"):
         self.assertNotIn(forbidden, text)
      self.assertEqual(sorted(cast_report(manifest)), ["sha256", "size"])

   def test_tool_problem_fails_closed_with_a_manifest(self) -> None:
      """An unpinned scanner version or a missing report is a failed manifest, not a pass."""
      for fake in (FakeGit(version="9.9.9"), FakeGit(report=None), FakeGit(shallow="true"), FakeGit(head="c" * 40)):
         with self.subTest(fake=vars(fake)):
            shutil.rmtree(self.root / "security-output", ignore_errors=True)
            self.assertEqual(self._scan(fake), 1)
            self.assertFalse(self._manifest()["passed"])

   def test_missing_run_facts_and_existing_output_fail(self) -> None:
      """Absent identity facts or a reused output directory exit 1 without scanning."""
      fake = FakeGit()
      with self.assertLogs("platform_ci", level="ERROR"):
         self.assertEqual(
            main(["scan", "--source", str(self.source), "--report", "r.json", "--output", str(self.root / "o")], {"CI_SOURCE_REVISION": SOURCE_REVISION}),
            1,
         )
      (self.root / "security-output").mkdir()
      self.assertEqual(self._scan(fake), 1)
      self.assertEqual(fake.calls, [])


def cast_report(manifest: dict[str, object]) -> dict[str, object]:
   """Return the manifest's report metadata mapping."""
   report = manifest["report"]
   assert isinstance(report, dict)
   return {str(key): value for key, value in report.items()}  # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]


class ReportCommandTest(unittest.TestCase):
   """The report command requires the plan and all three job results."""

   def _report(self, event_args: tuple[str, str], **results: str) -> int:
      plan = plan_for(example_configuration("maven.ci.yaml"), event(*event_args))
      environment = {"CI_PLAN": to_compact_json(plan)}
      environment.update({f"CI_{name.upper()}_RESULT": value for name, value in results.items()})
      return main(["report"], environment)

   def test_requires_every_result(self) -> None:
      """An absent secrets result fails closed even when verify succeeded."""
      with self.assertLogs("platform_ci", level="ERROR"):
         self.assertEqual(self._report(("pull_request", "refs/pull/1/merge"), prepare="success", verify="success"), 1)

   def test_outcomes(self) -> None:
      """Two successes pass a requested run; two skips pass an unrequested push as not verified; mixes fail."""
      requested = ("pull_request", "refs/pull/1/merge")
      unrequested = ("push", "refs/heads/feature-x")
      self.assertEqual(self._report(requested, prepare="success", verify="success", secrets="success"), 0)
      self.assertEqual(self._report(requested, prepare="success", verify="success", secrets="skipped"), 1)
      self.assertEqual(self._report(requested, prepare="success", verify="success", secrets="failure"), 1)
      self.assertEqual(self._report(unrequested, prepare="success", verify="skipped", secrets="skipped"), 0)
      self.assertEqual(self._report(unrequested, prepare="success", verify="skipped", secrets="success"), 1)
      with self.assertLogs("platform_ci", level="ERROR"):
         self.assertEqual(self._report(requested, prepare="failure", verify="skipped", secrets="skipped"), 1)