source: flexoentity/tests/test_id_stress.py@ 8aea9a0

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

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from flexoentity import FlexOID, EntityType, EntityState, FlexoEntity


# ──────────────────────────────────────────────────────────────────────────────
def test_bulk_generation_uniqueness():
    """Generate 10,000 IDs and assert uniqueness (statistical test)."""
    domain = "AF"
    etype = EntityType.QUESTION
    estate = EntityState.DRAFT
    seeds = [f"question {i}" for i in range(10_000)]

    ids = []
    for seed in seeds:
        flexo_id = FlexOID.generate(domain, etype, estate, seed)
        ids.append(flexo_id)

    assert len(ids) == len(set(ids)), "ID collisions detected in bulk generation"

    def test_disambiguator_trigger():
        """
        Generating the same entity twice with same inputs yields identical ID.
        (No runtime disambiguation; IDs are deterministic by design.)
        """
        domain = "AF"
        etype = EntityType.QUESTION
        estate = EntityState.DRAFT
        text = "identical question text"
        id1 = FlexOID.generate(domain, etype, estate, text)
        id2 = FlexOID.generate(domain, etype, estate, text)
        # IDs must be identical, because we now enforce determinism, not randomization
        assert id1 == id2
        assert id1.signature == id2.signature
    
def test_id_reproducibility_across_runs():
    """
    The same seed on a new process (fresh _seen_hashes)
    should yield the same base ID (without suffix).
    """
    domain = "AF"
    etype = EntityType.CATALOG
    estate = EntityState.DRAFT
    seed = "reproducibility test seed"
    id1 = FlexOID.generate(domain, etype, estate, seed)
    # Reset hash cache
    FlexOID._seen_hashes.clear()
    id2 = FlexOID.generate(domain, etype, estate, seed)
    assert id1 == id2
    assert id1.signature == id2.signature


def test_version_ceiling_enforcement():
    """Simulate approaching @999 to trigger obsolescence guard."""
    entity = FlexoEntity("AF", EntityType.EXAM, "Final Exam 2025", EntityState.DRAFT)
    entity.approve()
    # artificially bump version number to near ceiling

    entity.flexo_id = FlexOID.from_oid_and_version(entity.flexo_id, 998)

    # 998 → 999 is allowed
    entity.sign()
    assert entity.flexo_id.version == 999

    # 999 → 1000 should raise RuntimeError
    with pytest.raises(RuntimeError):
        entity.sign()


def test_massive_lifecycle_simulation():
    """
    Generate 100 random entities, simulate multiple edits and state transitions,
    ensure all final IDs and fingerprints are unique and valid.
    """
    import random
    texts = [f"random question {i}" for i in range(100)]
    entities = [FlexoEntity("AF", EntityType.QUESTION, t) for t in texts]

    for e in entities:
        # random edit, approval, signing
        e.modify_content(e.text_seed + " updated")
        e.approve()
        if random.random() > 0.3:
            e.sign()
        if random.random() > 0.6:
            e.publish()

    ids = [e.flexo_id for e in entities]
    fps = [e.flexo_id.signature for e in entities]
    assert len(ids) == len(set(ids)), "Duplicate IDs after random lifecycle"
    assert len(fps) == len(set(fps)), "Duplicate fingerprints after random lifecycle"