source: flexoentity/tests/test_id_stress.py@ 4e11d58

"""
Stress tests for the Flex-O ID lifecycle.
Focus: collision avoidance, version ceiling, reproducibility.
"""

import copy
import logging
import random

import pytest
from uuid import uuid4
from flexoentity import FlexOID, EntityType, EntityState, FlexoSignature

logger = logging.getLogger(__name__)

def test_bulk_generation_uniqueness(sample_domain):
    """
    Generate 100,000 IDs and ensure uniqueness using safe_generate().
    If a collision occurs, safe_generate() must resolve it automatically
    via salt + date adjustment.
    """
    entity_type = EntityType.ITEM
    estate = EntityState.DRAFT
    seeds = [f"question {i}" for i in range(100000)]

    # Simulate a simple in-memory repository for collision detection
    repo = {}

    def repo_get(oid_str):
        return repo.get(str(oid_str))

    # Generate IDs using safe_generate
    ids = []
    for seed in seeds:
        oid = FlexOID.safe_generate(sample_domain.domain_id, entity_type.value,
                                    estate.value, seed, repo=repo)
        assert isinstance(oid, FlexOID)
        ids.append(str(oid))
        repo[str(oid)] = oid  # register for future collision detection

    unique_count = len(set(ids))
    total_count = len(ids)
    collisions = total_count - unique_count

    logger.info(f"Generated {total_count} IDs ({collisions} collisions handled).")

    # Assert that safe_generate avoided duplicates
    assert total_count == unique_count, f"Unexpected duplicate IDs ({collisions} found)"

    # Sanity check: IDs should look canonical
    # assert all(id_str.startswith("GENERIC") for id_str in ids)
    assert all("@" in id_str for id_str in ids)

def test_id_generation_is_deterministic(sample_domain):
    """
    Generating the same entity twice with same inputs yields identical ID.
    (No runtime disambiguation; IDs are deterministic by design.)
    """
    entity_type = EntityType.ITEM
    estate = EntityState.DRAFT
    text = "identical question text"

    id1 = FlexOID.generate(sample_domain.domain_id, entity_type.value, estate.value, text)
    id2 = FlexOID.generate(sample_domain.domain_id, entity_type.value, estate.value, text)
    # IDs must be identical because generation is deterministic
    assert id1 == id2

def test_massive_lifecycle_simulation(cert_ref_linux, sample_domain):
    """
    Generate 100 random FlexoSignatures, mutate content, run through lifecycle,
    and ensure all FlexOIDs are unique and valid.
    """
    entities = []

    for i in range(100):
        sig = FlexoSignature.with_domain_id(
            domain_id="SIGTEST",
            signed_entity=sample_domain.flexo_id,
            signer_id=uuid4(),
            certificate_reference=cert_ref_linux,
            comment=f"Initial signature #{i}"
        )
        entities.append(sig)

    # Mutate + lifecycle transitions
    for i, e in enumerate(entities):
        # CONTENT CHANGE → fingerprint changes → hash → FlexOID.prefix changes
        e.comment += f" updated-{i}"
        e._update_fingerprint()

        # lifecycle transitions
        e.approve()
        if random.random() > 0.3:
            e.sign()
        if random.random() > 0.6:
            e.publish()

    # Check ID uniqueness
    flexoids = [str(e.flexo_id) for e in entities]
    assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs detected"