diff --git a/agent-framework/prometheus_swarm/database/transaction_evidence.py b/agent-framework/prometheus_swarm/database/transaction_evidence.py new file mode 100644 index 00000000..17de1d08 --- /dev/null +++ b/agent-framework/prometheus_swarm/database/transaction_evidence.py @@ -0,0 +1,48 @@ +"""Transaction Evidence Model.""" + +from datetime import datetime +from typing import Optional +from sqlmodel import SQLModel, Field, UniqueConstraint +import uuid + +class TransactionEvidence(SQLModel, table=True): + """ + Database model for Transaction Evidence with uniqueness constraints. + + Ensures each transaction evidence record is unique based on multiple attributes. + """ + # Custom table name in the database + __tablename__ = 'transaction_evidence' + + # Primary Key + id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True) + + # Transaction-specific fields + transaction_hash: str = Field(index=True, unique=True) + transaction_type: str + + # Timestamp fields + created_at: datetime = Field(default_factory=datetime.utcnow) + updated_at: Optional[datetime] = None + + # Optional additional fields can be added here + + @classmethod + def validate_transaction_evidence(cls, transaction_hash: str, transaction_type: str): + """ + Validate transaction evidence before creation. + + Args: + transaction_hash (str): Unique hash of the transaction + transaction_type (str): Type of the transaction + + Raises: + ValueError: If transaction evidence is invalid + """ + if not transaction_hash or not isinstance(transaction_hash, str): + raise ValueError("Transaction hash must be a non-empty string") + + if not transaction_type or not isinstance(transaction_type, str): + raise ValueError("Transaction type must be a non-empty string") + + # Optional: Add more specific validation rules here \ No newline at end of file diff --git a/agent-framework/tests/unit/database/test_transaction_evidence.py b/agent-framework/tests/unit/database/test_transaction_evidence.py new file mode 100644 index 00000000..764a1258 --- /dev/null +++ b/agent-framework/tests/unit/database/test_transaction_evidence.py @@ -0,0 +1,92 @@ +"""Tests for Transaction Evidence Model.""" + +import pytest +from sqlmodel import create_engine, SQLModel, Session +from prometheus_swarm.database.transaction_evidence import TransactionEvidence +import uuid + +@pytest.fixture(scope='function') +def session(): + """ + Create a new database session for each test function. + """ + # Use an in-memory SQLite database for testing + engine = create_engine('sqlite:///:memory:', echo=True) + SQLModel.metadata.create_all(engine) + with Session(engine) as session: + yield session + SQLModel.metadata.drop_all(engine) + +def test_create_transaction_evidence(session): + """ + Test creating a basic transaction evidence record. + """ + evidence = TransactionEvidence( + transaction_hash='abc123', + transaction_type='transfer' + ) + + session.add(evidence) + session.commit() + + assert evidence.id is not None + assert evidence.transaction_hash == 'abc123' + assert evidence.transaction_type == 'transfer' + +def test_uniqueness_constraint(session): + """ + Test that a duplicate transaction evidence record is not allowed. + """ + # First record + evidence1 = TransactionEvidence( + transaction_hash='unique_hash', + transaction_type='transfer' + ) + session.add(evidence1) + session.commit() + + # Try to add a duplicate record (same hash) + with pytest.raises(Exception) as excinfo: + evidence2 = TransactionEvidence( + transaction_hash='unique_hash', + transaction_type='another_transfer' + ) + session.add(evidence2) + session.commit() + + assert 'UNIQUE constraint' in str(excinfo.value) + +def test_validate_transaction_evidence(): + """ + Test validation method for transaction evidence. + """ + # Valid inputs + TransactionEvidence.validate_transaction_evidence('valid_hash', 'transfer') + + # Test invalid inputs + with pytest.raises(ValueError, match="Transaction hash must be a non-empty string"): + TransactionEvidence.validate_transaction_evidence('', 'transfer') + + with pytest.raises(ValueError, match="Transaction hash must be a non-empty string"): + TransactionEvidence.validate_transaction_evidence(None, 'transfer') # type: ignore + + with pytest.raises(ValueError, match="Transaction type must be a non-empty string"): + TransactionEvidence.validate_transaction_evidence('hash', '') + + with pytest.raises(ValueError, match="Transaction type must be a non-empty string"): + TransactionEvidence.validate_transaction_evidence('hash', None) # type: ignore + +def test_transaction_evidence_properties(): + """ + Test additional properties of TransactionEvidence. + """ + evidence = TransactionEvidence( + transaction_hash='test_hash', + transaction_type='test_type' + ) + + assert evidence.transaction_hash == 'test_hash' + assert evidence.transaction_type == 'test_type' + assert evidence.created_at is not None + assert evidence.updated_at is None + assert isinstance(evidence.id, uuid.UUID) \ No newline at end of file