585 lines
24 KiB
Python
Executable File
585 lines
24 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Enterprise Demo Fixtures Validation Script
|
|
|
|
Validates cross-references between JSON fixtures for enterprise demo sessions.
|
|
Checks that all referenced IDs exist and are consistent across files.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, List, Set, Any, Optional
|
|
from collections import defaultdict
|
|
import uuid
|
|
|
|
# Color codes for output
|
|
RED = '\033[91m'
|
|
GREEN = '\033[92m'
|
|
YELLOW = '\033[93m'
|
|
BLUE = '\033[94m'
|
|
RESET = '\033[0m'
|
|
|
|
class FixtureValidator:
|
|
def __init__(self, base_path: str = "shared/demo/fixtures/enterprise"):
|
|
self.base_path = Path(base_path)
|
|
self.parent_path = self.base_path / "parent"
|
|
self.children_paths = {}
|
|
|
|
# Load all fixture data
|
|
self.tenant_data = {}
|
|
self.user_data = {}
|
|
self.location_data = {}
|
|
self.product_data = {}
|
|
self.supplier_data = {}
|
|
self.recipe_data = {}
|
|
self.procurement_data = {}
|
|
self.order_data = {}
|
|
self.production_data = {}
|
|
|
|
# Track all IDs for validation
|
|
self.all_ids = defaultdict(set)
|
|
self.references = defaultdict(list)
|
|
|
|
# Expected IDs from tenant.json
|
|
self.expected_tenant_ids = set()
|
|
self.expected_user_ids = set()
|
|
self.expected_location_ids = set()
|
|
|
|
def load_all_fixtures(self) -> None:
|
|
"""Load all JSON fixtures from parent and children directories"""
|
|
print(f"{BLUE}Loading fixtures from {self.base_path}{RESET}")
|
|
|
|
# Load parent fixtures
|
|
self._load_parent_fixtures()
|
|
|
|
# Load children fixtures
|
|
self._load_children_fixtures()
|
|
|
|
print(f"{GREEN}✓ Loaded fixtures successfully{RESET}\n")
|
|
|
|
def _load_parent_fixtures(self) -> None:
|
|
"""Load parent enterprise fixtures"""
|
|
if not self.parent_path.exists():
|
|
print(f"{RED}✗ Parent fixtures directory not found: {self.parent_path}{RESET}")
|
|
sys.exit(1)
|
|
|
|
# Load in order to establish dependencies
|
|
files_to_load = [
|
|
"01-tenant.json",
|
|
"02-auth.json",
|
|
"03-inventory.json",
|
|
"04-recipes.json",
|
|
"05-suppliers.json",
|
|
"06-production.json",
|
|
"07-procurement.json",
|
|
"08-orders.json",
|
|
"09-sales.json",
|
|
"10-forecasting.json",
|
|
"11-orchestrator.json"
|
|
]
|
|
|
|
for filename in files_to_load:
|
|
filepath = self.parent_path / filename
|
|
if filepath.exists():
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
self._process_fixture_file(filename, data, "parent")
|
|
|
|
def _load_children_fixtures(self) -> None:
|
|
"""Load children enterprise fixtures"""
|
|
children_dir = self.base_path / "children"
|
|
if not children_dir.exists():
|
|
print(f"{YELLOW}⚠ Children fixtures directory not found: {children_dir}{RESET}")
|
|
return
|
|
|
|
# Find all child tenant directories
|
|
child_dirs = [d for d in children_dir.iterdir() if d.is_dir()]
|
|
|
|
for child_dir in child_dirs:
|
|
tenant_id = child_dir.name
|
|
self.children_paths[tenant_id] = child_dir
|
|
|
|
# Load child fixtures
|
|
files_to_load = [
|
|
"01-tenant.json",
|
|
"02-auth.json",
|
|
"03-inventory.json",
|
|
"04-recipes.json",
|
|
"05-suppliers.json",
|
|
"06-production.json",
|
|
"07-procurement.json",
|
|
"08-orders.json",
|
|
"09-sales.json",
|
|
"10-forecasting.json",
|
|
"11-orchestrator.json"
|
|
]
|
|
|
|
for filename in files_to_load:
|
|
filepath = child_dir / filename
|
|
if filepath.exists():
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
self._process_fixture_file(filename, data, tenant_id)
|
|
|
|
def _process_fixture_file(self, filename: str, data: Any, context: str) -> None:
|
|
"""Process a fixture file and extract IDs and references"""
|
|
print(f" Processing {filename} ({context})...")
|
|
|
|
if filename == "01-tenant.json":
|
|
self._process_tenant_data(data, context)
|
|
elif filename == "02-auth.json":
|
|
self._process_auth_data(data, context)
|
|
elif filename == "03-inventory.json":
|
|
self._process_inventory_data(data, context)
|
|
elif filename == "04-recipes.json":
|
|
self._process_recipe_data(data, context)
|
|
elif filename == "05-suppliers.json":
|
|
self._process_supplier_data(data, context)
|
|
elif filename == "06-production.json":
|
|
self._process_production_data(data, context)
|
|
elif filename == "07-procurement.json":
|
|
self._process_procurement_data(data, context)
|
|
elif filename == "08-orders.json":
|
|
self._process_order_data(data, context)
|
|
elif filename == "09-sales.json":
|
|
self._process_sales_data(data, context)
|
|
elif filename == "10-forecasting.json":
|
|
self._process_forecasting_data(data, context)
|
|
elif filename == "11-orchestrator.json":
|
|
self._process_orchestrator_data(data, context)
|
|
|
|
def _process_tenant_data(self, data: Any, context: str) -> None:
|
|
"""Process tenant.json data"""
|
|
tenant = data.get("tenant", {})
|
|
owner = data.get("owner", {})
|
|
subscription = data.get("subscription", {})
|
|
children = data.get("children", [])
|
|
|
|
# Store tenant data
|
|
tenant_id = tenant.get("id")
|
|
if tenant_id:
|
|
self.tenant_data[tenant_id] = tenant
|
|
self.all_ids["tenant"].add(tenant_id)
|
|
|
|
if context == "parent":
|
|
self.expected_tenant_ids.add(tenant_id)
|
|
|
|
# Store owner user
|
|
owner_id = owner.get("id")
|
|
if owner_id:
|
|
self.user_data[owner_id] = owner
|
|
self.all_ids["user"].add(owner_id)
|
|
self.expected_user_ids.add(owner_id)
|
|
|
|
# Store subscription
|
|
subscription_id = subscription.get("id")
|
|
if subscription_id:
|
|
self.all_ids["subscription"].add(subscription_id)
|
|
|
|
# Store child tenants
|
|
for child in children:
|
|
child_id = child.get("id")
|
|
if child_id:
|
|
self.tenant_data[child_id] = child
|
|
self.all_ids["tenant"].add(child_id)
|
|
self.expected_tenant_ids.add(child_id)
|
|
|
|
# Track parent-child relationship
|
|
self.references["parent_child"].append({
|
|
"parent": tenant_id,
|
|
"child": child_id,
|
|
"context": context
|
|
})
|
|
|
|
def _process_auth_data(self, data: Any, context: str) -> None:
|
|
"""Process auth.json data"""
|
|
users = data.get("users", [])
|
|
|
|
for user in users:
|
|
user_id = user.get("id")
|
|
tenant_id = user.get("tenant_id")
|
|
|
|
if user_id:
|
|
self.user_data[user_id] = user
|
|
self.all_ids["user"].add(user_id)
|
|
self.expected_user_ids.add(user_id)
|
|
|
|
# Track user-tenant relationship
|
|
if tenant_id:
|
|
self.references["user_tenant"].append({
|
|
"user_id": user_id,
|
|
"tenant_id": tenant_id,
|
|
"context": context
|
|
})
|
|
|
|
def _process_inventory_data(self, data: Any, context: str) -> None:
|
|
"""Process inventory.json data"""
|
|
products = data.get("products", [])
|
|
ingredients = data.get("ingredients", [])
|
|
locations = data.get("locations", [])
|
|
|
|
# Store products
|
|
for product in products:
|
|
product_id = product.get("id")
|
|
tenant_id = product.get("tenant_id")
|
|
created_by = product.get("created_by")
|
|
|
|
if product_id:
|
|
self.product_data[product_id] = product
|
|
self.all_ids["product"].add(product_id)
|
|
|
|
# Track product-tenant relationship
|
|
if tenant_id:
|
|
self.references["product_tenant"].append({
|
|
"product_id": product_id,
|
|
"tenant_id": tenant_id,
|
|
"context": context
|
|
})
|
|
|
|
# Track product-user relationship
|
|
if created_by:
|
|
self.references["product_user"].append({
|
|
"product_id": product_id,
|
|
"user_id": created_by,
|
|
"context": context
|
|
})
|
|
|
|
# Store ingredients
|
|
for ingredient in ingredients:
|
|
ingredient_id = ingredient.get("id")
|
|
tenant_id = ingredient.get("tenant_id")
|
|
created_by = ingredient.get("created_by")
|
|
|
|
if ingredient_id:
|
|
self.product_data[ingredient_id] = ingredient
|
|
self.all_ids["ingredient"].add(ingredient_id)
|
|
|
|
# Track ingredient-tenant relationship
|
|
if tenant_id:
|
|
self.references["ingredient_tenant"].append({
|
|
"ingredient_id": ingredient_id,
|
|
"tenant_id": tenant_id,
|
|
"context": context
|
|
})
|
|
|
|
# Track ingredient-user relationship
|
|
if created_by:
|
|
self.references["ingredient_user"].append({
|
|
"ingredient_id": ingredient_id,
|
|
"user_id": created_by,
|
|
"context": context
|
|
})
|
|
|
|
# Store locations
|
|
for location in locations:
|
|
location_id = location.get("id")
|
|
if location_id:
|
|
self.location_data[location_id] = location
|
|
self.all_ids["location"].add(location_id)
|
|
self.expected_location_ids.add(location_id)
|
|
|
|
def _process_recipe_data(self, data: Any, context: str) -> None:
|
|
"""Process recipes.json data"""
|
|
recipes = data.get("recipes", [])
|
|
|
|
for recipe in recipes:
|
|
recipe_id = recipe.get("id")
|
|
tenant_id = recipe.get("tenant_id")
|
|
finished_product_id = recipe.get("finished_product_id")
|
|
|
|
if recipe_id:
|
|
self.recipe_data[recipe_id] = recipe
|
|
self.all_ids["recipe"].add(recipe_id)
|
|
|
|
# Track recipe-tenant relationship
|
|
if tenant_id:
|
|
self.references["recipe_tenant"].append({
|
|
"recipe_id": recipe_id,
|
|
"tenant_id": tenant_id,
|
|
"context": context
|
|
})
|
|
|
|
# Track recipe-product relationship
|
|
if finished_product_id:
|
|
self.references["recipe_product"].append({
|
|
"recipe_id": recipe_id,
|
|
"product_id": finished_product_id,
|
|
"context": context
|
|
})
|
|
|
|
def _process_supplier_data(self, data: Any, context: str) -> None:
|
|
"""Process suppliers.json data"""
|
|
suppliers = data.get("suppliers", [])
|
|
|
|
for supplier in suppliers:
|
|
supplier_id = supplier.get("id")
|
|
tenant_id = supplier.get("tenant_id")
|
|
|
|
if supplier_id:
|
|
self.supplier_data[supplier_id] = supplier
|
|
self.all_ids["supplier"].add(supplier_id)
|
|
|
|
# Track supplier-tenant relationship
|
|
if tenant_id:
|
|
self.references["supplier_tenant"].append({
|
|
"supplier_id": supplier_id,
|
|
"tenant_id": tenant_id,
|
|
"context": context
|
|
})
|
|
|
|
def _process_production_data(self, data: Any, context: str) -> None:
|
|
"""Process production.json data"""
|
|
# Extract production-related IDs
|
|
pass
|
|
|
|
def _process_procurement_data(self, data: Any, context: str) -> None:
|
|
"""Process procurement.json data"""
|
|
# Extract procurement-related IDs
|
|
pass
|
|
|
|
def _process_order_data(self, data: Any, context: str) -> None:
|
|
"""Process orders.json data"""
|
|
# Extract order-related IDs
|
|
pass
|
|
|
|
def _process_sales_data(self, data: Any, context: str) -> None:
|
|
"""Process sales.json data"""
|
|
# Extract sales-related IDs
|
|
pass
|
|
|
|
def _process_forecasting_data(self, data: Any, context: str) -> None:
|
|
"""Process forecasting.json data"""
|
|
# Extract forecasting-related IDs
|
|
pass
|
|
|
|
def _process_orchestrator_data(self, data: Any, context: str) -> None:
|
|
"""Process orchestrator.json data"""
|
|
# Extract orchestrator-related IDs
|
|
pass
|
|
|
|
def validate_all_references(self) -> bool:
|
|
"""Validate all cross-references in the fixtures"""
|
|
print(f"{BLUE}Validating cross-references...{RESET}")
|
|
|
|
all_valid = True
|
|
|
|
# Validate user-tenant relationships
|
|
if "user_tenant" in self.references:
|
|
print(f"\n{YELLOW}Validating User-Tenant relationships...{RESET}")
|
|
for ref in self.references["user_tenant"]:
|
|
user_id = ref["user_id"]
|
|
tenant_id = ref["tenant_id"]
|
|
context = ref["context"]
|
|
|
|
if user_id not in self.user_data:
|
|
print(f"{RED}✗ User {user_id} referenced but not found in user data (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if tenant_id not in self.tenant_data:
|
|
print(f"{RED}✗ Tenant {tenant_id} referenced by user {user_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate parent-child relationships
|
|
if "parent_child" in self.references:
|
|
print(f"\n{YELLOW}Validating Parent-Child relationships...{RESET}")
|
|
for ref in self.references["parent_child"]:
|
|
parent_id = ref["parent"]
|
|
child_id = ref["child"]
|
|
context = ref["context"]
|
|
|
|
if parent_id not in self.tenant_data:
|
|
print(f"{RED}✗ Parent tenant {parent_id} not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if child_id not in self.tenant_data:
|
|
print(f"{RED}✗ Child tenant {child_id} not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate product-tenant relationships
|
|
if "product_tenant" in self.references:
|
|
print(f"\n{YELLOW}Validating Product-Tenant relationships...{RESET}")
|
|
for ref in self.references["product_tenant"]:
|
|
product_id = ref["product_id"]
|
|
tenant_id = ref["tenant_id"]
|
|
context = ref["context"]
|
|
|
|
if product_id not in self.product_data:
|
|
print(f"{RED}✗ Product {product_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if tenant_id not in self.tenant_data:
|
|
print(f"{RED}✗ Tenant {tenant_id} referenced by product {product_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate product-user relationships
|
|
if "product_user" in self.references:
|
|
print(f"\n{YELLOW}Validating Product-User relationships...{RESET}")
|
|
for ref in self.references["product_user"]:
|
|
product_id = ref["product_id"]
|
|
user_id = ref["user_id"]
|
|
context = ref["context"]
|
|
|
|
if product_id not in self.product_data:
|
|
print(f"{RED}✗ Product {product_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if user_id not in self.user_data:
|
|
print(f"{RED}✗ User {user_id} referenced by product {product_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate ingredient-tenant relationships
|
|
if "ingredient_tenant" in self.references:
|
|
print(f"\n{YELLOW}Validating Ingredient-Tenant relationships...{RESET}")
|
|
for ref in self.references["ingredient_tenant"]:
|
|
ingredient_id = ref["ingredient_id"]
|
|
tenant_id = ref["tenant_id"]
|
|
context = ref["context"]
|
|
|
|
if ingredient_id not in self.product_data:
|
|
print(f"{RED}✗ Ingredient {ingredient_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if tenant_id not in self.tenant_data:
|
|
print(f"{RED}✗ Tenant {tenant_id} referenced by ingredient {ingredient_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate ingredient-user relationships
|
|
if "ingredient_user" in self.references:
|
|
print(f"\n{YELLOW}Validating Ingredient-User relationships...{RESET}")
|
|
for ref in self.references["ingredient_user"]:
|
|
ingredient_id = ref["ingredient_id"]
|
|
user_id = ref["user_id"]
|
|
context = ref["context"]
|
|
|
|
if ingredient_id not in self.product_data:
|
|
print(f"{RED}✗ Ingredient {ingredient_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if user_id not in self.user_data:
|
|
print(f"{RED}✗ User {user_id} referenced by ingredient {ingredient_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate recipe-tenant relationships
|
|
if "recipe_tenant" in self.references:
|
|
print(f"\n{YELLOW}Validating Recipe-Tenant relationships...{RESET}")
|
|
for ref in self.references["recipe_tenant"]:
|
|
recipe_id = ref["recipe_id"]
|
|
tenant_id = ref["tenant_id"]
|
|
context = ref["context"]
|
|
|
|
if recipe_id not in self.recipe_data:
|
|
print(f"{RED}✗ Recipe {recipe_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if tenant_id not in self.tenant_data:
|
|
print(f"{RED}✗ Tenant {tenant_id} referenced by recipe {recipe_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate recipe-product relationships
|
|
if "recipe_product" in self.references:
|
|
print(f"\n{YELLOW}Validating Recipe-Product relationships...{RESET}")
|
|
for ref in self.references["recipe_product"]:
|
|
recipe_id = ref["recipe_id"]
|
|
product_id = ref["product_id"]
|
|
context = ref["context"]
|
|
|
|
if recipe_id not in self.recipe_data:
|
|
print(f"{RED}✗ Recipe {recipe_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if product_id not in self.product_data:
|
|
print(f"{RED}✗ Product {product_id} referenced by recipe {recipe_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate supplier-tenant relationships
|
|
if "supplier_tenant" in self.references:
|
|
print(f"\n{YELLOW}Validating Supplier-Tenant relationships...{RESET}")
|
|
for ref in self.references["supplier_tenant"]:
|
|
supplier_id = ref["supplier_id"]
|
|
tenant_id = ref["tenant_id"]
|
|
context = ref["context"]
|
|
|
|
if supplier_id not in self.supplier_data:
|
|
print(f"{RED}✗ Supplier {supplier_id} referenced but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
if tenant_id not in self.tenant_data:
|
|
print(f"{RED}✗ Tenant {tenant_id} referenced by supplier {supplier_id} but not found (context: {context}){RESET}")
|
|
all_valid = False
|
|
|
|
# Validate UUID format for all IDs
|
|
print(f"\n{YELLOW}Validating UUID formats...{RESET}")
|
|
for entity_type, ids in self.all_ids.items():
|
|
for entity_id in ids:
|
|
try:
|
|
uuid.UUID(entity_id)
|
|
except ValueError:
|
|
print(f"{RED}✗ Invalid UUID format for {entity_type} ID: {entity_id}{RESET}")
|
|
all_valid = False
|
|
|
|
# Check for duplicate IDs
|
|
print(f"\n{YELLOW}Checking for duplicate IDs...{RESET}")
|
|
all_entities = []
|
|
for ids in self.all_ids.values():
|
|
all_entities.extend(ids)
|
|
|
|
duplicates = [id for id in all_entities if all_entities.count(id) > 1]
|
|
if duplicates:
|
|
print(f"{RED}✗ Found duplicate IDs: {', '.join(duplicates)}{RESET}")
|
|
all_valid = False
|
|
|
|
if all_valid:
|
|
print(f"{GREEN}✓ All cross-references are valid!{RESET}")
|
|
else:
|
|
print(f"{RED}✗ Found validation errors!{RESET}")
|
|
|
|
return all_valid
|
|
|
|
def generate_summary(self) -> None:
|
|
"""Generate a summary of the loaded fixtures"""
|
|
print(f"\n{BLUE}=== Fixture Summary ==={RESET}")
|
|
print(f"Tenants: {len(self.tenant_data)}")
|
|
print(f"Users: {len(self.user_data)}")
|
|
print(f"Products: {len(self.product_data)}")
|
|
print(f"Suppliers: {len(self.supplier_data)}")
|
|
print(f"Recipes: {len(self.recipe_data)}")
|
|
print(f"Locations: {len(self.location_data)}")
|
|
|
|
print(f"\nEntity Types: {list(self.all_ids.keys())}")
|
|
|
|
for entity_type, ids in self.all_ids.items():
|
|
print(f" {entity_type}: {len(ids)} IDs")
|
|
|
|
print(f"\nReference Types: {list(self.references.keys())}")
|
|
for ref_type, refs in self.references.items():
|
|
print(f" {ref_type}: {len(refs)} references")
|
|
|
|
def run_validation(self) -> bool:
|
|
"""Run the complete validation process"""
|
|
print(f"{BLUE}=== Enterprise Demo Fixtures Validator ==={RESET}")
|
|
print(f"Base Path: {self.base_path}\n")
|
|
|
|
try:
|
|
self.load_all_fixtures()
|
|
self.generate_summary()
|
|
return self.validate_all_references()
|
|
|
|
except Exception as e:
|
|
print(f"{RED}✗ Validation failed with error: {e}{RESET}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
validator = FixtureValidator()
|
|
success = validator.run_validation()
|
|
|
|
if success:
|
|
print(f"\n{GREEN}=== Validation Complete: All checks passed! ==={RESET}")
|
|
sys.exit(0)
|
|
else:
|
|
print(f"\n{RED}=== Validation Complete: Errors found! ==={RESET}")
|
|
sys.exit(1)
|