267 lines
9.7 KiB
Python
267 lines
9.7 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import copy
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import unittest
|
||
|
|
from dataclasses import FrozenInstanceError
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from jsonschema import Draft202012Validator
|
||
|
|
|
||
|
|
from docforge.errors import DocForgeError
|
||
|
|
from docforge.projection_policy import (
|
||
|
|
ProjectionPolicyV2,
|
||
|
|
compose_projection_policy,
|
||
|
|
)
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
POLICY_SCHEMA = json.loads(
|
||
|
|
(ROOT / "schemas" / "projection-policy.schema.json").read_text(encoding="utf-8")
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class ProjectionPolicyV2Tests(unittest.TestCase):
|
||
|
|
def test_compatible_defaults_are_independent_and_schema_valid(self) -> None:
|
||
|
|
cases = (
|
||
|
|
(
|
||
|
|
{
|
||
|
|
"manual_configured": False,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
},
|
||
|
|
("disabled", "disabled", "on-demand"),
|
||
|
|
),
|
||
|
|
(
|
||
|
|
{
|
||
|
|
"manual_configured": True,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
},
|
||
|
|
("explicit", "disabled", "on-demand"),
|
||
|
|
),
|
||
|
|
(
|
||
|
|
{
|
||
|
|
"manual_configured": True,
|
||
|
|
"portable_graph_configured": True,
|
||
|
|
"application_enabled": True,
|
||
|
|
},
|
||
|
|
("auto", "explicit", "on-demand"),
|
||
|
|
),
|
||
|
|
(
|
||
|
|
{
|
||
|
|
"manual_configured": True,
|
||
|
|
"portable_graph_configured": True,
|
||
|
|
"application_enabled": False,
|
||
|
|
"live_viewer_available": False,
|
||
|
|
},
|
||
|
|
("explicit", "explicit", "disabled"),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
validator = Draft202012Validator(POLICY_SCHEMA)
|
||
|
|
for arguments, expected in cases:
|
||
|
|
with self.subTest(arguments=arguments):
|
||
|
|
policy = compose_projection_policy(**arguments)
|
||
|
|
self.assertEqual(
|
||
|
|
expected, (policy.manual, policy.portable_graph, policy.live_viewer)
|
||
|
|
)
|
||
|
|
validator.validate(policy.as_dict())
|
||
|
|
|
||
|
|
def test_explicit_selections_can_narrow_each_consumer_independently(self) -> None:
|
||
|
|
policy = compose_projection_policy(
|
||
|
|
manual="explicit",
|
||
|
|
portable_graph="disabled",
|
||
|
|
live_viewer="disabled",
|
||
|
|
manual_configured=True,
|
||
|
|
portable_graph_configured=True,
|
||
|
|
application_enabled=True,
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
{
|
||
|
|
"schema_version": 2,
|
||
|
|
"manual": "explicit",
|
||
|
|
"portable_graph": "disabled",
|
||
|
|
"live_viewer": "disabled",
|
||
|
|
},
|
||
|
|
policy.as_dict(),
|
||
|
|
)
|
||
|
|
|
||
|
|
fully_disabled = compose_projection_policy(
|
||
|
|
manual="disabled",
|
||
|
|
portable_graph="disabled",
|
||
|
|
live_viewer="disabled",
|
||
|
|
manual_configured=False,
|
||
|
|
portable_graph_configured=False,
|
||
|
|
application_enabled=False,
|
||
|
|
live_viewer_available=False,
|
||
|
|
)
|
||
|
|
self.assertEqual("disabled", fully_disabled.manual)
|
||
|
|
self.assertEqual("disabled", fully_disabled.portable_graph)
|
||
|
|
self.assertEqual("disabled", fully_disabled.live_viewer)
|
||
|
|
|
||
|
|
def test_policy_is_frozen_and_hashes_exact_canonical_payload(self) -> None:
|
||
|
|
first = compose_projection_policy(
|
||
|
|
manual_configured=True,
|
||
|
|
portable_graph_configured=True,
|
||
|
|
application_enabled=False,
|
||
|
|
)
|
||
|
|
second = compose_projection_policy(
|
||
|
|
manual_configured=True,
|
||
|
|
portable_graph_configured=True,
|
||
|
|
application_enabled=False,
|
||
|
|
)
|
||
|
|
canonical = json.dumps(
|
||
|
|
first.as_dict(),
|
||
|
|
sort_keys=True,
|
||
|
|
separators=(",", ":"),
|
||
|
|
ensure_ascii=False,
|
||
|
|
).encode("utf-8")
|
||
|
|
self.assertEqual(first.as_dict(), second.as_dict())
|
||
|
|
self.assertEqual(hashlib.sha256(canonical).hexdigest(), first.policy_hash)
|
||
|
|
self.assertEqual(first.policy_hash, second.policy_hash)
|
||
|
|
|
||
|
|
changed = compose_projection_policy(
|
||
|
|
manual="disabled",
|
||
|
|
manual_configured=True,
|
||
|
|
portable_graph_configured=True,
|
||
|
|
application_enabled=False,
|
||
|
|
)
|
||
|
|
self.assertNotEqual(first.policy_hash, changed.policy_hash)
|
||
|
|
with self.assertRaises(FrozenInstanceError):
|
||
|
|
first.manual = "disabled" # type: ignore[misc]
|
||
|
|
|
||
|
|
def test_invalid_modes_and_direct_construction_fail_closed(self) -> None:
|
||
|
|
cases = (
|
||
|
|
("manual", {"manual": "sometimes"}),
|
||
|
|
("portable_graph", {"portable_graph": "auto"}),
|
||
|
|
("live_viewer", {"live_viewer": "always"}),
|
||
|
|
("manual", {"manual": 1}),
|
||
|
|
)
|
||
|
|
for expected_projection, selection in cases:
|
||
|
|
with self.subTest(selection=selection):
|
||
|
|
with self.assertRaises(DocForgeError) as raised:
|
||
|
|
compose_projection_policy(
|
||
|
|
**selection,
|
||
|
|
manual_configured=True,
|
||
|
|
portable_graph_configured=True,
|
||
|
|
application_enabled=True,
|
||
|
|
)
|
||
|
|
self.assertEqual("invalid_projection_policy", raised.exception.code)
|
||
|
|
self.assertEqual(
|
||
|
|
expected_projection,
|
||
|
|
raised.exception.details["projection"],
|
||
|
|
)
|
||
|
|
|
||
|
|
with self.assertRaises(DocForgeError) as direct:
|
||
|
|
ProjectionPolicyV2(
|
||
|
|
manual="automatic", # type: ignore[arg-type]
|
||
|
|
portable_graph="explicit",
|
||
|
|
live_viewer="on-demand",
|
||
|
|
)
|
||
|
|
self.assertEqual("invalid_projection_policy", direct.exception.code)
|
||
|
|
|
||
|
|
def test_resource_and_capability_unavailability_are_distinct(self) -> None:
|
||
|
|
cases = (
|
||
|
|
(
|
||
|
|
{"manual": "explicit"},
|
||
|
|
{
|
||
|
|
"manual_configured": False,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
},
|
||
|
|
"manual",
|
||
|
|
"manual_render_config",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
{"manual": "auto"},
|
||
|
|
{
|
||
|
|
"manual_configured": True,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
},
|
||
|
|
"manual",
|
||
|
|
"canonical_application",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
{"portable_graph": "explicit"},
|
||
|
|
{
|
||
|
|
"manual_configured": False,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
},
|
||
|
|
"portable_graph",
|
||
|
|
"portable_graph_render_config",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
{"live_viewer": "on-demand"},
|
||
|
|
{
|
||
|
|
"manual_configured": False,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
"live_viewer_available": False,
|
||
|
|
},
|
||
|
|
"live_viewer",
|
||
|
|
"live_viewer_runtime",
|
||
|
|
),
|
||
|
|
)
|
||
|
|
for selection, availability, projection, required in cases:
|
||
|
|
with self.subTest(selection=selection):
|
||
|
|
with self.assertRaises(DocForgeError) as raised:
|
||
|
|
compose_projection_policy(**selection, **availability)
|
||
|
|
self.assertEqual("projection_policy_unavailable", raised.exception.code)
|
||
|
|
self.assertEqual(projection, raised.exception.details["projection"])
|
||
|
|
self.assertEqual(required, raised.exception.details["required"])
|
||
|
|
|
||
|
|
def test_availability_inputs_must_be_real_booleans(self) -> None:
|
||
|
|
cases = (
|
||
|
|
{"manual_configured": 1},
|
||
|
|
{"portable_graph_configured": 0},
|
||
|
|
{"application_enabled": "yes"},
|
||
|
|
{"live_viewer_available": None},
|
||
|
|
)
|
||
|
|
for replacement in cases:
|
||
|
|
arguments = {
|
||
|
|
"manual_configured": False,
|
||
|
|
"portable_graph_configured": False,
|
||
|
|
"application_enabled": False,
|
||
|
|
"live_viewer_available": True,
|
||
|
|
**replacement,
|
||
|
|
}
|
||
|
|
with self.subTest(replacement=replacement):
|
||
|
|
with self.assertRaises(DocForgeError) as raised:
|
||
|
|
compose_projection_policy(**arguments)
|
||
|
|
self.assertEqual("invalid_projection_policy", raised.exception.code)
|
||
|
|
|
||
|
|
def test_schema_rejects_every_runtime_contract_drift(self) -> None:
|
||
|
|
validator = Draft202012Validator(POLICY_SCHEMA)
|
||
|
|
valid = compose_projection_policy(
|
||
|
|
manual_configured=True,
|
||
|
|
portable_graph_configured=True,
|
||
|
|
application_enabled=True,
|
||
|
|
).as_dict()
|
||
|
|
invalid_documents: list[dict[str, object]] = []
|
||
|
|
for field in ("schema_version", "manual", "portable_graph", "live_viewer"):
|
||
|
|
missing = copy.deepcopy(valid)
|
||
|
|
missing.pop(field)
|
||
|
|
invalid_documents.append(missing)
|
||
|
|
for field, value in (
|
||
|
|
("schema_version", 1),
|
||
|
|
("manual", "on-demand"),
|
||
|
|
("portable_graph", "auto"),
|
||
|
|
("live_viewer", "explicit"),
|
||
|
|
):
|
||
|
|
changed = copy.deepcopy(valid)
|
||
|
|
changed[field] = value
|
||
|
|
invalid_documents.append(changed)
|
||
|
|
extra = copy.deepcopy(valid)
|
||
|
|
extra["project_path"] = "/private/project"
|
||
|
|
invalid_documents.append(extra)
|
||
|
|
|
||
|
|
for document in invalid_documents:
|
||
|
|
with self.subTest(document=document):
|
||
|
|
self.assertFalse(validator.is_valid(document))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|