resource-control/tests/test_capability.py

import json
import sys
import unittest
from copy import deepcopy
from pathlib import Path

ROOT = Path(__file__).parents[1]
sys.path.insert(0, str(ROOT / "tools"))
from capability import (
    consumption_profile,
    evidence_coverage,
    load_canon,
    review,
    validate_provision,
    validate_requirement,
)

RECORD = json.loads((ROOT / "data/capability/platform-audit-storage.json").read_text())


def canon():
    return load_canon()


class CanonBindingTest(unittest.TestCase):
    """We read the canon rather than copying it; drift must fail here."""

    def setUp(self):
        self.canon = canon()

    def test_catalog_is_the_version_we_restated_against(self):
        self.assertEqual("0.4.0", self.canon["version"])
        self.assertEqual("0.6.0", self.canon["canon_version"])
        self.assertEqual(self.canon["version"], RECORD["canon"]["model_version"])
        self.assertEqual(self.canon["canon_version"], RECORD["canon"]["canon_version"])

    def test_human_effort_and_intelligence_classes_exist_with_native_units(self):
        classes = self.canon["resource_classes"]
        self.assertEqual("hour", classes["H"]["native_unit"])
        self.assertEqual("internal", classes["H"]["supply"])
        self.assertEqual("constrained", classes["H"]["capacity_behaviour"])
        self.assertEqual("token", classes["I"]["native_unit"])

    def test_unknown_capability_is_rejected(self):
        with self.assertRaisesRegex(ValueError, "unknown capability"):
            validate_requirement({"capability": "data.telepathy"}, self.canon)


class RequirementTest(unittest.TestCase):
    def setUp(self):
        self.canon = canon()
        self.requirement = deepcopy(RECORD["requires"][0])

    def test_the_real_requirement_validates(self):
        validate_requirement(self.requirement, self.canon)

    def test_target_key_must_be_a_declared_quality_dimension(self):
        self.requirement["targets"]["chattiness"] = {"value": 1, "unit": "x"}
        with self.assertRaisesRegex(ValueError, "quality dimension"):
            validate_requirement(self.requirement, self.canon)

    def test_constraint_dimension_must_be_declared(self):
        self.requirement["constraints"][0]["dimension"] = "vibes"
        with self.assertRaisesRegex(ValueError, "quality dimension"):
            validate_requirement(self.requirement, self.canon)

    def test_predicate_must_be_in_the_closed_set(self):
        self.requirement["constraints"][0]["predicate"] = "sort_of_near"
        with self.assertRaisesRegex(ValueError, "closed set"):
            validate_requirement(self.requirement, self.canon)

    def test_profile_must_be_declared_on_that_capability(self):
        self.requirement["profile"] = "volume-of-vibes"
        with self.assertRaisesRegex(ValueError, "profile"):
            validate_requirement(self.requirement, self.canon)

    def test_the_constraint_that_decided_procurement_is_expressible(self):
        constraint = self.requirement["constraints"][0]
        self.assertEqual("geographical_separation", constraint["dimension"])
        self.assertEqual("not_in", constraint["predicate"])
        self.assertIn("railiance01", [entry["id"] for entry in constraint["of"]])


class ProvisionTest(unittest.TestCase):
    def setUp(self):
        self.canon = canon()
        self.provisions = {p["capability"]: deepcopy(p) for p in RECORD["provisions"]}

    def test_both_real_provisions_validate(self):
        for provision in self.provisions.values():
            validate_provision(provision, self.canon)

    def test_consumption_unit_must_be_the_class_native_unit(self):
        provision = self.provisions["data.object"]
        row = next(r for r in provision["consumes"] if r["class"] == "H")
        row["quantity"]["unit"] = "eur"
        with self.assertRaisesRegex(ValueError, "native unit"):
            validate_provision(provision, self.canon)

    def test_currency_cannot_be_smuggled_in_as_a_class(self):
        provision = self.provisions["data.object"]
        provision["consumes"].append(
            {"class": "EUR", "quantity": {"value": 7.35, "unit": "eur"}, "basis": "quoted"}
        )
        with self.assertRaisesRegex(ValueError, "unknown resource class"):
            validate_provision(provision, self.canon)

    def test_unknown_consumption_may_not_be_recorded_as_zero(self):
        provision = self.provisions["data.backup"]
        row = next(r for r in provision["consumes"] if r["basis"] == "unknown")
        row["quantity"]["value"] = 0
        with self.assertRaises(ValueError):
            validate_provision(provision, self.canon)

    def test_evidence_hook_must_be_declared_on_the_capability(self):
        provision = self.provisions["data.backup"]
        provision["evidence"][0]["hook"] = "vibes_check"
        with self.assertRaisesRegex(ValueError, "evidence hook"):
            validate_provision(provision, self.canon)

    def test_duplicate_class_rows_are_rejected(self):
        provision = self.provisions["data.backup"]
        provision["consumes"].append(deepcopy(provision["consumes"][0]))
        with self.assertRaisesRegex(ValueError, "duplicate"):
            validate_provision(provision, self.canon)

    def test_backup_provision_satisfies_all_four_declared_evidence_hooks(self):
        coverage = evidence_coverage(self.provisions["data.backup"], self.canon)
        self.assertTrue(coverage["complete"])
        self.assertEqual([], coverage["missing"])
        self.assertEqual(4, len(coverage["supplied"]))

    def test_object_provision_is_honest_about_missing_hooks(self):
        coverage = evidence_coverage(self.provisions["data.object"], self.canon)
        self.assertFalse(coverage["complete"])
        self.assertIn("object_integrity_tests", coverage["missing"])

    def test_using_another_capability_is_a_relation_not_a_p_row(self):
        """CAP-R11. The credential-custody P row was the wrong kind."""
        backup = self.provisions["data.backup"]
        classes = {r["class"] for r in backup["consumes"]}
        self.assertNotIn("P", classes)
        relations = {u["capability"]: u["relation"] for u in backup["uses_provisions"]}
        self.assertEqual("may_use", relations["security.secrets"])
        self.assertEqual("depends_on", relations["data.object"])

    def test_depends_on_must_be_declared_between_the_capabilities(self):
        backup = self.provisions["data.backup"]
        entry = next(u for u in backup["uses_provisions"] if u["capability"] == "security.secrets")
        entry["relation"] = "depends_on"
        with self.assertRaisesRegex(ValueError, "does not declare depends_on"):
            validate_provision(backup, self.canon)

    def test_relation_outside_the_canon_vocabulary_is_rejected(self):
        backup = self.provisions["data.backup"]
        backup["uses_provisions"][0]["relation"] = "uses"
        with self.assertRaisesRegex(ValueError, "depends_on or may_use"):
            validate_provision(backup, self.canon)

    def test_a_used_provision_must_name_its_provider(self):
        backup = self.provisions["data.backup"]
        backup["uses_provisions"][0]["provider"] = ""
        with self.assertRaisesRegex(ValueError, "must name a provider"):
            validate_provision(backup, self.canon)

    def test_effort_and_tokens_are_recorded_in_native_units(self):
        rows = {r["class"]: r for r in self.provisions["data.object"]["consumes"]}
        self.assertEqual("hour", rows["H"]["quantity"]["unit"])
        self.assertEqual("token", rows["I"]["quantity"]["unit"])
        self.assertEqual("unknown", rows["I"]["basis"])


class ReviewTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.report = review(deepcopy(RECORD), canon())

    def test_requirement_is_reported_as_below_the_asked_maturity(self):
        requirement = self.report["requirements"][0]
        self.assertEqual("D5", requirement["required"])
        self.assertEqual("D4", requirement["provided"])
        self.assertEqual("below_requirement", requirement["status"])

    def test_provider_comparison_grades_as_indicative_not_evidenced(self):
        grade = self.report["alternatives_grade"]
        self.assertEqual("indicative", grade["grade"])
        self.assertEqual("assumed", grade["weakest"])
        self.assertEqual(0.25, grade["evidenced_ratio"])

    def test_consumption_profiles_report_what_is_still_unknown(self):
        for provision in self.report["provisions"]:
            self.assertEqual("unknown", provision["consumption"]["weakest"])
            self.assertGreater(provision["consumption"]["count"], 0)


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