mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-09-04 04:01:07 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Zohaib Hassnain
|
98900af751 | ||
|
Zohaib Hassnain
|
8bceff105c | ||
|
Zohaib Hassnain
|
b574e2e6b4 |
@@ -22,7 +22,6 @@ icon: "sitemap"
|
||||
| `LLMOntologyGenerator` | LLM-powered ontology generation for complex domains |
|
||||
| `SHACLGenerator` | Generate SHACL shapes from an ontology or KG schema |
|
||||
| `OntologyValidator` | Validate any graph against SHACL shapes: returns `SHACLValidationReport` |
|
||||
| `OntologyQualityGate` | Run deterministic ontology/KG quality checks for CI |
|
||||
| `OWLGenerator` | Serialize ontologies to Turtle, RDF/XML, JSON-LD |
|
||||
| `NamespaceManager` | IRI generation, prefix management, and namespace binding |
|
||||
| `OntologyEvaluator` | Coverage, completeness, and granularity quality metrics |
|
||||
@@ -82,38 +81,9 @@ engine.export_owl(ontology, "ontology.ttl", format="turtle")
|
||||
| :------ | :----------- |
|
||||
| `from_data(data)` | Run the 5-stage pipeline on entity/relationship data |
|
||||
| `validate_graph(kg, ontology=...)` | Check a knowledge graph against generated SHACL shapes |
|
||||
| `quality_check(ontology, graph_data=...)` | Return a deterministic quality report and CI-friendly pass/fail result |
|
||||
| `export_owl(ontology, path, format)` | Serialize to `"turtle"`, `"xml"`, or `"json-ld"` |
|
||||
| `evaluate(ontology, kg)` | Compute coverage, completeness, and granularity metrics |
|
||||
|
||||
### Ontology Quality Gate
|
||||
|
||||
Use the quality gate before export or deployment to catch structural issues
|
||||
without adding a runtime dependency:
|
||||
|
||||
```python
|
||||
from semantica.ontology import ontology_quality_check
|
||||
|
||||
report = ontology_quality_check(
|
||||
ontology,
|
||||
graph_data=kg,
|
||||
thresholds={"min_coverage": 0.8},
|
||||
)
|
||||
|
||||
if not report.passed:
|
||||
for issue in report.issues:
|
||||
print(issue.code, issue.message)
|
||||
```
|
||||
|
||||
The report checks class/property coverage, orphan schema elements, domain and
|
||||
range references, and unresolved KG relationship endpoints. It includes
|
||||
machine-readable issue codes, severity, counts, metrics, and threshold
|
||||
failures. The first version reports findings only; it does not auto-fix data.
|
||||
|
||||
### Thresholds
|
||||
|
||||
`min_coverage` (default `0.0`) sets the minimum required `coverage` score, the average of class and property coverage from `0.0` to `1.0`; the gate fails below it. `max_errors` (default `0.0`) caps how many `error`/`critical` issues are allowed before the gate fails. `max_warnings` (default `None`) caps `warning` issues the same way, and `None` means warnings alone never fail the gate. `fail_on_warnings` is a separate parameter, not a `thresholds` key, passed to `OntologyQualityGate(...)` or `.check(...)` directly; when `True`, a single warning fails the gate regardless of `max_warnings`.
|
||||
|
||||
## OntologyGenerator (5-Stage Pipeline)
|
||||
|
||||
**`OntologyGenerator`** auto-generates a formal ontology from your knowledge graph entities and relationships:
|
||||
|
||||
@@ -301,10 +301,7 @@ class GraphValidator:
|
||||
code="ORPHAN_NODES",
|
||||
message=f"Found {len(isolates)} orphan nodes (no relationships).",
|
||||
severity=ValidationSeverity.WARNING,
|
||||
details={
|
||||
"count": len(isolates),
|
||||
"ids": sorted(isolates, key=str)[:10],
|
||||
} # Limit output deterministically
|
||||
details={"count": len(isolates), "ids": isolates[:10]} # Limit output
|
||||
))
|
||||
|
||||
except Exception as e:
|
||||
|
||||
@@ -162,13 +162,6 @@ from .ontology_validator import (
|
||||
run_shacl_validation,
|
||||
validate_ontology,
|
||||
)
|
||||
from .quality_gate import (
|
||||
OntologyQualityGate,
|
||||
OntologyQualityReport,
|
||||
QualityIssue,
|
||||
QualitySeverity,
|
||||
ontology_quality_check,
|
||||
)
|
||||
from .owl_generator import OWLGenerator
|
||||
from .property_generator import PropertyGenerator
|
||||
from .registry import MethodRegistry, method_registry
|
||||
@@ -201,11 +194,6 @@ __all__ = [
|
||||
"SHACLValidationReport",
|
||||
"SHACLViolation",
|
||||
"run_shacl_validation",
|
||||
"OntologyQualityGate",
|
||||
"OntologyQualityReport",
|
||||
"QualityIssue",
|
||||
"QualitySeverity",
|
||||
"ontology_quality_check",
|
||||
# OWL/RDF generation
|
||||
"OWLGenerator",
|
||||
# Requirements and competency questions
|
||||
|
||||
@@ -9,7 +9,6 @@ from .property_generator import PropertyGenerator
|
||||
from .owl_generator import OWLGenerator
|
||||
from .ontology_evaluator import OntologyEvaluator
|
||||
from .ontology_validator import OntologyValidator
|
||||
from .quality_gate import OntologyQualityGate, OntologyQualityReport
|
||||
from .llm_generator import LLMOntologyGenerator
|
||||
from ..semantic_extract.triplet_extractor import Triplet
|
||||
|
||||
@@ -26,9 +25,6 @@ class OntologyEngine:
|
||||
self.owl = OWLGenerator(**config)
|
||||
self.evaluator = OntologyEvaluator(**config)
|
||||
self.validator = OntologyValidator(**config)
|
||||
self.quality_gate = OntologyQualityGate(
|
||||
validator=self.validator, evaluator=self.evaluator
|
||||
)
|
||||
self.llm = LLMOntologyGenerator(**config)
|
||||
self.store = config.get("store")
|
||||
|
||||
@@ -597,15 +593,6 @@ class OntologyEngine:
|
||||
def validate(self, ontology: Dict[str, Any], **options):
|
||||
return self.validator.validate(ontology, **options)
|
||||
|
||||
def quality_check(
|
||||
self,
|
||||
ontology: Dict[str, Any],
|
||||
graph_data: Optional[Dict[str, Any]] = None,
|
||||
**options,
|
||||
) -> OntologyQualityReport:
|
||||
"""Run deterministic ontology quality checks suitable for CI."""
|
||||
return self.quality_gate.check(ontology, graph_data=graph_data, **options)
|
||||
|
||||
def to_owl(self, ontology: Dict[str, Any], format: str = "turtle", **options):
|
||||
return self.owl.generate_owl(ontology, format=format, **options)
|
||||
|
||||
|
||||
@@ -1,855 +0,0 @@
|
||||
"""Deterministic quality checks for ontology and KG pipelines."""
|
||||
|
||||
import copy
|
||||
import math
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Set, Tuple
|
||||
|
||||
from ..kg.graph_validator import GraphValidator
|
||||
from .ontology_evaluator import OntologyEvaluator
|
||||
from .ontology_validator import OntologyValidator
|
||||
|
||||
|
||||
class QualitySeverity(str, Enum):
|
||||
"""Severity assigned to a quality finding."""
|
||||
|
||||
INFO = "info"
|
||||
WARNING = "warning"
|
||||
ERROR = "error"
|
||||
CRITICAL = "critical"
|
||||
|
||||
|
||||
@dataclass
|
||||
class QualityIssue:
|
||||
"""A single machine-readable ontology quality finding."""
|
||||
|
||||
code: str
|
||||
message: str
|
||||
severity: QualitySeverity
|
||||
element_id: Optional[str] = None
|
||||
element_type: Optional[str] = None
|
||||
details: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Return a JSON-friendly representation of the issue."""
|
||||
return {
|
||||
"code": self.code,
|
||||
"message": self.message,
|
||||
"severity": self.severity.value,
|
||||
"element_id": self.element_id,
|
||||
"element_type": self.element_type,
|
||||
"details": self.details,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class OntologyQualityReport:
|
||||
"""Result returned by :class:`OntologyQualityGate`."""
|
||||
|
||||
passed: bool
|
||||
issues: List[QualityIssue] = field(default_factory=list)
|
||||
stats: Dict[str, int] = field(default_factory=dict)
|
||||
metrics: Dict[str, float] = field(default_factory=dict)
|
||||
thresholds: Dict[str, Optional[float]] = field(default_factory=dict)
|
||||
threshold_failures: List[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def error_count(self) -> int:
|
||||
"""Number of error and critical findings."""
|
||||
return sum(
|
||||
issue.severity in (QualitySeverity.ERROR, QualitySeverity.CRITICAL)
|
||||
for issue in self.issues
|
||||
)
|
||||
|
||||
@property
|
||||
def warning_count(self) -> int:
|
||||
"""Number of warning findings."""
|
||||
return sum(issue.severity == QualitySeverity.WARNING for issue in self.issues)
|
||||
|
||||
@property
|
||||
def info_count(self) -> int:
|
||||
"""Number of informational findings."""
|
||||
return sum(issue.severity == QualitySeverity.INFO for issue in self.issues)
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""Return a JSON-friendly representation of the report."""
|
||||
return {
|
||||
"passed": self.passed,
|
||||
"issues": [issue.to_dict() for issue in self.issues],
|
||||
"stats": {
|
||||
**self.stats,
|
||||
"issues": len(self.issues),
|
||||
"errors": self.error_count,
|
||||
"warnings": self.warning_count,
|
||||
"infos": self.info_count,
|
||||
},
|
||||
"metrics": self.metrics,
|
||||
"thresholds": self.thresholds,
|
||||
"threshold_failures": self.threshold_failures,
|
||||
}
|
||||
|
||||
|
||||
class OntologyQualityGate:
|
||||
"""Run deterministic, CI-friendly ontology quality checks."""
|
||||
|
||||
DEFAULT_THRESHOLDS: Dict[str, Optional[float]] = {
|
||||
"min_coverage": 0.0,
|
||||
"max_errors": 0.0,
|
||||
"max_warnings": None,
|
||||
}
|
||||
_DATA_PROPERTY_TYPES = {"data", "datatype", "data_property", "literal"}
|
||||
_OBJECT_PROPERTY_TYPES = {"object", "object_property", "relationship"}
|
||||
_KNOWN_DATATYPES = {
|
||||
"string",
|
||||
"boolean",
|
||||
"decimal",
|
||||
"float",
|
||||
"double",
|
||||
"integer",
|
||||
"int",
|
||||
"long",
|
||||
"short",
|
||||
"byte",
|
||||
"date",
|
||||
"datetime",
|
||||
"datetimestamp",
|
||||
"time",
|
||||
"duration",
|
||||
"anyuri",
|
||||
}
|
||||
_BUILTIN_CLASSES = {
|
||||
"owl:thing",
|
||||
"rdfs:resource",
|
||||
"http://www.w3.org/2002/07/owl#thing",
|
||||
"http://www.w3.org/2000/01/rdf-schema#resource",
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
validator: Optional[OntologyValidator] = None,
|
||||
evaluator: Optional[OntologyEvaluator] = None,
|
||||
thresholds: Optional[Mapping[str, Optional[float]]] = None,
|
||||
fail_on_warnings: bool = False,
|
||||
) -> None:
|
||||
self.validator = validator or OntologyValidator()
|
||||
self.evaluator = evaluator or OntologyEvaluator()
|
||||
self.thresholds = dict(self.DEFAULT_THRESHOLDS)
|
||||
if thresholds:
|
||||
self.thresholds.update(thresholds)
|
||||
self.fail_on_warnings = fail_on_warnings
|
||||
|
||||
def check(
|
||||
self,
|
||||
ontology: Any,
|
||||
graph_data: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
thresholds: Optional[Mapping[str, Optional[float]]] = None,
|
||||
fail_on_warnings: Optional[bool] = None,
|
||||
competency_questions: Optional[List[str]] = None,
|
||||
) -> OntologyQualityReport:
|
||||
"""Check an ontology and optionally its instance graph.
|
||||
|
||||
``graph_data`` is optional because an ontology can be checked before
|
||||
instances are available. When omitted, embedded ``entities`` and
|
||||
``relationships`` are checked when present.
|
||||
"""
|
||||
active_thresholds = dict(self.thresholds)
|
||||
if thresholds:
|
||||
active_thresholds.update(thresholds)
|
||||
min_coverage, max_errors, max_warnings = self._validate_thresholds(
|
||||
active_thresholds
|
||||
)
|
||||
should_fail_on_warnings = (
|
||||
self.fail_on_warnings if fail_on_warnings is None else fail_on_warnings
|
||||
)
|
||||
issues: List[QualityIssue] = []
|
||||
|
||||
if not isinstance(ontology, dict):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"INVALID_ONTOLOGY",
|
||||
"Ontology must be a dictionary.",
|
||||
QualitySeverity.CRITICAL,
|
||||
element_type="ontology",
|
||||
)
|
||||
return self._build_report(
|
||||
issues,
|
||||
classes=0,
|
||||
properties=0,
|
||||
entities=0,
|
||||
relationships=0,
|
||||
metrics={
|
||||
"coverage": 0.0,
|
||||
"class_coverage": 0.0,
|
||||
"property_coverage": 0.0,
|
||||
},
|
||||
thresholds=active_thresholds,
|
||||
min_coverage=min_coverage,
|
||||
max_errors=max_errors,
|
||||
max_warnings=max_warnings,
|
||||
fail_on_warnings=should_fail_on_warnings,
|
||||
)
|
||||
|
||||
validation = self.validator.validate(ontology)
|
||||
for message in getattr(validation, "errors", []) or []:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"VALIDATOR_ERROR",
|
||||
str(message),
|
||||
QualitySeverity.ERROR,
|
||||
element_type="ontology",
|
||||
)
|
||||
for message in getattr(validation, "warnings", []) or []:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"VALIDATOR_WARNING",
|
||||
str(message),
|
||||
QualitySeverity.WARNING,
|
||||
element_type="ontology",
|
||||
)
|
||||
|
||||
classes = self._read_collection(ontology, "classes", issues)
|
||||
properties = self._read_collection(ontology, "properties", issues)
|
||||
class_aliases, class_ids = self._index_elements(
|
||||
classes, "class", "MISSING_CLASS_ID", issues
|
||||
)
|
||||
referenced_classes: Set[str] = set()
|
||||
self._mark_hierarchy(classes, class_aliases, referenced_classes)
|
||||
|
||||
property_with_endpoints = 0
|
||||
for index, prop in enumerate(properties):
|
||||
if not isinstance(prop, dict):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"INVALID_PROPERTY",
|
||||
f"Property at index {index} must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="property",
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
|
||||
prop_id = self._identifier(prop)
|
||||
if prop_id is None:
|
||||
prop_id = f"property[{index}]"
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_PROPERTY_ID",
|
||||
f"Property at index {index} has no name or URI.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="property",
|
||||
details={"index": index},
|
||||
)
|
||||
|
||||
raw_prop_type = prop.get("type")
|
||||
prop_type = (
|
||||
str(raw_prop_type).strip().lower() if raw_prop_type is not None else ""
|
||||
)
|
||||
if not prop_type:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_PROPERTY_TYPE",
|
||||
f"Property '{prop_id}' has no type.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
elif (
|
||||
prop_type not in self._DATA_PROPERTY_TYPES | self._OBJECT_PROPERTY_TYPES
|
||||
):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_PROPERTY_TYPE",
|
||||
f"Property '{prop_id}' has unknown type '{prop_type}'.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
|
||||
domains = self._values(prop, "domain")
|
||||
ranges = self._values(prop, "range")
|
||||
if domains or ranges:
|
||||
property_with_endpoints += 1
|
||||
else:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"ORPHAN_PROPERTY",
|
||||
f"Property '{prop_id}' has no domain or range.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
if not domains:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_DOMAIN",
|
||||
f"Property '{prop_id}' has no domain.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
for domain in domains:
|
||||
matched = self._match_class(domain, class_aliases)
|
||||
if matched:
|
||||
referenced_classes.add(matched)
|
||||
elif not self._is_builtin_class(domain):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_DOMAIN",
|
||||
f"Property '{prop_id}' references unknown domain '{domain}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"domain": domain},
|
||||
)
|
||||
|
||||
if not ranges:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"MISSING_RANGE",
|
||||
f"Property '{prop_id}' has no range.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
)
|
||||
for range_value in ranges:
|
||||
matched = self._match_class(range_value, class_aliases)
|
||||
if matched:
|
||||
referenced_classes.add(matched)
|
||||
self._check_range(
|
||||
prop_id,
|
||||
prop_type,
|
||||
range_value,
|
||||
matched is not None,
|
||||
issues,
|
||||
)
|
||||
|
||||
graph = graph_data
|
||||
if graph is None and ("entities" in ontology or "relationships" in ontology):
|
||||
graph = ontology
|
||||
graph_entities, graph_relationships = self._read_graph(graph, issues)
|
||||
if self._has_valid_graph_shape(graph):
|
||||
self._check_graph(graph, issues)
|
||||
self._mark_graph_types(graph_entities, class_aliases, referenced_classes)
|
||||
|
||||
for class_id in class_ids:
|
||||
if class_id not in referenced_classes:
|
||||
self._add_issue(
|
||||
issues,
|
||||
"ORPHAN_CLASS",
|
||||
f"Class '{class_id}' is not connected to a property, hierarchy, or graph entity type.",
|
||||
QualitySeverity.WARNING,
|
||||
element_id=class_id,
|
||||
element_type="class",
|
||||
)
|
||||
|
||||
class_coverage = (
|
||||
len(referenced_classes & set(class_ids)) / len(class_ids)
|
||||
if class_ids
|
||||
else 0.0
|
||||
)
|
||||
property_coverage = (
|
||||
property_with_endpoints / len(properties) if properties else 1.0
|
||||
)
|
||||
metrics: Dict[str, float] = {
|
||||
"coverage": (
|
||||
(class_coverage + property_coverage) / 2
|
||||
if classes or properties
|
||||
else 0.0
|
||||
),
|
||||
"class_coverage": class_coverage,
|
||||
"property_coverage": property_coverage,
|
||||
"validator_valid": 1.0 if getattr(validation, "valid", True) else 0.0,
|
||||
}
|
||||
if competency_questions is not None:
|
||||
evaluation_ontology = self._prepare_for_evaluation(
|
||||
ontology, classes, properties
|
||||
)
|
||||
evaluation = self._evaluate_competency_questions(
|
||||
evaluation_ontology, competency_questions=competency_questions
|
||||
)
|
||||
metrics["competency_question_coverage"] = evaluation.coverage_score
|
||||
metrics["completeness"] = evaluation.completeness_score
|
||||
|
||||
return self._build_report(
|
||||
issues,
|
||||
classes=len(classes),
|
||||
properties=len(properties),
|
||||
entities=len(graph_entities),
|
||||
relationships=len(graph_relationships),
|
||||
metrics=metrics,
|
||||
thresholds=active_thresholds,
|
||||
min_coverage=min_coverage,
|
||||
max_errors=max_errors,
|
||||
max_warnings=max_warnings,
|
||||
fail_on_warnings=should_fail_on_warnings,
|
||||
)
|
||||
|
||||
def _check_range(
|
||||
self,
|
||||
prop_id: str,
|
||||
prop_type: str,
|
||||
range_value: Any,
|
||||
is_class: bool,
|
||||
issues: List[QualityIssue],
|
||||
) -> None:
|
||||
if prop_type in self._DATA_PROPERTY_TYPES:
|
||||
if not self._is_known_datatype(range_value):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"INVALID_DATATYPE_RANGE",
|
||||
f"Data property '{prop_id}' has invalid range '{range_value}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"range": range_value},
|
||||
)
|
||||
elif prop_type in self._OBJECT_PROPERTY_TYPES:
|
||||
if not is_class and not self._is_builtin_class(range_value):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_RANGE",
|
||||
f"Object property '{prop_id}' references unknown range '{range_value}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"range": range_value},
|
||||
)
|
||||
elif (
|
||||
not is_class
|
||||
and not self._is_builtin_class(range_value)
|
||||
and not self._is_known_datatype(range_value)
|
||||
):
|
||||
self._add_issue(
|
||||
issues,
|
||||
"UNKNOWN_RANGE",
|
||||
f"Property '{prop_id}' references unknown range '{range_value}'.",
|
||||
QualitySeverity.ERROR,
|
||||
element_id=prop_id,
|
||||
element_type="property",
|
||||
details={"range": range_value},
|
||||
)
|
||||
|
||||
def _build_report(
|
||||
self,
|
||||
issues: List[QualityIssue],
|
||||
*,
|
||||
classes: int,
|
||||
properties: int,
|
||||
entities: int,
|
||||
relationships: int,
|
||||
metrics: Dict[str, float],
|
||||
thresholds: Dict[str, Optional[float]],
|
||||
min_coverage: float,
|
||||
max_errors: float,
|
||||
max_warnings: Optional[float],
|
||||
fail_on_warnings: bool,
|
||||
) -> OntologyQualityReport:
|
||||
failures: List[str] = []
|
||||
error_count = sum(
|
||||
issue.severity in (QualitySeverity.ERROR, QualitySeverity.CRITICAL)
|
||||
for issue in issues
|
||||
)
|
||||
warning_count = sum(
|
||||
issue.severity == QualitySeverity.WARNING for issue in issues
|
||||
)
|
||||
if error_count > max_errors:
|
||||
failures.append("max_errors")
|
||||
if max_warnings is not None and warning_count > max_warnings:
|
||||
failures.append("max_warnings")
|
||||
if fail_on_warnings and warning_count:
|
||||
failures.append("fail_on_warnings")
|
||||
if metrics.get("coverage", 0.0) < min_coverage:
|
||||
failures.append("min_coverage")
|
||||
return OntologyQualityReport(
|
||||
passed=not failures,
|
||||
issues=issues,
|
||||
stats={
|
||||
"classes": classes,
|
||||
"properties": properties,
|
||||
"entities": entities,
|
||||
"relationships": relationships,
|
||||
},
|
||||
metrics=metrics,
|
||||
thresholds=thresholds,
|
||||
threshold_failures=failures,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _validate_thresholds(
|
||||
thresholds: Mapping[str, Optional[float]],
|
||||
) -> Tuple[float, float, Optional[float]]:
|
||||
min_coverage = float(thresholds.get("min_coverage", 0.0) or 0.0)
|
||||
max_errors = float(thresholds.get("max_errors", 0.0) or 0.0)
|
||||
warning_value = thresholds.get("max_warnings")
|
||||
max_warnings = None if warning_value is None else float(warning_value)
|
||||
if not all(
|
||||
math.isfinite(value)
|
||||
for value in (min_coverage, max_errors)
|
||||
if value is not None
|
||||
) or (max_warnings is not None and not math.isfinite(max_warnings)):
|
||||
raise ValueError("quality thresholds must be finite numbers")
|
||||
if not 0.0 <= min_coverage <= 1.0:
|
||||
raise ValueError("min_coverage must be between 0.0 and 1.0")
|
||||
if max_errors < 0 or (max_warnings is not None and max_warnings < 0):
|
||||
raise ValueError("error and warning thresholds cannot be negative")
|
||||
return min_coverage, max_errors, max_warnings
|
||||
|
||||
@classmethod
|
||||
def _prepare_for_evaluation(
|
||||
cls,
|
||||
ontology: Dict[str, Any],
|
||||
classes: Iterable[Any],
|
||||
properties: Iterable[Any],
|
||||
) -> Dict[str, Any]:
|
||||
"""Make a shallow, evaluator-safe view without changing caller data."""
|
||||
prepared = dict(ontology)
|
||||
prepared["classes"] = [
|
||||
cls._prepare_element(element)
|
||||
for element in classes
|
||||
if isinstance(element, dict)
|
||||
]
|
||||
prepared["properties"] = [
|
||||
cls._prepare_element(element)
|
||||
for element in properties
|
||||
if isinstance(element, dict)
|
||||
]
|
||||
return prepared
|
||||
|
||||
@classmethod
|
||||
def _prepare_element(cls, element: Dict[str, Any]) -> Dict[str, Any]:
|
||||
prepared = dict(element)
|
||||
identifier = cls._identifier(prepared)
|
||||
if identifier is not None and not str(prepared.get("name", "")).strip():
|
||||
prepared["name"] = identifier
|
||||
return prepared
|
||||
|
||||
def _evaluate_competency_questions(
|
||||
self, ontology: Dict[str, Any], competency_questions: List[str]
|
||||
) -> Any:
|
||||
"""Evaluate with an isolated question manager for repeatable checks."""
|
||||
evaluator = copy.copy(self.evaluator)
|
||||
manager = getattr(self.evaluator, "competency_questions_manager", None)
|
||||
if manager is not None and hasattr(manager, "questions"):
|
||||
isolated_manager = copy.copy(manager)
|
||||
isolated_manager.questions = []
|
||||
evaluator.competency_questions_manager = isolated_manager
|
||||
return evaluator.evaluate_ontology(
|
||||
ontology, competency_questions=competency_questions
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _read_collection(
|
||||
cls, ontology: Dict[str, Any], key: str, issues: List[QualityIssue]
|
||||
) -> List[Any]:
|
||||
if key not in ontology:
|
||||
cls._add_issue(
|
||||
issues,
|
||||
f"MISSING_{key.upper()}",
|
||||
f"Ontology has no {key} defined.",
|
||||
QualitySeverity.WARNING,
|
||||
element_type="ontology",
|
||||
)
|
||||
return []
|
||||
value = ontology[key]
|
||||
if not isinstance(value, list):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
f"INVALID_{key.upper()}",
|
||||
f"Ontology '{key}' must be a list.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="ontology",
|
||||
)
|
||||
return []
|
||||
return value
|
||||
|
||||
@classmethod
|
||||
def _index_elements(
|
||||
cls,
|
||||
elements: Iterable[Any],
|
||||
element_type: str,
|
||||
missing_code: str,
|
||||
issues: List[QualityIssue],
|
||||
) -> Tuple[Dict[str, str], List[str]]:
|
||||
aliases: Dict[str, str] = {}
|
||||
identifiers: List[str] = []
|
||||
for index, element in enumerate(elements):
|
||||
identifier = cls._identifier(element)
|
||||
if identifier is None:
|
||||
cls._add_issue(
|
||||
issues,
|
||||
missing_code,
|
||||
f"{element_type.title()} at index {index} has no name or URI.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type=element_type,
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
identifiers.append(identifier)
|
||||
aliases_for_element: Set[str] = set()
|
||||
for candidate in cls._identifiers(element):
|
||||
aliases_for_element.update(cls._term_aliases(candidate))
|
||||
for alias in sorted(aliases_for_element):
|
||||
aliases.setdefault(alias, identifier)
|
||||
return aliases, identifiers
|
||||
|
||||
@classmethod
|
||||
def _mark_hierarchy(
|
||||
cls,
|
||||
classes: Iterable[Any],
|
||||
aliases: Mapping[str, str],
|
||||
referenced: Set[str],
|
||||
) -> None:
|
||||
for class_entry in classes:
|
||||
if not isinstance(class_entry, dict):
|
||||
continue
|
||||
identifier = cls._identifier(class_entry)
|
||||
for key in (
|
||||
"subClassOf",
|
||||
"subclassOf",
|
||||
"parent",
|
||||
"superclass",
|
||||
"superclasses",
|
||||
):
|
||||
parents = cls._values(class_entry, key)
|
||||
if parents and identifier:
|
||||
referenced.add(identifier)
|
||||
for parent in parents:
|
||||
matched = cls._match_class(parent, aliases)
|
||||
if matched:
|
||||
referenced.add(matched)
|
||||
|
||||
@classmethod
|
||||
def _mark_graph_types(
|
||||
cls,
|
||||
entities: Iterable[Any],
|
||||
aliases: Mapping[str, str],
|
||||
referenced: Set[str],
|
||||
) -> None:
|
||||
for entity in entities:
|
||||
if not isinstance(entity, dict):
|
||||
continue
|
||||
entity_type = entity.get("type") or entity.get("entity_type")
|
||||
matched = cls._match_class(entity_type, aliases)
|
||||
if matched:
|
||||
referenced.add(matched)
|
||||
|
||||
@classmethod
|
||||
def _check_graph(cls, graph: Dict[str, Any], issues: List[QualityIssue]) -> None:
|
||||
safe_graph = cls._prepare_graph_for_validation(graph, issues)
|
||||
result = GraphValidator().validate(safe_graph)
|
||||
code_map = {
|
||||
"DANGLING_EDGE": "UNRESOLVED_RELATIONSHIP_ENDPOINT",
|
||||
"ORPHAN_NODES": "ORPHAN_ENTITY",
|
||||
}
|
||||
severity_map = {
|
||||
"info": QualitySeverity.INFO,
|
||||
"warning": QualitySeverity.WARNING,
|
||||
"error": QualitySeverity.ERROR,
|
||||
"critical": QualitySeverity.CRITICAL,
|
||||
}
|
||||
for graph_issue in result.issues:
|
||||
severity = severity_map.get(
|
||||
graph_issue.severity.value, QualitySeverity.ERROR
|
||||
)
|
||||
details = dict(graph_issue.details or {})
|
||||
if graph_issue.code == "ORPHAN_NODES" and "ids" in details:
|
||||
details["ids"] = sorted(details["ids"], key=str)
|
||||
cls._add_issue(
|
||||
issues,
|
||||
code_map.get(graph_issue.code, graph_issue.code),
|
||||
graph_issue.message,
|
||||
severity,
|
||||
element_id=graph_issue.element_id,
|
||||
element_type=graph_issue.element_type,
|
||||
details=details,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _prepare_graph_for_validation(
|
||||
cls, graph: Dict[str, Any], issues: List[QualityIssue]
|
||||
) -> Dict[str, List[Dict[str, Any]]]:
|
||||
"""Normalize supported graph aliases and isolate malformed members."""
|
||||
entities: List[Dict[str, Any]] = []
|
||||
raw_entities = graph.get("entities", [])
|
||||
raw_relationships = graph.get("relationships", [])
|
||||
for index, entity in enumerate(
|
||||
raw_entities if isinstance(raw_entities, list) else []
|
||||
):
|
||||
if not isinstance(entity, dict):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_ENTITY",
|
||||
f"Entity at index {index} must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="entity",
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
normalized = dict(entity)
|
||||
if not normalized.get("name") and normalized.get("text") is not None:
|
||||
normalized["name"] = normalized["text"]
|
||||
entities.append(normalized)
|
||||
|
||||
relationships: List[Dict[str, Any]] = []
|
||||
for index, relationship in enumerate(
|
||||
raw_relationships if isinstance(raw_relationships, list) else []
|
||||
):
|
||||
if not isinstance(relationship, dict):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_RELATIONSHIP",
|
||||
f"Relationship at index {index} must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="relationship",
|
||||
details={"index": index},
|
||||
)
|
||||
continue
|
||||
relationships.append(dict(relationship))
|
||||
|
||||
return {"entities": entities, "relationships": relationships}
|
||||
|
||||
@classmethod
|
||||
def _read_graph(
|
||||
cls, graph: Optional[Dict[str, Any]], issues: List[QualityIssue]
|
||||
) -> Tuple[List[Any], List[Any]]:
|
||||
if graph is None:
|
||||
return [], []
|
||||
if not isinstance(graph, dict):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_GRAPH",
|
||||
"Graph data must be a dictionary.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="graph",
|
||||
)
|
||||
return [], []
|
||||
entities = graph.get("entities", [])
|
||||
relationships = graph.get("relationships", [])
|
||||
if not isinstance(entities, list) or not isinstance(relationships, list):
|
||||
cls._add_issue(
|
||||
issues,
|
||||
"INVALID_GRAPH",
|
||||
"Graph entities and relationships must be lists.",
|
||||
QualitySeverity.ERROR,
|
||||
element_type="graph",
|
||||
)
|
||||
return [], []
|
||||
return entities, relationships
|
||||
|
||||
@staticmethod
|
||||
def _has_valid_graph_shape(graph: Optional[Dict[str, Any]]) -> bool:
|
||||
return bool(
|
||||
isinstance(graph, dict)
|
||||
and isinstance(graph.get("entities", []), list)
|
||||
and isinstance(graph.get("relationships", []), list)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _identifier(element: Any) -> Optional[str]:
|
||||
identifiers = OntologyQualityGate._identifiers(element)
|
||||
return identifiers[0] if identifiers else None
|
||||
|
||||
@staticmethod
|
||||
def _identifiers(element: Any) -> List[str]:
|
||||
if isinstance(element, dict):
|
||||
values = []
|
||||
for key in ("name", "uri", "id", "@id"):
|
||||
value = element.get(key)
|
||||
if value is not None and str(value).strip():
|
||||
values.append(str(value).strip())
|
||||
return values
|
||||
if isinstance(element, str) and element.strip():
|
||||
return [element.strip()]
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _values(element: Dict[str, Any], key: str) -> List[Any]:
|
||||
value = element.get(key)
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
values = [item for item in value if item is not None and str(item).strip()]
|
||||
return sorted(values, key=str) if isinstance(value, set) else values
|
||||
return [value] if str(value).strip() else []
|
||||
|
||||
@classmethod
|
||||
def _term_aliases(cls, value: Any) -> Set[str]:
|
||||
if isinstance(value, dict):
|
||||
value = cls._identifier(value)
|
||||
if value is None:
|
||||
return set()
|
||||
text = str(value).strip().strip("<>")
|
||||
if not text:
|
||||
return set()
|
||||
aliases = {text, text.lower()}
|
||||
for separator in ("#", "/"):
|
||||
if separator in text:
|
||||
local = text.rstrip("/").rsplit(separator, 1)[-1]
|
||||
aliases.update({local, local.lower()})
|
||||
if ":" in text and not text.startswith(("http://", "https://")):
|
||||
local = text.rsplit(":", 1)[-1]
|
||||
aliases.update({local, local.lower()})
|
||||
return aliases
|
||||
|
||||
@classmethod
|
||||
def _match_class(cls, value: Any, aliases: Mapping[str, str]) -> Optional[str]:
|
||||
for alias in sorted(cls._term_aliases(value)):
|
||||
if alias in aliases:
|
||||
return aliases[alias]
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _is_builtin_class(cls, value: Any) -> bool:
|
||||
return any(alias in cls._BUILTIN_CLASSES for alias in cls._term_aliases(value))
|
||||
|
||||
@classmethod
|
||||
def _is_known_datatype(cls, value: Any) -> bool:
|
||||
return any(alias in cls._KNOWN_DATATYPES for alias in cls._term_aliases(value))
|
||||
|
||||
@staticmethod
|
||||
def _add_issue(
|
||||
issues: List[QualityIssue],
|
||||
code: str,
|
||||
message: str,
|
||||
severity: QualitySeverity,
|
||||
*,
|
||||
element_id: Optional[str] = None,
|
||||
element_type: Optional[str] = None,
|
||||
details: Optional[Dict[str, Any]] = None,
|
||||
) -> None:
|
||||
issues.append(
|
||||
QualityIssue(
|
||||
code=code,
|
||||
message=message,
|
||||
severity=severity,
|
||||
element_id=element_id,
|
||||
element_type=element_type,
|
||||
details=details or {},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def ontology_quality_check(
|
||||
ontology: Any,
|
||||
graph_data: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
thresholds: Optional[Mapping[str, Optional[float]]] = None,
|
||||
fail_on_warnings: bool = False,
|
||||
competency_questions: Optional[List[str]] = None,
|
||||
validator: Optional[OntologyValidator] = None,
|
||||
evaluator: Optional[OntologyEvaluator] = None,
|
||||
) -> OntologyQualityReport:
|
||||
"""Convenience wrapper around :class:`OntologyQualityGate`."""
|
||||
gate = OntologyQualityGate(
|
||||
validator=validator,
|
||||
evaluator=evaluator,
|
||||
thresholds=thresholds,
|
||||
fail_on_warnings=fail_on_warnings,
|
||||
)
|
||||
return gate.check(
|
||||
ontology,
|
||||
graph_data=graph_data,
|
||||
competency_questions=competency_questions,
|
||||
)
|
||||
@@ -1,308 +0,0 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from semantica.ontology import (
|
||||
OntologyEngine,
|
||||
OntologyQualityGate,
|
||||
QualitySeverity,
|
||||
ontology_quality_check,
|
||||
)
|
||||
|
||||
|
||||
def _ontology():
|
||||
return {
|
||||
"classes": [
|
||||
{"name": "Person", "uri": "https://example.org/Person"},
|
||||
{"name": "Company", "uri": "https://example.org/Company"},
|
||||
],
|
||||
"properties": [
|
||||
{
|
||||
"name": "worksFor",
|
||||
"type": "object",
|
||||
"domain": ["Person"],
|
||||
"range": ["Company"],
|
||||
},
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": ["Person"],
|
||||
"range": "string",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def test_quality_gate_reports_a_healthy_ontology():
|
||||
report = ontology_quality_check(_ontology())
|
||||
|
||||
assert report.passed
|
||||
assert report.metrics["coverage"] == 1.0
|
||||
assert report.error_count == 0
|
||||
assert report.to_dict()["stats"]["properties"] == 2
|
||||
|
||||
|
||||
def test_quality_gate_finds_schema_and_endpoint_problems():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person"}, {"name": "Unused"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "worksFor",
|
||||
"type": "object",
|
||||
"domain": ["Person"],
|
||||
"range": ["MissingCompany"],
|
||||
},
|
||||
{"name": "unattached", "type": "data"},
|
||||
],
|
||||
}
|
||||
graph = {
|
||||
"entities": [{"id": "p1", "type": "Person"}],
|
||||
"relationships": [
|
||||
{"source_id": "p1", "target_id": "missing", "type": "worksFor"}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology, graph_data=graph)
|
||||
codes = {issue.code for issue in report.issues}
|
||||
|
||||
assert not report.passed
|
||||
assert "UNKNOWN_RANGE" in codes
|
||||
assert "UNRESOLVED_RELATIONSHIP_ENDPOINT" in codes
|
||||
assert "ORPHAN_CLASS" in codes
|
||||
assert "MISSING_RANGE" in codes
|
||||
assert any(issue.severity == QualitySeverity.WARNING for issue in report.issues)
|
||||
|
||||
|
||||
def test_quality_gate_supports_thresholds_and_legacy_endpoint_keys():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": "Person",
|
||||
"range": "xsd:string",
|
||||
}
|
||||
],
|
||||
}
|
||||
graph = {
|
||||
"entities": [{"entity_id": "p1", "type": "Person", "name": "Alice"}],
|
||||
"relationships": [{"source": "p1", "target": "p1", "type": "knows"}],
|
||||
}
|
||||
|
||||
report = ontology_quality_check(
|
||||
ontology,
|
||||
graph_data=graph,
|
||||
thresholds={"min_coverage": 1.0},
|
||||
)
|
||||
|
||||
assert report.passed
|
||||
assert report.threshold_failures == []
|
||||
assert report.stats["relationships"] == 1
|
||||
|
||||
|
||||
def test_quality_gate_can_fail_on_warnings():
|
||||
report = ontology_quality_check(
|
||||
{"classes": [{"name": "Person"}], "properties": []},
|
||||
fail_on_warnings=True,
|
||||
)
|
||||
|
||||
assert not report.passed
|
||||
assert "fail_on_warnings" in report.threshold_failures
|
||||
|
||||
|
||||
def test_engine_exposes_quality_check():
|
||||
report = OntologyEngine().quality_check(_ontology())
|
||||
|
||||
assert report.passed
|
||||
|
||||
|
||||
def test_quality_gate_accepts_canonical_context_graph_entities_without_mutation():
|
||||
graph = {
|
||||
"entities": [
|
||||
{"id": "p1", "text": "Alice", "type": "Person"},
|
||||
{"id": "c1", "text": "Acme", "type": "Company"},
|
||||
],
|
||||
"relationships": [{"source_id": "p1", "target_id": "c1", "type": "worksFor"}],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(_ontology(), graph_data=graph)
|
||||
|
||||
assert report.passed
|
||||
assert "name" not in graph["entities"][0]
|
||||
|
||||
|
||||
def test_quality_gate_returns_structured_findings_for_malformed_graph_members():
|
||||
report = OntologyQualityGate().check(
|
||||
_ontology(), graph_data={"entities": [None], "relationships": [None]}
|
||||
)
|
||||
codes = {issue.code for issue in report.issues}
|
||||
|
||||
assert not report.passed
|
||||
assert {"INVALID_ENTITY", "INVALID_RELATIONSHIP"} <= codes
|
||||
|
||||
|
||||
def test_quality_gate_returns_structured_finding_for_invalid_graph_containers():
|
||||
report = OntologyQualityGate().check(
|
||||
_ontology(), graph_data={"entities": None, "relationships": []}
|
||||
)
|
||||
|
||||
assert not report.passed
|
||||
assert any(issue.code == "INVALID_GRAPH" for issue in report.issues)
|
||||
|
||||
|
||||
def test_quality_gate_does_not_undercount_identical_dangling_edges():
|
||||
graph = {
|
||||
"entities": [{"id": "p1", "name": "Alice", "type": "Person"}],
|
||||
"relationships": [
|
||||
{"source": "p1", "target": "missing", "type": "knows"},
|
||||
{"source": "p1", "target": "missing", "type": "knows"},
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(
|
||||
_ontology(), graph_data=graph, thresholds={"max_errors": 1}
|
||||
)
|
||||
|
||||
endpoint_errors = [
|
||||
issue
|
||||
for issue in report.issues
|
||||
if issue.code == "UNRESOLVED_RELATIONSHIP_ENDPOINT"
|
||||
]
|
||||
assert len(endpoint_errors) == 2
|
||||
assert not report.passed
|
||||
assert "max_errors" in report.threshold_failures
|
||||
|
||||
|
||||
def test_quality_gate_indexes_all_class_identifier_aliases():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person", "uri": "https://example.org/PersonType"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": "https://example.org/PersonType",
|
||||
"range": "string",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology)
|
||||
codes = {issue.code for issue in report.issues}
|
||||
|
||||
assert "UNKNOWN_DOMAIN" not in codes
|
||||
|
||||
|
||||
def test_quality_gate_recognizes_subclass_of_hierarchies():
|
||||
ontology = {
|
||||
"classes": [
|
||||
{"name": "Person"},
|
||||
{"name": "Employee", "subclassOf": "Person"},
|
||||
],
|
||||
"properties": [
|
||||
{
|
||||
"name": "name",
|
||||
"type": "data",
|
||||
"domain": "Employee",
|
||||
"range": "string",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology)
|
||||
|
||||
assert not any(issue.code == "ORPHAN_CLASS" for issue in report.issues)
|
||||
|
||||
|
||||
def test_quality_gate_keeps_competency_evaluation_safe_for_malformed_members():
|
||||
ontology = {
|
||||
"classes": [None, {"uri": "https://example.org/Person"}],
|
||||
"properties": [
|
||||
None,
|
||||
{
|
||||
"uri": "https://example.org/name",
|
||||
"type": "data",
|
||||
"domain": "Person",
|
||||
"range": "string",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(
|
||||
ontology, competency_questions=["What is a person's name?"]
|
||||
)
|
||||
|
||||
assert isinstance(report.metrics["competency_question_coverage"], float)
|
||||
|
||||
|
||||
def test_quality_gate_reports_validator_warnings_and_can_fail_on_them():
|
||||
class WarningValidator:
|
||||
def validate(self, ontology):
|
||||
return SimpleNamespace(valid=True, errors=[], warnings=["review me"])
|
||||
|
||||
report = OntologyQualityGate(
|
||||
validator=WarningValidator(), fail_on_warnings=True
|
||||
).check(_ontology())
|
||||
|
||||
assert report.warning_count == 1
|
||||
assert "VALIDATOR_WARNING" in {issue.code for issue in report.issues}
|
||||
assert "fail_on_warnings" in report.threshold_failures
|
||||
|
||||
|
||||
def test_quality_gate_isolates_competency_questions_between_checks():
|
||||
engine = OntologyEngine()
|
||||
|
||||
first = engine.quality_check(_ontology(), competency_questions=["Who is a Person?"])
|
||||
second = engine.quality_check(
|
||||
_ontology(), competency_questions=["What is a location?"]
|
||||
)
|
||||
|
||||
assert first.metrics["competency_question_coverage"] == 1.0
|
||||
assert second.metrics["competency_question_coverage"] == 0.0
|
||||
|
||||
|
||||
def test_quality_gate_treats_null_property_type_as_missing():
|
||||
ontology = {
|
||||
"classes": [{"name": "Person"}],
|
||||
"properties": [
|
||||
{
|
||||
"name": "value",
|
||||
"type": None,
|
||||
"domain": "Person",
|
||||
"range": "string",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(ontology)
|
||||
|
||||
assert "MISSING_PROPERTY_TYPE" in {issue.code for issue in report.issues}
|
||||
assert report.error_count >= 1
|
||||
|
||||
|
||||
def test_quality_gate_normalizes_orphan_ids_for_deterministic_reports():
|
||||
graph = {
|
||||
"entities": [
|
||||
{"id": "p2", "name": "Bob", "type": "Person"},
|
||||
{"id": "p1", "name": "Alice", "type": "Person"},
|
||||
],
|
||||
"relationships": [],
|
||||
}
|
||||
|
||||
report = OntologyQualityGate().check(_ontology(), graph_data=graph)
|
||||
orphan = next(issue for issue in report.issues if issue.code == "ORPHAN_ENTITY")
|
||||
|
||||
assert orphan.details["ids"] == ["p1", "p2"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"thresholds",
|
||||
[
|
||||
{"min_coverage": float("nan")},
|
||||
{"max_errors": float("inf")},
|
||||
{"max_warnings": float("-inf")},
|
||||
],
|
||||
)
|
||||
def test_quality_gate_rejects_non_finite_thresholds(thresholds):
|
||||
with pytest.raises(ValueError, match="finite"):
|
||||
OntologyQualityGate().check(_ontology(), thresholds=thresholds)
|
||||
Reference in New Issue
Block a user