Anto01
6c30224341
Config Layer: - spec_models.py: Pydantic models for AtomizerSpec v2.0 - spec_validator.py: Semantic validation with detailed error reporting Extractors: - custom_extractor_loader.py: Runtime custom extractor loading - spec_extractor_builder.py: Build extractors from spec definitions Tools: - migrate_to_spec_v2.py: CLI tool for batch migration Tests: - test_migrator.py: Migration tests - test_spec_manager.py: SpecManager service tests - test_spec_api.py: REST API tests - test_mcp_tools.py: MCP tool tests - test_e2e_unified_config.py: End-to-end config tests
480 lines
17 KiB
Python
480 lines
17 KiB
Python
"""
|
|
End-to-End Tests for AtomizerSpec v2.0 Unified Configuration
|
|
|
|
Tests the complete workflow from spec creation through optimization setup.
|
|
|
|
P4.10: End-to-end testing
|
|
"""
|
|
|
|
import json
|
|
import pytest
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "atomizer-dashboard" / "backend"))
|
|
|
|
|
|
# ============================================================================
|
|
# End-to-End Test Scenarios
|
|
# ============================================================================
|
|
|
|
class TestE2ESpecWorkflow:
|
|
"""End-to-end tests for complete spec workflow."""
|
|
|
|
@pytest.fixture
|
|
def e2e_study_dir(self):
|
|
"""Create a temporary study directory for E2E testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
study_dir = Path(tmpdir) / "e2e_test_study"
|
|
study_dir.mkdir()
|
|
|
|
# Create standard Atomizer study structure
|
|
(study_dir / "1_setup").mkdir()
|
|
(study_dir / "2_iterations").mkdir()
|
|
(study_dir / "3_results").mkdir()
|
|
|
|
yield study_dir
|
|
|
|
def test_create_spec_from_scratch(self, e2e_study_dir):
|
|
"""Test creating a new AtomizerSpec from scratch."""
|
|
from optimization_engine.config.spec_models import AtomizerSpec
|
|
|
|
# Create a minimal spec
|
|
spec_data = {
|
|
"meta": {
|
|
"version": "2.0",
|
|
"created": datetime.now().isoformat() + "Z",
|
|
"modified": datetime.now().isoformat() + "Z",
|
|
"created_by": "api",
|
|
"modified_by": "api",
|
|
"study_name": "e2e_test_study",
|
|
"description": "End-to-end test study"
|
|
},
|
|
"model": {
|
|
"sim": {"path": "model.sim", "solver": "nastran"}
|
|
},
|
|
"design_variables": [
|
|
{
|
|
"id": "dv_001",
|
|
"name": "thickness",
|
|
"expression_name": "thickness",
|
|
"type": "continuous",
|
|
"bounds": {"min": 1.0, "max": 10.0},
|
|
"baseline": 5.0,
|
|
"enabled": True,
|
|
"canvas_position": {"x": 50, "y": 100}
|
|
}
|
|
],
|
|
"extractors": [
|
|
{
|
|
"id": "ext_001",
|
|
"name": "Mass Extractor",
|
|
"type": "mass",
|
|
"builtin": True,
|
|
"outputs": [{"name": "mass", "units": "kg"}],
|
|
"canvas_position": {"x": 740, "y": 100}
|
|
}
|
|
],
|
|
"objectives": [
|
|
{
|
|
"id": "obj_001",
|
|
"name": "mass",
|
|
"direction": "minimize",
|
|
"source": {"extractor_id": "ext_001", "output_name": "mass"},
|
|
"canvas_position": {"x": 1020, "y": 100}
|
|
}
|
|
],
|
|
"constraints": [],
|
|
"optimization": {
|
|
"algorithm": {"type": "TPE"},
|
|
"budget": {"max_trials": 50}
|
|
},
|
|
"canvas": {
|
|
"edges": [
|
|
{"source": "dv_001", "target": "model"},
|
|
{"source": "model", "target": "solver"},
|
|
{"source": "solver", "target": "ext_001"},
|
|
{"source": "ext_001", "target": "obj_001"},
|
|
{"source": "obj_001", "target": "optimization"}
|
|
],
|
|
"layout_version": "2.0"
|
|
}
|
|
}
|
|
|
|
# Validate with Pydantic
|
|
spec = AtomizerSpec.model_validate(spec_data)
|
|
assert spec.meta.study_name == "e2e_test_study"
|
|
assert spec.meta.version == "2.0"
|
|
assert len(spec.design_variables) == 1
|
|
assert len(spec.extractors) == 1
|
|
assert len(spec.objectives) == 1
|
|
|
|
# Save to file
|
|
spec_path = e2e_study_dir / "atomizer_spec.json"
|
|
with open(spec_path, "w") as f:
|
|
json.dump(spec_data, f, indent=2)
|
|
|
|
assert spec_path.exists()
|
|
|
|
def test_load_and_modify_spec(self, e2e_study_dir):
|
|
"""Test loading an existing spec and modifying it."""
|
|
from optimization_engine.config.spec_models import AtomizerSpec
|
|
from optimization_engine.config.spec_validator import SpecValidator
|
|
|
|
# First create the spec
|
|
spec_data = {
|
|
"meta": {
|
|
"version": "2.0",
|
|
"created": datetime.now().isoformat() + "Z",
|
|
"modified": datetime.now().isoformat() + "Z",
|
|
"created_by": "api",
|
|
"modified_by": "api",
|
|
"study_name": "e2e_test_study"
|
|
},
|
|
"model": {
|
|
"sim": {"path": "model.sim", "solver": "nastran"}
|
|
},
|
|
"design_variables": [
|
|
{
|
|
"id": "dv_001",
|
|
"name": "thickness",
|
|
"expression_name": "thickness",
|
|
"type": "continuous",
|
|
"bounds": {"min": 1.0, "max": 10.0},
|
|
"baseline": 5.0,
|
|
"enabled": True,
|
|
"canvas_position": {"x": 50, "y": 100}
|
|
}
|
|
],
|
|
"extractors": [
|
|
{
|
|
"id": "ext_001",
|
|
"name": "Mass Extractor",
|
|
"type": "mass",
|
|
"builtin": True,
|
|
"outputs": [{"name": "mass", "units": "kg"}],
|
|
"canvas_position": {"x": 740, "y": 100}
|
|
}
|
|
],
|
|
"objectives": [
|
|
{
|
|
"id": "obj_001",
|
|
"name": "mass",
|
|
"direction": "minimize",
|
|
"source": {"extractor_id": "ext_001", "output_name": "mass"},
|
|
"canvas_position": {"x": 1020, "y": 100}
|
|
}
|
|
],
|
|
"constraints": [],
|
|
"optimization": {
|
|
"algorithm": {"type": "TPE"},
|
|
"budget": {"max_trials": 50}
|
|
},
|
|
"canvas": {
|
|
"edges": [
|
|
{"source": "dv_001", "target": "model"},
|
|
{"source": "model", "target": "solver"},
|
|
{"source": "solver", "target": "ext_001"},
|
|
{"source": "ext_001", "target": "obj_001"},
|
|
{"source": "obj_001", "target": "optimization"}
|
|
],
|
|
"layout_version": "2.0"
|
|
}
|
|
}
|
|
|
|
spec_path = e2e_study_dir / "atomizer_spec.json"
|
|
with open(spec_path, "w") as f:
|
|
json.dump(spec_data, f, indent=2)
|
|
|
|
# Load and modify
|
|
with open(spec_path) as f:
|
|
loaded_data = json.load(f)
|
|
|
|
# Modify bounds
|
|
loaded_data["design_variables"][0]["bounds"]["max"] = 15.0
|
|
loaded_data["meta"]["modified"] = datetime.now().isoformat() + "Z"
|
|
loaded_data["meta"]["modified_by"] = "api"
|
|
|
|
# Validate modified spec
|
|
validator = SpecValidator()
|
|
report = validator.validate(loaded_data, strict=False)
|
|
assert report.valid is True
|
|
|
|
# Save modified spec
|
|
with open(spec_path, "w") as f:
|
|
json.dump(loaded_data, f, indent=2)
|
|
|
|
# Reload and verify
|
|
spec = AtomizerSpec.model_validate(loaded_data)
|
|
assert spec.design_variables[0].bounds.max == 15.0
|
|
|
|
def test_spec_manager_workflow(self, e2e_study_dir):
|
|
"""Test the SpecManager service workflow."""
|
|
try:
|
|
from api.services.spec_manager import SpecManager, SpecManagerError
|
|
except ImportError:
|
|
pytest.skip("SpecManager not available")
|
|
|
|
# Create initial spec
|
|
spec_data = {
|
|
"meta": {
|
|
"version": "2.0",
|
|
"created": datetime.now().isoformat() + "Z",
|
|
"modified": datetime.now().isoformat() + "Z",
|
|
"created_by": "api",
|
|
"modified_by": "api",
|
|
"study_name": "e2e_test_study"
|
|
},
|
|
"model": {
|
|
"sim": {"path": "model.sim", "solver": "nastran"}
|
|
},
|
|
"design_variables": [
|
|
{
|
|
"id": "dv_001",
|
|
"name": "thickness",
|
|
"expression_name": "thickness",
|
|
"type": "continuous",
|
|
"bounds": {"min": 1.0, "max": 10.0},
|
|
"baseline": 5.0,
|
|
"enabled": True,
|
|
"canvas_position": {"x": 50, "y": 100}
|
|
}
|
|
],
|
|
"extractors": [
|
|
{
|
|
"id": "ext_001",
|
|
"name": "Mass Extractor",
|
|
"type": "mass",
|
|
"builtin": True,
|
|
"outputs": [{"name": "mass", "units": "kg"}],
|
|
"canvas_position": {"x": 740, "y": 100}
|
|
}
|
|
],
|
|
"objectives": [
|
|
{
|
|
"id": "obj_001",
|
|
"name": "mass",
|
|
"direction": "minimize",
|
|
"source": {"extractor_id": "ext_001", "output_name": "mass"},
|
|
"canvas_position": {"x": 1020, "y": 100}
|
|
}
|
|
],
|
|
"constraints": [],
|
|
"optimization": {
|
|
"algorithm": {"type": "TPE"},
|
|
"budget": {"max_trials": 50}
|
|
},
|
|
"canvas": {
|
|
"edges": [
|
|
{"source": "dv_001", "target": "model"},
|
|
{"source": "model", "target": "solver"},
|
|
{"source": "solver", "target": "ext_001"},
|
|
{"source": "ext_001", "target": "obj_001"},
|
|
{"source": "obj_001", "target": "optimization"}
|
|
],
|
|
"layout_version": "2.0"
|
|
}
|
|
}
|
|
|
|
spec_path = e2e_study_dir / "atomizer_spec.json"
|
|
with open(spec_path, "w") as f:
|
|
json.dump(spec_data, f, indent=2)
|
|
|
|
# Use SpecManager
|
|
manager = SpecManager(e2e_study_dir)
|
|
|
|
# Test exists
|
|
assert manager.exists() is True
|
|
|
|
# Test load
|
|
spec = manager.load()
|
|
assert spec.meta.study_name == "e2e_test_study"
|
|
|
|
# Test get hash
|
|
hash1 = manager.get_hash()
|
|
assert isinstance(hash1, str)
|
|
assert len(hash1) > 0
|
|
|
|
# Test validation
|
|
report = manager.validate_and_report()
|
|
assert report.valid is True
|
|
|
|
|
|
class TestE2EMigrationWorkflow:
|
|
"""End-to-end tests for legacy config migration."""
|
|
|
|
@pytest.fixture
|
|
def legacy_study_dir(self):
|
|
"""Create a study with legacy optimization_config.json."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
study_dir = Path(tmpdir) / "legacy_study"
|
|
study_dir.mkdir()
|
|
|
|
legacy_config = {
|
|
"study_name": "legacy_study",
|
|
"description": "Test legacy config migration",
|
|
"nx_settings": {
|
|
"sim_file": "model.sim",
|
|
"nx_install_path": "C:\\Program Files\\Siemens\\NX2506"
|
|
},
|
|
"design_variables": [
|
|
{
|
|
"name": "width",
|
|
"parameter": "width",
|
|
"bounds": [5.0, 20.0],
|
|
"baseline": 10.0,
|
|
"units": "mm"
|
|
}
|
|
],
|
|
"objectives": [
|
|
{"name": "mass", "goal": "minimize", "weight": 1.0}
|
|
],
|
|
"optimization": {
|
|
"algorithm": "TPE",
|
|
"n_trials": 100
|
|
}
|
|
}
|
|
|
|
config_path = study_dir / "optimization_config.json"
|
|
with open(config_path, "w") as f:
|
|
json.dump(legacy_config, f, indent=2)
|
|
|
|
yield study_dir
|
|
|
|
def test_migrate_legacy_config(self, legacy_study_dir):
|
|
"""Test migrating a legacy config to AtomizerSpec v2.0."""
|
|
from optimization_engine.config.migrator import SpecMigrator
|
|
|
|
# Run migration
|
|
migrator = SpecMigrator(legacy_study_dir)
|
|
legacy_path = legacy_study_dir / "optimization_config.json"
|
|
|
|
with open(legacy_path) as f:
|
|
legacy = json.load(f)
|
|
|
|
spec = migrator.migrate(legacy)
|
|
|
|
# Verify migration results
|
|
assert spec["meta"]["version"] == "2.0"
|
|
assert spec["meta"]["study_name"] == "legacy_study"
|
|
assert len(spec["design_variables"]) == 1
|
|
assert spec["design_variables"][0]["bounds"]["min"] == 5.0
|
|
assert spec["design_variables"][0]["bounds"]["max"] == 20.0
|
|
|
|
def test_migration_preserves_semantics(self, legacy_study_dir):
|
|
"""Test that migration preserves the semantic meaning of the config."""
|
|
from optimization_engine.config.migrator import SpecMigrator
|
|
from optimization_engine.config.spec_models import AtomizerSpec
|
|
|
|
migrator = SpecMigrator(legacy_study_dir)
|
|
legacy_path = legacy_study_dir / "optimization_config.json"
|
|
|
|
with open(legacy_path) as f:
|
|
legacy = json.load(f)
|
|
|
|
spec_dict = migrator.migrate(legacy)
|
|
|
|
# Validate with Pydantic
|
|
spec = AtomizerSpec.model_validate(spec_dict)
|
|
|
|
# Check semantic preservation
|
|
# - Study name should be preserved
|
|
assert spec.meta.study_name == legacy["study_name"]
|
|
|
|
# - Design variable bounds should be preserved
|
|
legacy_dv = legacy["design_variables"][0]
|
|
new_dv = spec.design_variables[0]
|
|
assert new_dv.bounds.min == legacy_dv["bounds"][0]
|
|
assert new_dv.bounds.max == legacy_dv["bounds"][1]
|
|
|
|
# - Optimization settings should be preserved
|
|
assert spec.optimization.algorithm.type.value == legacy["optimization"]["algorithm"]
|
|
assert spec.optimization.budget.max_trials == legacy["optimization"]["n_trials"]
|
|
|
|
|
|
class TestE2EExtractorIntegration:
|
|
"""End-to-end tests for extractor integration with specs."""
|
|
|
|
def test_build_extractors_from_spec(self):
|
|
"""Test building extractors from a spec."""
|
|
from optimization_engine.extractors import build_extractors_from_spec
|
|
|
|
spec_data = {
|
|
"meta": {
|
|
"version": "2.0",
|
|
"created": datetime.now().isoformat() + "Z",
|
|
"modified": datetime.now().isoformat() + "Z",
|
|
"created_by": "api",
|
|
"modified_by": "api",
|
|
"study_name": "extractor_test"
|
|
},
|
|
"model": {
|
|
"sim": {"path": "model.sim", "solver": "nastran"}
|
|
},
|
|
"design_variables": [
|
|
{
|
|
"id": "dv_001",
|
|
"name": "thickness",
|
|
"expression_name": "thickness",
|
|
"type": "continuous",
|
|
"bounds": {"min": 1.0, "max": 10.0},
|
|
"baseline": 5.0,
|
|
"enabled": True,
|
|
"canvas_position": {"x": 50, "y": 100}
|
|
}
|
|
],
|
|
"extractors": [
|
|
{
|
|
"id": "ext_001",
|
|
"name": "Mass Extractor",
|
|
"type": "mass",
|
|
"builtin": True,
|
|
"outputs": [{"name": "mass", "units": "kg"}],
|
|
"canvas_position": {"x": 740, "y": 100}
|
|
}
|
|
],
|
|
"objectives": [
|
|
{
|
|
"id": "obj_001",
|
|
"name": "mass",
|
|
"direction": "minimize",
|
|
"source": {"extractor_id": "ext_001", "output_name": "mass"},
|
|
"canvas_position": {"x": 1020, "y": 100}
|
|
}
|
|
],
|
|
"constraints": [],
|
|
"optimization": {
|
|
"algorithm": {"type": "TPE"},
|
|
"budget": {"max_trials": 50}
|
|
},
|
|
"canvas": {
|
|
"edges": [
|
|
{"source": "dv_001", "target": "model"},
|
|
{"source": "model", "target": "solver"},
|
|
{"source": "solver", "target": "ext_001"},
|
|
{"source": "ext_001", "target": "obj_001"},
|
|
{"source": "obj_001", "target": "optimization"}
|
|
],
|
|
"layout_version": "2.0"
|
|
}
|
|
}
|
|
|
|
# Build extractors
|
|
extractors = build_extractors_from_spec(spec_data)
|
|
|
|
# Verify extractors were built
|
|
assert isinstance(extractors, dict)
|
|
assert "ext_001" in extractors
|
|
|
|
|
|
# ============================================================================
|
|
# Run Tests
|
|
# ============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|