source: flexoentity/tests/test_id_stress.py@ 2f650ac

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

from flexoentity import FlexOID, EntityType, EntityState, Domain

from tests.conftest import DummyEntity

# ──────────────────────────────────────────────────────────────────────────────
def test_bulk_generation_uniqueness():
    """Generate 10,000 IDs and assert uniqueness (statistical test)."""
    domain = Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT,
                    fullname="Signal Corps", classification="RESTRICTED", owner="MESE")

    etype = EntityType.QUESTION
    estate = EntityState.DRAFT
    seeds = [f"question {i}" for i in range(10_000)]

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

    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 = Domain(domain="SIG", etype=EntityType.DOMAIN, state=EntityState.DRAFT,
                    fullname="Signal Corps", classification="RESTRICTED")
    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 = DummyEntity(domain="AF", etype=EntityType.EXAM, state=EntityState.DRAFT, seed="Final Exam 2025")
    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.
    """
    entities = [DummyEntity(domain="AF", etype=EntityType.QUESTION, state=EntityState.DRAFT, seed=f"random question {i}") for i in range(100)]

    for e in entities:
        # random edit, approval, signing
        e._seed += " updated"
        e._update_fingerprint()
        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"