diff --git a/abevalflow/certification.py b/abevalflow/certification.py index 5d0bd11..d7c53f0 100644 --- a/abevalflow/certification.py +++ b/abevalflow/certification.py @@ -88,7 +88,7 @@ class CheckId(StrEnum): CheckId.FUNCTIONAL_VALIDATION, CheckId.INSTRUCTION_QUALITY, # CheckId.REGISTRY_GOVERNANCE, # Not yet implemented - # CheckId.OPERATIONAL_POLICY_COMPLIANCE, # Not yet implemented + CheckId.OPERATIONAL_POLICY_COMPLIANCE, ] CERTIFIED_CHECKS = [ @@ -510,6 +510,7 @@ def compute_certification( metadata_valid: bool = True, has_eval_assets: bool = True, policy: CertificationPolicy | None = None, + operational_policy_result: CheckResult | None = None, ) -> CertificationResult: """Compute certification levels from gate results. @@ -581,16 +582,19 @@ def compute_certification( ), ) - all_checks.setdefault( - CheckId.OPERATIONAL_POLICY_COMPLIANCE, - CheckResult( - check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, - name="Operational Policy Compliance", - passed=False, - score=0.0, - message="Operational policy check not implemented", - ), - ) + if operational_policy_result is not None: + all_checks[CheckId.OPERATIONAL_POLICY_COMPLIANCE] = operational_policy_result + else: + all_checks.setdefault( + CheckId.OPERATIONAL_POLICY_COMPLIANCE, + CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=False, + score=0.0, + message="Operational policy check not implemented", + ), + ) all_checks.setdefault( CheckId.ENTERPRISE_BEHAVIORAL_TESTING, diff --git a/abevalflow/operational_policy.py b/abevalflow/operational_policy.py new file mode 100644 index 0000000..96aff92 --- /dev/null +++ b/abevalflow/operational_policy.py @@ -0,0 +1,177 @@ +"""Operational policy compliance checks for skill submissions. + +Deterministic checks that validate submissions meet operational standards: +resource limits, timeout compliance, error handling patterns, and logging +suppression. No LLM calls required. + +These checks complement (not duplicate) Foundational-level schema validation. +Schema validation (Pydantic) enforces field types and ``gt=0`` constraints; +these checks enforce *policy limits* — e.g., ``cpus <= MAX_CPUS``. +""" + +from __future__ import annotations + +import ast +import logging +import re +from pathlib import Path + +import yaml +from pydantic import ValidationError + +from abevalflow.certification import CheckId, CheckResult +from abevalflow.schemas import OperationalLimits, SubmissionMetadata + +logger = logging.getLogger(__name__) + +LOGGING_SUPPRESSION_PATTERNS = [ + re.compile(r"\bdo\s+not\s+log\b", re.IGNORECASE), + re.compile(r"\bdisable\s+logging\b", re.IGNORECASE), + re.compile(r"\bsuppress\s+output\b", re.IGNORECASE), + re.compile(r"\bno\s+logging\b", re.IGNORECASE), + re.compile(r"\bturn\s+off\s+log", re.IGNORECASE), +] + +SECURITY_QUALIFIERS = re.compile( + r"\b(passwords?|credentials?|secrets?|pii|tokens?|sensitive|personal)\b", + re.IGNORECASE, +) + + +EXPECTED_RESOURCE_FIELDS = ["cpus", "memory_mb", "agent_timeout_sec"] + + +def _check_resource_declaration(raw_data: dict) -> list[str]: + """Check that resource fields are explicitly declared, not just using defaults.""" + missing = [f for f in EXPECTED_RESOURCE_FIELDS if f not in raw_data] + if missing: + return [f"Resource fields not explicitly declared: {', '.join(missing)}"] + return [] + + +def _check_resource_limits(metadata: SubmissionMetadata, limits: OperationalLimits) -> list[str]: + """Check that resource requests are within policy limits.""" + issues = [] + if metadata.cpus > limits.max_cpus: + issues.append(f"CPU request ({metadata.cpus}) exceeds max allowed ({limits.max_cpus})") + if metadata.memory_mb > limits.max_memory_mb: + issues.append(f"Memory request ({metadata.memory_mb}MB) exceeds max allowed ({limits.max_memory_mb}MB)") + return issues + + +def _check_timeout_compliance(metadata: SubmissionMetadata, limits: OperationalLimits) -> list[str]: + """Check that timeouts are within reasonable bounds.""" + issues = [] + if metadata.agent_timeout_sec > limits.max_agent_timeout_sec: + issues.append( + f"Agent timeout ({metadata.agent_timeout_sec}s) exceeds max allowed ({limits.max_agent_timeout_sec}s)" + ) + return issues + + +def _check_error_handling(test_file: Path) -> list[str]: + """Check for bare except patterns that hide failures.""" + issues = [] + if not test_file.exists(): + return issues + + try: + tree = ast.parse(test_file.read_text()) + except SyntaxError: + return issues + + for node in ast.walk(tree): + if not isinstance(node, ast.ExceptHandler): + continue + handler_body = node.body + if len(handler_body) != 1: + continue + stmt = handler_body[0] + is_pass = isinstance(stmt, ast.Pass) + is_ellipsis = isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and stmt.value.value is ... + if not (is_pass or is_ellipsis): + continue + if node.type is None: + issues.append(f"Bare 'except: pass' at line {node.lineno} — hides all errors") + elif isinstance(node.type, ast.Name) and node.type.id == "Exception": + issues.append(f"'except Exception: pass' at line {node.lineno} — hides all errors") + + return issues + + +def _check_logging_suppression(skill_path: Path) -> list[str]: + """Check SKILL.md for patterns that suppress agent logging. + + Skips lines that contain security qualifiers (e.g., "Do not log passwords") + since those are legitimate security instructions, not logging suppression. + """ + issues = [] + if not skill_path.exists(): + return issues + + content = skill_path.read_text() + for line in content.splitlines(): + for pattern in LOGGING_SUPPRESSION_PATTERNS: + match = pattern.search(line) + if match and not SECURITY_QUALIFIERS.search(line): + issues.append(f"Logging suppression pattern found: '{match.group()}'") + + return issues + + +def check_operational_policy( + submission_dir: Path, + limits: OperationalLimits | None = None, +) -> CheckResult: + """Run all operational policy checks on a submission. + + Args: + submission_dir: Path to the submission directory containing + metadata.yaml, skills/SKILL.md, and tests/test_outputs.py. + limits: Optional configurable limits from CertificationPolicy. + Uses OperationalLimits defaults if not provided. + + Returns: + CheckResult with aggregated pass/fail, score, and message. + """ + if limits is None: + limits = OperationalLimits() + + all_issues: list[str] = [] + + metadata_path = submission_dir / "metadata.yaml" + if metadata_path.exists(): + try: + data = yaml.safe_load(metadata_path.read_text()) or {} + all_issues.extend(_check_resource_declaration(data)) + metadata = SubmissionMetadata(**data) + all_issues.extend(_check_resource_limits(metadata, limits)) + all_issues.extend(_check_timeout_compliance(metadata, limits)) + except (ValidationError, yaml.YAMLError) as e: + logger.warning("Failed to parse metadata.yaml for policy check: %s", e) + else: + logger.info("No metadata.yaml found, skipping resource/timeout checks") + + test_file = submission_dir / "tests" / "test_outputs.py" + all_issues.extend(_check_error_handling(test_file)) + + skills_dir = submission_dir / "skills" + if skills_dir.exists(): + for skill_file in skills_dir.glob("*.md"): + all_issues.extend(_check_logging_suppression(skill_file)) + + passed = len(all_issues) == 0 + score = 1.0 if passed else max(0.0, 1.0 - (len(all_issues) * 0.25)) + + if passed: + message = "All operational policy checks passed" + else: + message = f"Operational policy issues ({len(all_issues)}): " + "; ".join(all_issues) + + return CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=passed, + score=score, + message=message, + ) diff --git a/abevalflow/schemas.py b/abevalflow/schemas.py index ad9d05a..1887f31 100644 --- a/abevalflow/schemas.py +++ b/abevalflow/schemas.py @@ -233,6 +233,18 @@ class ExperimentType(StrEnum): CUSTOM = "custom" +class OperationalLimits(BaseModel): + """Configurable limits for operational policy compliance checks. + + Defaults for operational policy checks. Can be overridden via + certification_policy.operational_limits in metadata.yaml. + """ + + max_cpus: int = Field(default=4, gt=0, description="Maximum allowed CPU cores") + max_memory_mb: int = Field(default=8192, gt=0, description="Maximum allowed memory in MB") + max_agent_timeout_sec: float = Field(default=3600.0, gt=0, description="Maximum allowed agent timeout in seconds") + + class CertificationLevelPolicy(BaseModel): """Policy configuration for a single certification level. @@ -291,6 +303,9 @@ class CertificationPolicy(BaseModel): Example in metadata.yaml: certification_policy: + operational_limits: + max_cpus: 8 + max_memory_mb: 16384 foundational: checks: - valid_skill_structure @@ -320,6 +335,14 @@ class CertificationPolicy(BaseModel): default=None, description="Policy for Certified certification level", ) + operational_limits: OperationalLimits | None = Field( + default=None, + description=( + "Operational policy limits for resource, timeout, and other checks. " + "Used by the Trusted-level operational_policy_compliance check. " + "Overrides OperationalLimits defaults." + ), + ) def get_checks_for_level(self, level: str) -> list[str] | None: """Get custom check list for a level, or None to use defaults.""" @@ -343,6 +366,10 @@ def get_threshold(self, check_id: str) -> float | None: result = level_policy.thresholds[check_id] return result + def get_operational_limits(self) -> OperationalLimits | None: + """Get operational limits for the operational policy compliance check.""" + return self.operational_limits + class CopySpec(BaseModel): """A source directory and its destination path inside the container.""" diff --git a/config/certification_profiles.yaml b/config/certification_profiles.yaml index c922bc8..aeeb89b 100644 --- a/config/certification_profiles.yaml +++ b/config/certification_profiles.yaml @@ -25,6 +25,7 @@ profiles: - evaluation_assets - functional_validation - instruction_quality + - operational_policy_compliance certified: checks: - enterprise_structure_validation @@ -42,7 +43,7 @@ profiles: checks: - evaluation_assets - functional_validation - # - operational_policy_compliance # Not yet implemented + - operational_policy_compliance certified: checks: - enterprise_structure_validation @@ -62,6 +63,7 @@ profiles: - evaluation_assets - advanced_security_validation - functional_validation + - operational_policy_compliance certified: checks: - enterprise_structure_validation @@ -81,6 +83,7 @@ profiles: - evaluation_assets - functional_validation - instruction_quality + - operational_policy_compliance certified: checks: - enterprise_structure_validation @@ -105,7 +108,7 @@ profiles: - instruction_quality # Not yet implemented: # - registry_governance - # - operational_policy_compliance + - operational_policy_compliance certified: checks: - enterprise_structure_validation diff --git a/scripts/aggregate_scorecard.py b/scripts/aggregate_scorecard.py index 971ec10..a8eb9eb 100644 --- a/scripts/aggregate_scorecard.py +++ b/scripts/aggregate_scorecard.py @@ -45,6 +45,7 @@ from abevalflow.gates.base import GateResult from abevalflow.gates.quality import get_all_quality_gates from abevalflow.gates.security import get_all_security_gates +from abevalflow.operational_policy import check_operational_policy from abevalflow.schemas import CertificationPolicy, GatePolicy, SubmissionMetadata from abevalflow.scorecard import Scorecard, apply_combination_logic @@ -315,12 +316,21 @@ def aggregate_scorecard( has_eval_assets = (submission_dir / "evals" / "evals.json").exists() or (submission_dir / "tests").exists() + operational_limits = certification_policy.get_operational_limits() if certification_policy else None + operational_policy_result = check_operational_policy(submission_dir, limits=operational_limits) + logger.info( + "Operational policy: passed=%s, score=%.3f", + operational_policy_result.passed, + operational_policy_result.score, + ) + certification = compute_certification( gates=gates, validation_passed=validation_passed, metadata_valid=metadata_valid, has_eval_assets=has_eval_assets, policy=certification_policy, + operational_policy_result=operational_policy_result, ) logger.info( "Certification levels: foundational=%s, trusted=%s, certified=%s (highest=%s)", diff --git a/tests/test_certification.py b/tests/test_certification.py index b0461af..471cd76 100644 --- a/tests/test_certification.py +++ b/tests/test_certification.py @@ -347,8 +347,8 @@ def test_trusted_has_expected_checks(self) -> None: assert CheckId.ADVANCED_SECURITY_VALIDATION in TRUSTED_CHECKS assert CheckId.FUNCTIONAL_VALIDATION in TRUSTED_CHECKS assert CheckId.INSTRUCTION_QUALITY in TRUSTED_CHECKS - # registry_governance and operational_policy_compliance not yet implemented - assert len(TRUSTED_CHECKS) == 4 + # registry_governance not yet implemented + assert len(TRUSTED_CHECKS) == 5 def test_certified_has_expected_checks(self) -> None: assert CheckId.ENTERPRISE_STRUCTURE_VALIDATION in CERTIFIED_CHECKS @@ -1116,3 +1116,94 @@ def test_profile_integrates_with_compute_certification(self) -> None: # Should use skill profile's check configuration assert result is not None assert result.foundational is not None + + +class TestOperationalPolicyCertificationIntegration: + def test_passing_check_enables_trusted(self) -> None: + passing_result = CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=True, + score=1.0, + message="All passed", + ) + engine_gate = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + message="Pass", + ) + security_gate = GateResult( + gate_name="security", + gate_type=GateType.SECURITY, + passed=True, + score=1.0, + mode=GateMode.BLOCK, + message="Pass", + findings=[], + ) + quality_gate = GateResult( + gate_name="quality", + gate_type=GateType.QUALITY, + passed=True, + score=0.8, + mode=GateMode.WARN, + message="Pass", + ) + result = compute_certification( + gates=[engine_gate, security_gate, quality_gate], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=passing_result, + ) + assert result.trusted.passed is True + op_check = next(c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE) + assert op_check.passed is True + + def test_failing_check_blocks_trusted(self) -> None: + failing_result = CheckResult( + check_id=CheckId.OPERATIONAL_POLICY_COMPLIANCE, + name="Operational Policy Compliance", + passed=False, + score=0.25, + message="CPU exceeds limit", + ) + engine_gate = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + message="Pass", + ) + result = compute_certification( + gates=[engine_gate], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=failing_result, + ) + assert result.trusted.passed is False + + def test_none_result_uses_default_not_implemented(self) -> None: + engine_gate = GateResult( + gate_name="evaluation", + gate_type=GateType.ENGINE, + passed=True, + score=0.9, + mode=GateMode.BLOCK, + message="Pass", + ) + result = compute_certification( + gates=[engine_gate], + validation_passed=True, + metadata_valid=True, + has_eval_assets=True, + operational_policy_result=None, + ) + op_check = next(c for c in result.trusted.checks if c.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE) + assert op_check.passed is False + assert "not implemented" in op_check.message diff --git a/tests/test_operational_policy.py b/tests/test_operational_policy.py new file mode 100644 index 0000000..8e86a0b --- /dev/null +++ b/tests/test_operational_policy.py @@ -0,0 +1,326 @@ +"""Tests for operational policy compliance checks.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + +from abevalflow.certification import CheckId +from abevalflow.operational_policy import ( + _check_error_handling, + _check_logging_suppression, + _check_resource_declaration, + _check_resource_limits, + _check_timeout_compliance, + check_operational_policy, +) +from abevalflow.schemas import OperationalLimits, SubmissionMetadata + +_DEFAULTS = OperationalLimits() +MAX_CPUS = _DEFAULTS.max_cpus +MAX_MEMORY_MB = _DEFAULTS.max_memory_mb +MAX_AGENT_TIMEOUT_SEC = _DEFAULTS.max_agent_timeout_sec + + +def _make_metadata(**overrides) -> SubmissionMetadata: + base = {"name": "test-submission"} + base.update(overrides) + return SubmissionMetadata(**base) + + +def _write_submission(tmp_path: Path, metadata: dict | None = None, skill_content: str = "", test_content: str = ""): + if metadata is not None: + (tmp_path / "metadata.yaml").write_text(yaml.dump(metadata)) + skills_dir = tmp_path / "skills" + skills_dir.mkdir(exist_ok=True) + if skill_content: + (skills_dir / "SKILL.md").write_text(skill_content) + tests_dir = tmp_path / "tests" + tests_dir.mkdir(exist_ok=True) + if test_content: + (tests_dir / "test_outputs.py").write_text(test_content) + + +class TestResourceDeclaration: + def test_all_declared(self): + data = {"name": "test", "cpus": 2, "memory_mb": 4096, "agent_timeout_sec": 600} + issues = _check_resource_declaration(data) + assert issues == [] + + def test_none_declared(self): + data = {"name": "test"} + issues = _check_resource_declaration(data) + assert len(issues) == 1 + assert "cpus" in issues[0] + assert "memory_mb" in issues[0] + assert "agent_timeout_sec" in issues[0] + + def test_partial_declared(self): + data = {"name": "test", "cpus": 2} + issues = _check_resource_declaration(data) + assert len(issues) == 1 + assert "memory_mb" in issues[0] + assert "agent_timeout_sec" in issues[0] + assert "cpus" not in issues[0] + + +class TestResourceLimits: + def test_default_values_pass(self): + metadata = _make_metadata() + issues = _check_resource_limits(metadata, OperationalLimits()) + assert issues == [] + + def test_within_limits_pass(self): + metadata = _make_metadata(cpus=MAX_CPUS, memory_mb=MAX_MEMORY_MB) + issues = _check_resource_limits(metadata, OperationalLimits()) + assert issues == [] + + def test_cpu_exceeds_limit(self): + metadata = _make_metadata(cpus=MAX_CPUS + 1) + issues = _check_resource_limits(metadata, OperationalLimits()) + assert len(issues) == 1 + assert "CPU" in issues[0] + + def test_memory_exceeds_limit(self): + metadata = _make_metadata(memory_mb=MAX_MEMORY_MB + 1) + issues = _check_resource_limits(metadata, OperationalLimits()) + assert len(issues) == 1 + assert "Memory" in issues[0] + + def test_both_exceed(self): + metadata = _make_metadata(cpus=MAX_CPUS + 1, memory_mb=MAX_MEMORY_MB + 1) + issues = _check_resource_limits(metadata, OperationalLimits()) + assert len(issues) == 2 + + +class TestTimeoutCompliance: + def test_default_timeout_passes(self): + metadata = _make_metadata() + issues = _check_timeout_compliance(metadata, OperationalLimits()) + assert issues == [] + + def test_within_limit_passes(self): + metadata = _make_metadata(agent_timeout_sec=MAX_AGENT_TIMEOUT_SEC) + issues = _check_timeout_compliance(metadata, OperationalLimits()) + assert issues == [] + + def test_exceeds_limit(self): + metadata = _make_metadata(agent_timeout_sec=MAX_AGENT_TIMEOUT_SEC + 1) + issues = _check_timeout_compliance(metadata, OperationalLimits()) + assert len(issues) == 1 + assert "timeout" in issues[0].lower() + + +class TestErrorHandling: + def test_no_test_file(self, tmp_path: Path): + issues = _check_error_handling(tmp_path / "nonexistent.py") + assert issues == [] + + def test_clean_test_file(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except ValueError as e:\n" + " raise AssertionError(f'Failed: {e}')\n" + ) + issues = _check_error_handling(test_file) + assert issues == [] + + def test_bare_except_pass(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text("def test_something():\n try:\n result = run()\n except:\n pass\n") + issues = _check_error_handling(test_file) + assert len(issues) == 1 + assert "Bare" in issues[0] + + def test_except_exception_pass(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n try:\n result = run()\n except Exception:\n pass\n" + ) + issues = _check_error_handling(test_file) + assert len(issues) == 1 + assert "Exception" in issues[0] + + def test_except_with_body_is_ok(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text( + "def test_something():\n" + " try:\n" + " result = run()\n" + " except Exception:\n" + " print('error')\n" + " raise\n" + ) + issues = _check_error_handling(test_file) + assert issues == [] + + def test_except_ellipsis(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text("def test_something():\n try:\n result = run()\n except:\n ...\n") + issues = _check_error_handling(test_file) + assert len(issues) == 1 + + def test_syntax_error_is_ignored(self, tmp_path: Path): + test_file = tmp_path / "test_outputs.py" + test_file.write_text("def broken(:\n") + issues = _check_error_handling(test_file) + assert issues == [] + + +class TestLoggingSuppression: + def test_no_skill_file(self, tmp_path: Path): + issues = _check_logging_suppression(tmp_path / "nonexistent.md") + assert issues == [] + + def test_clean_skill(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nThis skill helps with code review.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + def test_do_not_log(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDo not log any output.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_disable_logging(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nPlease disable logging for this task.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_suppress_output(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nSuppress output from the agent.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_case_insensitive(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDISABLE LOGGING immediately.\n") + issues = _check_logging_suppression(skill_file) + assert len(issues) == 1 + + def test_security_qualified_not_flagged(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDo not log passwords or PII.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + def test_security_qualified_credentials(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nNo logging of credentials.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + def test_security_qualified_tokens(self, tmp_path: Path): + skill_file = tmp_path / "SKILL.md" + skill_file.write_text("# My Skill\n\nDo not log sensitive tokens.\n") + issues = _check_logging_suppression(skill_file) + assert issues == [] + + +class TestCheckOperationalPolicy: + def test_all_pass(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "good-skill", "cpus": 2, "memory_mb": 4096, "agent_timeout_sec": 600}, + skill_content="# Good Skill\n\nHelps with testing.\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.check_id == CheckId.OPERATIONAL_POLICY_COMPLIANCE + assert result.passed is True + assert result.score == 1.0 + + def test_resource_limit_failure(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "heavy-skill", "cpus": MAX_CPUS + 1}, + skill_content="# Skill\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert "CPU" in result.message + + def test_error_handling_failure(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "bad-tests"}, + skill_content="# Skill\n", + test_content="def test_bad():\n try:\n run()\n except:\n pass\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert "except" in result.message.lower() or "Bare" in result.message + + def test_logging_suppression_failure(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "noisy-skill"}, + skill_content="# Skill\n\nDo not log any output.\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert "log" in result.message.lower() + + def test_no_metadata_still_checks_files(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata=None, + skill_content="# Skill\n\nDisable logging please.\n", + test_content="def test_ok():\n assert True\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + + def test_empty_submission(self, tmp_path: Path): + result = check_operational_policy(tmp_path) + assert result.passed is True + assert result.score == 1.0 + + def test_score_degrades_with_issues(self, tmp_path: Path): + _write_submission( + tmp_path, + metadata={"name": "bad", "cpus": MAX_CPUS + 1, "memory_mb": MAX_MEMORY_MB + 1}, + skill_content="# Skill\n\nDo not log output.\n", + test_content="def test_bad():\n try:\n x()\n except:\n pass\n", + ) + result = check_operational_policy(tmp_path) + assert result.passed is False + assert result.score < 1.0 + + +class TestCustomLimits: + _full_resources = {"cpus": 2, "memory_mb": 4096, "agent_timeout_sec": 600} + + def test_custom_limits_allow_higher_cpu(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "cpus": 8} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_cpus=10)) + assert result.passed is True + + def test_custom_limits_still_enforce(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "cpus": 8} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_cpus=4)) + assert result.passed is False + + def test_custom_memory_limit(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "memory_mb": 16384} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_memory_mb=32768)) + assert result.passed is True + + def test_custom_timeout_limit(self, tmp_path: Path): + meta = {"name": "big-skill", **self._full_resources, "agent_timeout_sec": 7200} + _write_submission(tmp_path, metadata=meta) + result = check_operational_policy(tmp_path, limits=OperationalLimits(max_agent_timeout_sec=10000)) + assert result.passed is True