Changeset ef964d8 in flexoentity for tests


Ignore:
Timestamp:
11/27/25 18:12:23 (7 weeks ago)
Author:
Enrico Schwass <ennoausberlin@…>
Branches:
master
Children:
4e11d58
Parents:
9a50e0b
Message:

new serialization structure adopted and tests fixed

Location:
tests
Files:
5 edited

Legend:

Unmodified
Added
Removed

  • tests/conftest.py

    r9a50e0b ref964d8  
    1 # tests/stubs/single_choice_question.py
    2 from dataclasses import dataclass, field
    31import pytest
    42import platform
    53from pathlib import Path
    64from datetime import datetime
    7 from typing import List
    8 from flexoentity import FlexOID, FlexoEntity, EntityType, EntityState, Domain, get_signing_backend, CertificateReference
     5from flexoentity import Domain, FlexoSignature
     6from flexoentity import get_signing_backend, CertificateReference
    97
    108
    … …  
    1816    return FixedDate
    1917
    20 @dataclass
    21 class AnswerOption:
    22     id: str
    23     text: str
    24     points: float = 0.0
    25 
    26     def to_dict(self):
    27         return {"id": self.id, "text": self.text, "points": self.points}
    28 
    29     @classmethod
    30     def from_dict(cls, data):
    31         return cls(
    32             id=data.get("id", ""),
    33             text=data.get("text", ""),
    34             points=data.get("points", 0.0)
    35         )
    36 
    37 
    38 @dataclass
    39 class SingleChoiceQuestion(FlexoEntity):
    40     """A minimal stub to test FlexoEntity integration."""
    41     ENTITY_TYPE = EntityType.ITEM
    42 
    43     text: str = ""
    44     options: List[AnswerOption] = field(default_factory=list)
    45 
    46     def __post_init__(self):
    47         # If no FlexOID yet, generate a draft ID now.
    48         if not getattr(self, "flexo_id", None):
    49             self.flexo_id = FlexOID.safe_generate(
    50                 domain_id=self.domain_id,
    51                 entity_type=SingleChoiceQuestion.ENTITY_TYPE.value,     # 'I'
    52                 state=EntityState.DRAFT.value,        # 'D'
    53                 text=self.text_seed or self.text,
    54                 version=1,
    55             )
    56 
    57     @classmethod
    58     def default(cls):
    59         return cls()
    60 
    61     def _serialize_content(self):
    62         return {
    63             "text": self.text,
    64             "options": [opt.to_dict() for opt in self.options],
    65         }
    66 
    67     @property
    68     def text_seed(self) -> str:
    69         """Include answer options (and points) for deterministic ID generation."""
    70 
    71         joined = "|".join(
    72             f"{opt.text.strip()}:{opt.points}"
    73             for opt in sorted(self.options, key=lambda o: o.text.strip().lower())
    74         )
    75         return f"{self.text}{joined}"
    76 
    77     @classmethod
    78     def from_dict(cls, data):
    79         obj = cls(text=data.get("text", ""),
    80             options=[AnswerOption.from_dict(o) for o in data.get("options", [])],
    81         )
    82         # restore FlexoEntity core fields
    83         if "flexo_id" in data:
    84             obj.flexo_id = FlexOID.to_dict(data["flexo_id"])
    85         return obj
    8618
    8719@pytest.fixture
    … …  
    9224                                 description="ALL ABOUT ARITHMETIC IN PYTHON")
    9325
    94 @pytest.fixture
    95 def sample_question(sample_domain):
    96     q = SingleChoiceQuestion.with_domain_id(domain_id=sample_domain.domain_id,
    97                                             text="What is 2 + 2?",
    98                                             options=[])
    99     q._update_fingerprint()
    100     return q
    10126
    10227SYSTEM = platform.system()
    … …  
    181106    except Exception as e:
    182107        pytest.skip(f"Backend unavailable or misconfigured: {e}")
     108
     109@pytest.fixture
     110def sample_signature(sample_domain, cert_ref_linux):
     111    return FlexoSignature.with_domain_id(domain_id="SIG", signed_entity=sample_domain,
     112                                         certificate_reference=cert_ref_linux,
     113                                         comment="This is a mock signature")
     114

  • tests/test_flexoid.py

    r9a50e0b ref964d8  
    111111# ──────────────────────────────────────────────
    112112
    113 def test_canonical_seed_for_strings_and_dicts():
     113def test_canonical_seed_for_strings():
    114114    s1 = "  Hello   world "
    115115    s2 = "Hello world"
    116116    assert canonical_seed(s1) == canonical_seed(s2)
    117 
    118     d1 = {"b": 1, "a": 2}
    119     d2 = {"a": 2, "b": 1}
    120     assert canonical_seed(d1) == canonical_seed(d2)
    121 
    122     class Dummy:
    123         def __init__(self):
    124             self.x = 1
    125             self.y = 2
    126     obj = Dummy()
    127     assert isinstance(canonical_seed(obj), str)
    128 
    129117
    130118# ──────────────────────────────────────────────

  • tests/test_id_lifecycle.py

    r9a50e0b ref964d8  
    33
    44
    5 # ──────────────────────────────────────────────────────────────────────────────
    6 # Tests adapted to use real SingleChoiceQuestion fixture instead of DummyEntity
    7 # ──────────────────────────────────────────────────────────────────────────────
     5def test_initial_state(sample_domain):
     6    assert sample_domain.state == EntityState.DRAFT
     7    assert sample_domain.flexo_id.version == 1
     8    assert FlexoEntity.verify_integrity(sample_domain)
    89
    9 def test_initial_state(sample_question):
    10     q = sample_question
    11     assert q.state == EntityState.DRAFT
    12     assert q.flexo_id.version == 1
    13     assert FlexoEntity.verify_integrity(q)
    14 
    15 def test_approval_does_not_bump_version(sample_question):
    16     q = sample_question
     10def test_approval_does_not_bump_version(sample_domain):
     11    q = sample_domain
    1712    q.approve()
    1813    assert q.state == EntityState.APPROVED
    1914    assert q.flexo_id.version == 1
    2015
    21 def test_signing_does_not_bump_version(sample_question):
    22     q = sample_question
     16def test_signing_does_not_bump_version(sample_domain):
     17    q = sample_domain
    2318    q.approve()
    2419    before = q.flexo_id
    … …  
    3833
    3934
    40 def test_publish_does_not_bump_version(sample_question):
    41     q = sample_question
     35def test_publish_does_not_bump_version(sample_domain):
     36    q = sample_domain
    4237    q.approve()
    4338    q.sign()
    … …  
    4843
    4944
    50 def test_modify_content_changes_fingerprint(sample_question):
    51     q = sample_question
    52     q.text += "Rephrased content"  # simulate text change
    53     changed = q._update_fingerprint()
     45def test_modify_content_changes_fingerprint(sample_signature):
     46    sample_signature.comment += "Rephrased content"  # simulate text change
     47    changed = sample_signature._update_fingerprint()
    5448    assert changed
    5549
    5650
    57 def test_no_version_bump_on_draft_edits(sample_question):
    58     q = sample_question
    59     q.text = "Minor draft edit"
    60     q._update_fingerprint()
    61     assert q.flexo_id.version == 1
     51def test_no_version_bump_on_draft_edits(sample_signature):
     52    sample_signature.comment = "Minor draft edit"
     53    sample_signature._update_fingerprint()
     54    assert sample_signature.flexo_id.version == 1
    6255
    6356
    64 def test_version_bump_after_edit_and_sign(sample_question):
    65     q = sample_question
    66     q.approve()
    67     v1 = str(q.flexo_id)
    68     q.text = "Changed content"
    69     q.sign()
    70     assert str(q.flexo_id) != v1
     57def test_version_bump_after_edit_and_sign(sample_signature):
     58    sample_signature.approve()
     59    v1 = str(sample_signature.flexo_id)
     60    sample_signature.comment = "Changed comment"
     61    sample_signature.sign()
     62    assert str(sample_signature.flexo_id) != v1
    7163
    7264
    73 def test_integrity_check_passes_and_fails(sample_question):
    74     q = sample_question
    75     q.approve()
    76     assert FlexoEntity.verify_integrity(q)
     65def test_integrity_check_passes_and_fails(sample_signature):
     66    sample_signature.approve()
     67    assert FlexoEntity.verify_integrity(sample_signature)
    7768
    7869    # simulate tampering
    79     q.text = "Tampered text"
    80     assert not FlexoEntity.verify_integrity(q)
     70    sample_signature.comment = "Tampered text"
     71    assert not FlexoEntity.verify_integrity(sample_signature)
    8172
    8273
    83 def test_obsolete_state(sample_question):
    84     q = sample_question
     74def test_obsolete_state(sample_domain):
     75    q = sample_domain
    8576    q.approve()
    8677    q.sign()
    … …  
    9081
    9182
    92 def test_clone_new_base_resets_lineage(sample_question):
    93     q = sample_question
     83def test_clone_new_base_resets_lineage(sample_domain):
     84    q = sample_domain
    9485    q.approve()
    9586    q.sign()
    … …  
    10293    assert q.flexo_id.version == 1
    10394
    104 def test_clone_new_base_sets_origin(sample_question):
    105     q = sample_question
     95def test_clone_new_base_sets_origin(sample_domain):
     96    q = sample_domain
    10697    q.approve()
    10798    q.sign()
    … …  
    115106    assert q.flexo_id != old_id
    116107
    117 def test_mass_version_increments_until_obsolete(sample_question):
    118     q = sample_question
     108def test_mass_version_increments_until_obsolete(sample_domain):
     109    q = sample_domain
    119110    q.approve()
    120111    for _ in range(FlexOID.MAX_VERSION - 1):

  • tests/test_id_stress.py

    r9a50e0b ref964d8  
    99
    1010import pytest
    11 
    12 from flexoentity import FlexOID, EntityType, EntityState
     11from uuid import uuid4
     12from flexoentity import FlexOID, EntityType, EntityState, FlexoSignature
    1313
    1414logger = logging.getLogger(__name__)
    … …  
    6666    assert id1 == id2
    6767
     68def test_massive_lifecycle_simulation(cert_ref_linux, sample_domain):
     69    """
     70    Generate 100 random FlexoSignatures, mutate content, run through lifecycle,
     71    and ensure all FlexOIDs are unique and valid.
     72    """
     73    entities = []
    6874
    69 def test_massive_lifecycle_simulation(sample_question):
    70     """
    71     Generate 100 random SingleChoiceQuestions, simulate multiple edits and state transitions,
    72     ensure all final IDs and fingerprints are unique and valid.
    73     """
    74     entities = [
    75         copy.deepcopy(sample_question) for _ in range(100)
    76     ]
     75    for i in range(100):
     76        sig = FlexoSignature.with_domain_id(
     77            domain_id="SIGTEST",
     78            signed_entity=sample_domain.flexo_id,
     79            signer_id=uuid4(),
     80            certificate_reference=cert_ref_linux,
     81            comment=f"Initial signature #{i}"
     82        )
     83        entities.append(sig)
    7784
     85    # Mutate + lifecycle transitions
    7886    for i, e in enumerate(entities):
    79         # random edit
    80         e.text += f" updated #{i}"
     87        # CONTENT CHANGE → fingerprint changes → hash → FlexOID.prefix changes
     88        e.comment += f" updated-{i}"
    8189        e._update_fingerprint()
    8290
    … …  
    8896            e.publish()
    8997
    90     flexoids = [e.flexo_id for e in entities]
    91     assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs after lifecycle simulation"
     98    # Check ID uniqueness
     99    flexoids = [str(e.flexo_id) for e in entities]
     100    assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs detected"

  • tests/test_persistance_integrity.py

    r9a50e0b ref964d8  
    66import pytest
    77
    8 from flexoentity import EntityState, EntityType, Domain
     8from flexoentity import Domain
     9
    910
    1011@pytest.fixture
    11 def approved_question():
    12     """Provide a fully approved and published SingleChoiceQuestion for persistence tests."""
    13     q = SingleChoiceQuestion(
    14         domain=Domain(domain="GEN", entity_type=EntityType.DOMAIN, state=EntityState.DRAFT),
    15         entity_type=None,  # SingleChoiceQuestion sets this internally to EntityType.ITEM
    16         state=EntityState.DRAFT,
    17         text="What is Ohm’s law?",
    18         options=[
    19             AnswerOption(id="OP1", text="U = R × I", points=1),
    20             AnswerOption(id="OP2", text="U = I / R", points=0),
    21             AnswerOption(id="OP3", text="R = U × I", points=0),
    22         ],
    23     )
    24     q.approve()
    25     q.sign()
    26     q.publish()
    27     return q
     12def approved_domain(sample_domain):
     13    """Provide a fully approved and published Domain for persistence tests."""
     14    sample_domain.approve()
     15    sample_domain.sign()
     16    sample_domain.publish()
     17    return sample_domain
    2818
    2919
    30 @pytest.mark.skip(reason="FlexOIDs regenerated on import; enable once JSON format is stable")
    31 def test_json_roundtrip_preserves_integrity(approved_question):
     20def test_json_roundtrip_preserves_integrity(approved_domain):
    3221    """
    3322    Export to JSON and reload — ensure fingerprints and signatures remain valid.
    3423    """
    35     json_str = approved_question.to_json()
    36     loaded = SingleChoiceQuestion.from_json(json_str)
     24    json_str = approved_domain.to_json()
     25    loaded = Domain.from_json(json_str)
    3726
    3827    # Fingerprint and state should match — integrity must pass
    39     assert SingleChoiceQuestion.verify_integrity(loaded)
     28    assert Domain.verify_integrity(loaded)
    4029
    4130    # Metadata should be preserved exactly
    42     assert approved_question.fingerprint == loaded.fingerprint
    43     assert approved_question.flexo_id == loaded.flexo_id
    44     assert loaded.state == approved_question.state
     31    assert approved_domain.fingerprint == loaded.fingerprint
     32    assert approved_domain.flexo_id == loaded.flexo_id
     33    assert loaded.state == approved_domain.state
    4534
    46 @pytest.mark.skip(reason="FlexOIDs regenerated on import; tampering detection not yet implemented")
    47 def test_json_tampering_detection(approved_question):
     35
     36def test_json_tampering_detection(approved_domain):
    4837    """Tampering with content should invalidate fingerprint verification."""
    49     json_str = approved_question.to_json()
     38    json_str = approved_domain.to_json()
    5039    tampered = json.loads(json_str)
    51     tampered["text"] = "Tampered content injection"
     40    tampered["content"]["fullname"] = "Tampered content injection"
    5241    tampered_json = json.dumps(tampered)
    5342
    54     loaded = SingleChoiceQuestion.from_json(tampered_json)
    55     assert not SingleChoiceQuestion.verify_integrity(loaded)
     43    loaded = Domain.from_json(tampered_json)
     44    assert not Domain.verify_integrity(loaded)
    5645
    57 @pytest.mark.skip(reason="FlexOIDs regenerated on import; corruption detection not yet applicable")
    58 def test_json_file_corruption(approved_question, tmp_path):
     46
     47def test_json_file_corruption(approved_domain, tmp_path):
    5948    """Simulate file corruption — integrity check must fail."""
    6049    file = tmp_path / "question.json"
    61     json_str = approved_question.to_json()
     50    json_str = approved_domain.to_json()
     51    print("JSON", json_str)
    6252    file.write_text(json_str)
    6353
    6454    # Corrupt the file (simulate accidental byte modification)
    65     corrupted = json_str.replace("Ohm’s", "Omm’s")
     55    corrupted = json_str.replace("ARITHMETIC", "ARITHM")
    6656    file.write_text(corrupted)
    6757
    68     loaded = SingleChoiceQuestion.from_json(file.read_text())
    69     assert not SingleChoiceQuestion.verify_integrity(loaded)
     58    loaded = Domain.from_json(file.read_text())
     59    assert not Domain.verify_integrity(loaded)
Note: See TracChangeset for help on using the changeset viewer.