busya
f75c15dbc6
- 10 translate plugin test files (100% coverage on 12 modules) - assistant/handler tools: 85+ tests covering dispatch, registry, resolvers, routes, llm_planner, 13 tool handlers - clean release: artifact_catalog_loader, mappers, approval, publication tests - API routes: translate_helpers, validation_service extensions, datasets to 100% - notifications: providers/service tests - services: profile_preference_service - docs/orthogonal-test-report.md — full speckit.tests audit - Fixes: 3 git_base async mock failures, 4 assistant handler permission-check patches - .gitignore: coverage artifacts
326 lines
14 KiB
Python
326 lines
14 KiB
Python
# #region Test.CleanRelease.PolicyEngine [C:3] [TYPE Module] [SEMANTICS test,clean-release,policy,validation,classification]
|
|
# @BRIEF Tests for CleanPolicyEngine — validate_policy, classify_artifact, validate_resource_source, evaluate_candidate.
|
|
# @RELATION BINDS_TO -> [PolicyEngine]
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from datetime import UTC, datetime
|
|
from src.models.clean_release import CleanPolicySnapshot, ResourceSourceEntry, ResourceSourceRegistry, SourceRegistrySnapshot
|
|
from src.services.clean_release.policy_engine import CleanPolicyEngine, PolicyValidationResult, SourceValidationResult
|
|
|
|
|
|
def _make_policy(**overrides) -> CleanPolicySnapshot:
|
|
"""Create a clean policy snapshot with defaults."""
|
|
params = {
|
|
"id": "policy-1",
|
|
"policy_id": "policy-1",
|
|
"policy_version": "1.0",
|
|
"content_json": {
|
|
"profile": "enterprise-clean",
|
|
"prohibited_artifact_categories": ["malware", "proprietary"],
|
|
"required_system_categories": ["system-lib"],
|
|
"external_source_forbidden": True,
|
|
},
|
|
"registry_snapshot_id": "registry-1",
|
|
"immutable": True,
|
|
}
|
|
params.update(overrides)
|
|
return CleanPolicySnapshot(**params)
|
|
|
|
|
|
def _make_registry(**overrides) -> SourceRegistrySnapshot:
|
|
"""Create a source registry snapshot with defaults."""
|
|
params = {
|
|
"id": "registry-1",
|
|
"registry_id": "registry-1",
|
|
"registry_version": "1.0",
|
|
"allowed_hosts": ["repo.internal.local", "artifacts.internal.local"],
|
|
"immutable": True,
|
|
}
|
|
params.update(overrides)
|
|
return SourceRegistrySnapshot(**params)
|
|
|
|
|
|
def _make_resource_registry(hosts: list[str] | None = None) -> ResourceSourceRegistry:
|
|
"""Create a ResourceSourceRegistry with entries."""
|
|
if hosts is None:
|
|
hosts = ["internal.local", "repo.internal.local"]
|
|
now = datetime.now(UTC)
|
|
entries = [
|
|
ResourceSourceEntry(source_id=f"src-{i}", host=h, protocol="https", purpose="internal", enabled=True)
|
|
for i, h in enumerate(hosts)
|
|
]
|
|
return ResourceSourceRegistry(
|
|
registry_id="reg-res-1", name="Test Resource Registry", entries=entries,
|
|
updated_at=now, updated_by="system",
|
|
)
|
|
|
|
|
|
class TestValidatePolicy:
|
|
"""CleanPolicyEngine.validate_policy — policy consistency validation."""
|
|
|
|
# #region test_validate_policy_ok [C:2] [TYPE Function]
|
|
def test_validate_policy_ok(self):
|
|
"""Valid enterprise-clean policy → ok=True."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert result.ok is True
|
|
assert result.blocking_reasons == []
|
|
# #endregion test_validate_policy_ok
|
|
|
|
# #region test_validate_missing_registry_ref [C:2] [TYPE Function]
|
|
def test_validate_missing_registry_ref(self):
|
|
"""Missing registry_snapshot_id → reason added."""
|
|
policy = _make_policy(registry_snapshot_id="")
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert result.ok is False
|
|
assert any("registry" in r.lower() for r in result.blocking_reasons)
|
|
# #endregion test_validate_missing_registry_ref
|
|
|
|
# #region test_validate_missing_prohibited [C:2] [TYPE Function]
|
|
def test_validate_missing_prohibited(self):
|
|
"""Enterprise policy without prohibited categories → reason added."""
|
|
policy = _make_policy(content_json={"profile": "enterprise-clean"})
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert any("prohibited" in r.lower() for r in result.blocking_reasons)
|
|
# #endregion test_validate_missing_prohibited
|
|
|
|
# #region test_validate_missing_external_forbidden [C:2] [TYPE Function]
|
|
def test_validate_missing_external_forbidden(self):
|
|
"""Enterprise policy without external_source_forbidden → reason added."""
|
|
policy = _make_policy(content_json={
|
|
"profile": "enterprise-clean",
|
|
"prohibited_artifact_categories": ["malware"],
|
|
})
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert any("external" in r.lower() for r in result.blocking_reasons)
|
|
# #endregion test_validate_missing_external_forbidden
|
|
|
|
# #region test_validate_registry_ref_mismatch [C:2] [TYPE Function]
|
|
def test_validate_registry_ref_mismatch(self):
|
|
"""Registry ref mismatch → reason added."""
|
|
policy = _make_policy(registry_snapshot_id="reg-other")
|
|
registry = _make_registry(id="registry-1")
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert any("mismatch" in r.lower() or "ref" in r.lower() for r in result.blocking_reasons)
|
|
# #endregion test_validate_registry_ref_mismatch
|
|
|
|
# #region test_validate_empty_allowed_hosts [C:2] [TYPE Function]
|
|
def test_validate_empty_allowed_hosts(self):
|
|
"""Registry with no allowed hosts → reason added."""
|
|
policy = _make_policy()
|
|
registry = _make_registry(allowed_hosts=[])
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert any("host" in r.lower() for r in result.blocking_reasons)
|
|
# #endregion test_validate_empty_allowed_hosts
|
|
|
|
# #region test_validate_standard_profile_no_prohibited_check [C:2] [TYPE Function]
|
|
def test_validate_standard_profile_no_prohibited_check(self):
|
|
"""Standard profile doesn't require prohibited categories."""
|
|
policy = _make_policy(content_json={"profile": "standard"})
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_policy()
|
|
assert not any("prohibited" in r.lower() for r in result.blocking_reasons)
|
|
# #endregion test_validate_standard_profile_no_prohibited_check
|
|
|
|
|
|
class TestClassifyArtifact:
|
|
"""CleanPolicyEngine.classify_artifact — artifact classification."""
|
|
|
|
# #region test_classify_required_system [C:2] [TYPE Function]
|
|
def test_classify_required_system(self):
|
|
"""Required system category → required-system."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.classify_artifact({"category": "system-lib"})
|
|
assert result == "required-system"
|
|
# #endregion test_classify_required_system
|
|
|
|
# #region test_classify_prohibited [C:2] [TYPE Function]
|
|
def test_classify_prohibited(self):
|
|
"""Prohibited category → excluded-prohibited."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.classify_artifact({"category": "malware"})
|
|
assert result == "excluded-prohibited"
|
|
# #endregion test_classify_prohibited
|
|
|
|
# #region test_classify_allowed [C:2] [TYPE Function]
|
|
def test_classify_allowed(self):
|
|
"""Unlisted category → allowed."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.classify_artifact({"category": "documentation"})
|
|
assert result == "allowed"
|
|
# #endregion test_classify_allowed
|
|
|
|
# #region test_classify_empty_category [C:2] [TYPE Function]
|
|
def test_classify_empty_category(self):
|
|
"""Empty category → allowed."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.classify_artifact({"category": ""})
|
|
assert result == "allowed"
|
|
# #endregion test_classify_empty_category
|
|
|
|
# #region test_classify_from_legacy_policy [C:2] [TYPE Function]
|
|
def test_classify_from_legacy_policy(self):
|
|
"""Policy without content_json but with direct attributes → works."""
|
|
policy = _make_policy(content_json={})
|
|
policy.prohibited_artifact_categories = ["bad"]
|
|
policy.required_system_categories = ["core"]
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
assert engine.classify_artifact({"category": "bad"}) == "excluded-prohibited"
|
|
assert engine.classify_artifact({"category": "core"}) == "required-system"
|
|
assert engine.classify_artifact({"category": "other"}) == "allowed"
|
|
# #endregion test_classify_from_legacy_policy
|
|
|
|
|
|
class TestValidateResourceSource:
|
|
"""CleanPolicyEngine.validate_resource_source — source endpoint validation."""
|
|
|
|
# #region test_empty_endpoint [C:2] [TYPE Function]
|
|
def test_empty_endpoint(self):
|
|
"""Empty endpoint → violation with blocked_release=True."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_resource_source("")
|
|
assert result.ok is False
|
|
assert result.violation is not None
|
|
assert result.violation["blocked_release"] is True
|
|
# #endregion test_empty_endpoint
|
|
|
|
# #region test_endpoint_in_allowlist [C:2] [TYPE Function]
|
|
def test_endpoint_in_allowlist(self):
|
|
"""Endpoint in allowed_hosts → ok=True."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_resource_source("repo.internal.local")
|
|
assert result.ok is True
|
|
assert result.violation is None
|
|
# #endregion test_endpoint_in_allowlist
|
|
|
|
# #region test_endpoint_not_in_allowlist [C:2] [TYPE Function]
|
|
def test_endpoint_not_in_allowlist(self):
|
|
"""Endpoint outside allowlist → violation."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_resource_source("external.bad.com")
|
|
assert result.ok is False
|
|
assert result.violation["location"] == "external.bad.com"
|
|
# #endregion test_endpoint_not_in_allowlist
|
|
|
|
# #region test_endpoint_case_insensitive_matching [C:2] [TYPE Function]
|
|
def test_endpoint_case_insensitive_matching(self):
|
|
"""Endpoint matching is case-insensitive — endpoint is lowered before comparison."""
|
|
policy = _make_policy()
|
|
registry = _make_registry(allowed_hosts=["repo.internal.local"])
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
# Different case → still found because endpoint is lowered
|
|
result = engine.validate_resource_source("REPO.INTERNAL.LOCAL")
|
|
assert result.ok is True
|
|
# Exact case → also found
|
|
result = engine.validate_resource_source("repo.internal.local")
|
|
assert result.ok is True
|
|
# #endregion test_endpoint_case_insensitive_matching
|
|
|
|
# #region test_registry_with_resource_source [C:2] [TYPE Function]
|
|
def test_registry_with_resource_source(self):
|
|
"""Using ResourceSourceRegistry instead of SourceRegistrySnapshot."""
|
|
registry = _make_resource_registry(hosts=["internal.local"])
|
|
policy = _make_policy()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
result = engine.validate_resource_source("internal.local")
|
|
assert result.ok is True
|
|
# #endregion test_registry_with_resource_source
|
|
|
|
|
|
class TestEvaluateCandidate:
|
|
"""CleanPolicyEngine.evaluate_candidate — full candidate evaluation."""
|
|
|
|
# #region test_evaluate_candidate_clean [C:2] [TYPE Function]
|
|
def test_evaluate_candidate_clean(self):
|
|
"""Clean candidate with valid sources → no violations."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
artifacts = [
|
|
{"path": "lib/core.so", "category": "system-lib"},
|
|
{"path": "doc/readme.md", "category": "documentation"},
|
|
]
|
|
sources = ["repo.internal.local"]
|
|
classified, violations = engine.evaluate_candidate(artifacts, sources)
|
|
assert len(classified) == 2
|
|
assert len(violations) == 0
|
|
assert classified[0]["classification"] == "required-system"
|
|
assert classified[1]["classification"] == "allowed"
|
|
# #endregion test_evaluate_candidate_clean
|
|
|
|
# #region test_evaluate_candidate_with_violations [C:2] [TYPE Function]
|
|
def test_evaluate_candidate_with_violations(self):
|
|
"""Candidate with prohibited artifacts and bad sources → violations."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
artifacts = [
|
|
{"path": "lib/malware.exe", "category": "malware"},
|
|
{"path": "lib/ok.so", "category": "system-lib"},
|
|
]
|
|
sources = ["repo.internal.local", "external.bad.com"]
|
|
classified, violations = engine.evaluate_candidate(artifacts, sources)
|
|
assert len(classified) == 2
|
|
assert len(violations) >= 1
|
|
assert violations[0]["category"] == "data-purity"
|
|
# #endregion test_evaluate_candidate_with_violations
|
|
|
|
# #region test_evaluate_candidate_empty [C:2] [TYPE Function]
|
|
def test_evaluate_candidate_empty(self):
|
|
"""Empty artifacts and sources → no violations."""
|
|
policy = _make_policy()
|
|
registry = _make_registry()
|
|
engine = CleanPolicyEngine(policy=policy, registry=registry)
|
|
classified, violations = engine.evaluate_candidate([], [])
|
|
assert classified == []
|
|
assert violations == []
|
|
# #endregion test_evaluate_candidate_empty
|
|
|
|
|
|
class TestPolicyValidationResult:
|
|
"""PolicyValidationResult and SourceValidationResult dataclasses."""
|
|
|
|
# #region test_policy_validation_result [C:2] [TYPE Function]
|
|
def test_policy_validation_result(self):
|
|
"""PolicyValidationResult dataclass."""
|
|
r = PolicyValidationResult(ok=True, blocking_reasons=[])
|
|
assert r.ok is True
|
|
assert r.blocking_reasons == []
|
|
# #endregion test_policy_validation_result
|
|
|
|
# #region test_source_validation_result [C:2] [TYPE Function]
|
|
def test_source_validation_result(self):
|
|
"""SourceValidationResult dataclass."""
|
|
r = SourceValidationResult(ok=False, violation={"code": "EXT"})
|
|
assert r.ok is False
|
|
assert r.violation["code"] == "EXT"
|
|
# #endregion test_source_validation_result
|