Initial upload: mockapi - OpenAPI Mock Server Generator

src/mockapi/core/generators.py

@@ -0,0 +1,204 @@
+"""Random Data Generator for JSON schemas."""
+
+import random
+import uuid
+from typing import Any, Dict, List, Optional
+
+from faker import Faker
+
+
+class DataGenerator:
+    """Generates realistic random test data from JSON schemas."""
+
+    def __init__(self, seed: Optional[int] = None, schemas: Optional[Dict[str, Any]] = None):
+        """Initialize the data generator.
+
+        Args:
+            seed: Random seed for reproducible data generation
+            schemas: Dictionary of schemas for $ref resolution
+        """
+        self.seed = seed
+        self.faker = Faker()
+        if seed is not None:
+            Faker.seed(seed)
+            random.seed(seed)
+        self._schemas_dict = schemas or {}
+
+        self._faker_providers = {
+            "email": lambda: self.faker.email(),
+            "uuid": lambda: str(uuid.uuid4()),
+            "uri": lambda: self.faker.uri(),
+            "url": lambda: self.faker.url(),
+            "date": lambda: self.faker.date(),
+            "date-time": lambda: self.faker.iso8601(),
+            "datetime": lambda: self.faker.iso8601(),
+            "time": lambda: self.faker.time(),
+            "phone": lambda: self.faker.phone_number(),
+            "name": lambda: self.faker.name(),
+            "first_name": lambda: self.faker.first_name(),
+            "last_name": lambda: self.faker.last_name(),
+            "company": lambda: self.faker.company(),
+            "address": lambda: self.faker.address(),
+            "city": lambda: self.faker.city(),
+            "country": lambda: self.faker.country(),
+            "sentence": lambda: self.faker.sentence(),
+            "paragraph": lambda: self.faker.paragraph(),
+            "text": lambda: self.faker.text(),
+            "username": lambda: self.faker.user_name(),
+            "password": lambda: self.faker.password(),
+            "ip_v4": lambda: self.faker.ipv4(),
+            "ip_v6": lambda: self.faker.ipv6(),
+            "slug": lambda: self.faker.slug(),
+            "color": lambda: self.faker.color_name(),
+            "currency": lambda: self.faker.currency()[0],
+            "currency_code": lambda: self.faker.currency_code(),
+        }
+
+    def generate(self, schema: Dict[str, Any]) -> Any:
+        """Generate data from a JSON schema.
+
+        Args:
+            schema: JSON schema definition
+
+        Returns:
+            Generated data matching the schema
+        """
+        if not schema or not isinstance(schema, dict):
+            return None
+
+        if "$ref" in schema:
+            return self._resolve_ref(schema["$ref"])
+
+        schema_type = schema.get("type")
+
+        if schema_type == "null":
+            return None
+        elif schema_type == "boolean":
+            return self._generate_boolean(schema)
+        elif schema_type == "integer":
+            return self._generate_integer(schema)
+        elif schema_type == "number":
+            return self._generate_number(schema)
+        elif schema_type == "string":
+            return self._generate_string(schema)
+        elif schema_type == "array":
+            return self._generate_array(schema)
+        elif schema_type == "object":
+            return self._generate_object(schema)
+
+        return None
+
+    def _resolve_ref(self, ref: str) -> Any:
+        """Resolve a $ref reference.
+
+        Args:
+            ref: Reference string like #/components/schemas/User
+
+        Returns:
+            Resolved schema or None
+        """
+        parts = ref.lstrip("#/").split("/")
+        
+        skip_prefixes = ["components", "schemas"]
+        start_idx = 0
+        for i, part in enumerate(parts):
+            if i < len(skip_prefixes) and part == skip_prefixes[i]:
+                start_idx = i + 1
+            else:
+                break
+        
+        parts = parts[start_idx:]
+        
+        if not parts:
+            return None
+            
+        current = self._schemas_dict
+
+        for part in parts:
+            if isinstance(current, dict):
+                current = current.get(part, {})
+            else:
+                return None
+
+        if isinstance(current, dict):
+            return self.generate(current)
+
+        return current
+
+    def _generate_boolean(self, schema: Dict[str, Any]) -> bool:
+        """Generate a random boolean."""
+        if "enum" in schema:
+            return random.choice(schema["enum"])
+        return random.choice([True, False])
+
+    def _generate_integer(self, schema: Dict[str, Any]) -> int:
+        """Generate a random integer."""
+        if "enum" in schema:
+            return random.choice(schema["enum"])
+
+        minimum = schema.get("minimum", 0)
+        maximum = schema.get("maximum", 10000)
+        return random.randint(int(minimum), int(maximum))
+
+    def _generate_number(self, schema: Dict[str, Any]) -> float:
+        """Generate a random number."""
+        if "enum" in schema:
+            return random.choice(schema["enum"])
+
+        minimum = schema.get("minimum", 0.0)
+        maximum = schema.get("maximum", 10000.0)
+        return random.uniform(float(minimum), float(maximum))
+
+    def _generate_string(self, schema: Dict[str, Any]) -> str:
+        """Generate a random string."""
+        if "enum" in schema:
+            return random.choice(schema["enum"])
+
+        format_type = schema.get("format", "")
+
+        if format_type in self._faker_providers:
+            return self._faker_providers[format_type]()
+
+        if "pattern" in schema:
+            return self._generate_by_pattern(schema["pattern"])
+
+        min_length = schema.get("minLength", 1)
+        max_length = schema.get("maxLength", 255)
+
+        return self.faker.text(max_nb_chars=random.randint(min_length, max_length))
+
+    def _generate_by_pattern(self, pattern: str) -> str:
+        """Generate a string matching a regex pattern.
+
+        This is a simplified implementation.
+        """
+        return self.faker.word()
+
+    def _generate_array(self, schema: Dict[str, Any]) -> List[Any]:
+        """Generate a random array."""
+        items = schema.get("items", {})
+
+        min_items = schema.get("minItems", 1)
+        max_items = schema.get("maxItems", 10)
+        count = random.randint(int(min_items), int(max_items))
+
+        return [self.generate(items) for _ in range(count)]
+
+    def _generate_object(self, schema: Dict[str, Any]) -> Dict[str, Any]:
+        """Generate a random object."""
+        properties = schema.get("properties", {})
+
+        result = {}
+
+        for prop_name, prop_schema in properties.items():
+            result[prop_name] = self.generate(prop_schema)
+
+        additional_props = schema.get("additionalProperties")
+        if additional_props and isinstance(additional_props, dict):
+            min_props = schema.get("minProperties", 0)
+            max_props = schema.get("maxProperties", 5)
+            count = random.randint(int(min_props), int(max_props))
+            for _ in range(count):
+                result[self.faker.word()] = self.generate(additional_props)
+
+        return result