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

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

import copy
import logging
import random

import pytest

from flexoentity import FlexOID, EntityType, EntityState

logger = logging.getLogger(__name__)

def test_bulk_generation_uniqueness(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(domain.domain, 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("GEN") for id_str in ids)
    assert all("@" in id_str for id_str in ids)

def test_id_generation_is_deterministic(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(domain.domain, entity_type.value, estate.value, text)
    id2 = FlexOID.generate(domain.domain, entity_type.value, estate.value, text)
    # IDs must be identical because generation is deterministic
    assert id1 == id2


# def test_id_reproducibility_across_runs(domain):
#     """
#     The same seed on a new process (fresh _seen_hashes)
#     should yield the same base ID (without suffix).
#     """
#     entity_type = EntityType.CATALOG
#     estate = EntityState.DRAFT
#     seed = "reproducibility test seed"

#     id1 = FlexOID.generate(domain.domain, entity_type.value, estate.value, seed)
#     FlexOID._seen_hashes.clear()
#     id2 = FlexOID.generate(domain.domain, entity_type.value, estate.value, seed)

#     assert id1 == id2


def test_version_ceiling_enforcement(sample_question):
    """Simulate approaching @999 to trigger obsolescence guard."""
    q = sample_question
    q.approve()

    # artificially bump version number to near ceiling
    q.flexo_id = FlexOID.from_oid_and_version(q.flexo_id, 998)

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

    # 999 → 1000 should raise RuntimeError
    with pytest.raises(ValueError):
        q.publish()


def test_massive_lifecycle_simulation(sample_question):
    """
    Generate 100 random SingleChoiceQuestions, simulate multiple edits and state transitions,
    ensure all final IDs and fingerprints are unique and valid.
    """
    entities = [
        copy.deepcopy(sample_question) for _ in range(100)
    ]

    for i, e in enumerate(entities):
        # random edit
        e.text += f" updated #{i}"
        e._update_fingerprint()

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

    flexoids = [e.flexo_id for e in entities]
    assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs after lifecycle simulation"