#!/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)