test_verification_depots.py #1

"""Real CLI output shapes must not become an empty successful depot inventory."""
import importlib.util
from pathlib import Path
import subprocess
import sys
import unittest
from unittest.mock import Mock, patch


class VerificationDepots(unittest.TestCase):
    def setUp(self):
        path = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin/p4verify.py"
        spec = importlib.util.spec_from_file_location("verified_depot_query", path)
        self.module = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = self.module
        self.addCleanup(sys.modules.pop, spec.name, None)
        spec.loader.exec_module(self.module)
        self.verify = self.module.P4Verify.__new__(self.module.P4Verify)
        self.verify.p4_base, self.verify.L = ["fixture-p4"], Mock()
        self.verify.bail = Mock(side_effect=RuntimeError("verification stopped"))

    def test_tagged_query_preserves_all_depot_names_and_types(self):
        def query(argv, **kwargs):
            # Captured shapes from p4 2026.1: untagged -Mj yields text in data.
            output = ('{"name":"depot","type":"local"}\n'
                      '{"name":"a.b-1","type":"stream"}\n'
                      '{"name":"spec","type":"spec"}\n'
                      '{"name":"unload","type":"unload"}\n') if "-ztag" in argv else '{"data":"Depot depot 2026/10/17 local depot/..."}\n'
            return subprocess.CompletedProcess(argv, 0, output, "")
        with patch.object(self.module.subprocess, "run", side_effect=query):
            self.assertEqual(self.verify._get_depots_by_type(), {"depot": "local", "a.b-1": "stream", "spec": "spec", "unload": "unload"})

    def test_query_failure_stops_verification_instead_of_empty_success(self):
        self.verify.run_p4_mj = Mock(return_value=([], None, 1))
        with self.assertRaises(RuntimeError):
            self.verify._get_depots_by_type()

    def test_unrecognized_record_cannot_silently_exclude_a_depot(self):
        for record in ({"data": "Depot unknown local unknown/..."}, {"data": "Permission denied", "severity": 3}):
            with self.subTest(record=record), self.assertRaises(RuntimeError):
                self.verify.run_p4_mj = Mock(return_value=([record], None, 0))
                self.verify._get_depots_by_type()


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