# Created by AG on 21-12-2025 from __future__ import annotations import base64 import random import string from typing import Any from datetime import date, datetime, timezone def _rand_bytes_to_b64( rng: random.Random, rand_min: int = 0, rand_max: int = 16 ) -> str: raw_data = bytes( rng.getrandbits(8) for _ in range(rng.randint(rand_min, rand_max)) ) return base64.b64encode(raw_data).decode("ascii") def _rand_fixed_bytes_to_b64( rng: random.Random, length: int ) -> str: raw_data = bytes( rng.getrandbits(8) for _ in range(length) ) return base64.b64encode(raw_data).decode("ascii") def _generate_random_string( rng: random.Random, minimum_length: int = 4, maximum_length: int = 20 ) -> str: randnum = rng.randint(minimum_length, maximum_length) alphanum = string.ascii_letters + string.digits return "".join(rng.choice(alphanum) for _ in range(randnum)) def _rand_email_gen( rng: random.Random ) -> str: email_length = rng.randint(3, 12) email_characters = string.ascii_lowercase + string.digits email = "".join(rng.choice(email_characters) for _ in range(email_length)) email_providers = [ "gmail.com", "yahoo.com", "outlook.com", "icloud.com", "proton.me", "hotmail.com" ] return f"{email}@{rng.choice(email_providers)}" def _is_branch_empty(branch: Any) -> bool: return branch == "null" or \ (isinstance(branch, dict) and branch.get("type") == "null") def _union_contains_string(contract: Any) -> bool: return isinstance(contract, list) and \ any( branch == "string" or (isinstance(branch, dict) and branch.get("type") == "string") for branch in contract ) def _is_string_valid_contract(contract: Any) -> bool: return contract == "string" or \ (isinstance(contract, dict) and contract.get("type") == "string") def _is_email_field_valid(field_name: str) -> bool: field_name = field_name.lower() return field_name == "email" or field_name.endswith("_email") or field_name.endswith("email") def _choose_union_branching( branches: list[Any], rng: random.Random ) -> Any: from src.utils import _probabilistic_choice empty_branches = [branch for branch in branches if _is_branch_empty(branch)] non_empty_branches = [branch for branch in branches if branch not in empty_branches] if empty_branches and _probabilistic_choice(rng, probability=0.25): return empty_branches[0] return rng.choice(non_empty_branches if non_empty_branches else branches) def _is_union_nullable(field_contract: Any) -> bool: check: bool = isinstance(field_contract, list) and any( branch == "null" or (isinstance(branch, dict) and branch.get("type") == "null") for branch in field_contract ) return check def _generate_primitive_data_type( contract: str, rng: random.Random ) -> Any: if contract == "null": return None if contract == "boolean": return bool(rng.randint(0, 1)) if contract == "int": return rng.randint(-1000, 5000) if contract == "long": return rng.randint(-10000, 200000) if contract in ["float", "double"]: return rng.uniform(-1000.0, 5000.0) if contract == "string": return _generate_random_string(rng=rng, minimum_length=5) if contract == "bytes": return _rand_bytes_to_b64(rng) return _generate_random_string(rng) def _generate_wrong_data_type( contract: Any, rng: random.Random ) -> Any: if isinstance(contract, list): return { "invalid": "data_type" } if isinstance(contract, str): if contract in ("int", "long", "float", "double"): return "InvalidNumber" if contract == "boolean": return "InvalidBoolean" if contract == "string": return 1337 return { "invalid": True } if isinstance(contract, dict): data_type = contract.get("type") if data_type == "array": return "InvalidArray" if data_type == "map": return ["Invalid", "Map"] if data_type == "record": return "InvalidRecord" if data_type == "enum": return "INVALID_SYMBOL" return { "invalid": "data_type" } return { "invalid": "dict" }