From bf7eaa4194df583262d34000bb64c308d7fafc19 Mon Sep 17 00:00:00 2001 From: sbsmith09 Date: Tue, 6 May 2025 21:08:05 +0000 Subject: [PATCH 1/4] Add TransactionEvidence model with uniqueness schema --- .../database/transaction_evidence.py | 62 +++++++++++++++++++ 1 file changed, 62 insertions(+) create mode 100644 agent-framework/prometheus_swarm/database/transaction_evidence.py 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..78be88f6 --- /dev/null +++ b/agent-framework/prometheus_swarm/database/transaction_evidence.py @@ -0,0 +1,62 @@ +from sqlalchemy import Column, String, DateTime, UUID, ForeignKey, UniqueConstraint +from sqlalchemy.orm import relationship +from sqlalchemy.sql import func +from .database import Base +import uuid + +class TransactionEvidence(Base): + """ + Database model for Transaction Evidence with uniqueness constraints. + + Ensures each transaction evidence record is unique based on multiple attributes. + """ + __tablename__ = 'transaction_evidence' + + # Primary Key + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + + # Transaction-specific fields + transaction_hash = Column(String, nullable=False) + transaction_type = Column(String, nullable=False) + + # Timestamp fields + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), onupdate=func.now()) + + # Optional foreign key reference (adjust as needed) + # user_id = Column(UUID(as_uuid=True), ForeignKey('users.id'), nullable=True) + + # Uniqueness Constraints + __table_args__ = ( + # Ensure unique combination of transaction hash and type + UniqueConstraint('transaction_hash', 'transaction_type', name='uq_transaction_evidence'), + ) + + def __repr__(self): + """ + String representation of the TransactionEvidence model. + + Returns: + str: A string describing the transaction evidence + """ + return f"" + + @classmethod + def validate_transaction_evidence(cls, transaction_hash, transaction_type): + """ + 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") + + # Add any additional validation logic here \ No newline at end of file From baf0a14b1144e060a6f59881ce86339e376da5cf Mon Sep 17 00:00:00 2001 From: sbsmith09 Date: Tue, 6 May 2025 21:08:22 +0000 Subject: [PATCH 2/4] Add comprehensive tests for TransactionEvidence model --- .../database/test_transaction_evidence.py | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 agent-framework/tests/unit/database/test_transaction_evidence.py 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..1c50d91b --- /dev/null +++ b/agent-framework/tests/unit/database/test_transaction_evidence.py @@ -0,0 +1,96 @@ +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from prometheus_swarm.database.database import Base +from prometheus_swarm.database.transaction_evidence import TransactionEvidence +import uuid + +@pytest.fixture(scope='function') +def db_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) + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + session = Session() + + yield session + + session.close() + Base.metadata.drop_all(engine) + +def test_create_transaction_evidence(db_session): + """ + Test creating a basic transaction evidence record. + """ + evidence = TransactionEvidence( + transaction_hash='abc123', + transaction_type='transfer' + ) + + db_session.add(evidence) + db_session.commit() + + assert evidence.id is not None + assert evidence.transaction_hash == 'abc123' + assert evidence.transaction_type == 'transfer' + +def test_uniqueness_constraint(db_session): + """ + Test that a duplicate transaction evidence record is not allowed. + """ + # First record + evidence1 = TransactionEvidence( + transaction_hash='unique_hash', + transaction_type='transfer' + ) + db_session.add(evidence1) + db_session.commit() + + # Try to add a duplicate record (same hash and type) + with pytest.raises(Exception) as excinfo: + evidence2 = TransactionEvidence( + transaction_hash='unique_hash', + transaction_type='transfer' + ) + db_session.add(evidence2) + db_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') + + 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) + +def test_transaction_evidence_repr(): + """ + Test the string representation of TransactionEvidence. + """ + evidence = TransactionEvidence( + id=uuid.uuid4(), + transaction_hash='test_hash', + transaction_type='test_type' + ) + + repr_str = repr(evidence) + assert 'TransactionEvidence' in repr_str + assert 'test_hash' in repr_str + assert 'test_type' in repr_str \ No newline at end of file From 26d9a25f5fd5ad443392f8764cda38372f0c3e9e Mon Sep 17 00:00:00 2001 From: sbsmith09 Date: Tue, 6 May 2025 21:08:41 +0000 Subject: [PATCH 3/4] Update TransactionEvidence model to use SQLModel --- .../database/transaction_evidence.py | 44 +++++++------------ 1 file changed, 15 insertions(+), 29 deletions(-) diff --git a/agent-framework/prometheus_swarm/database/transaction_evidence.py b/agent-framework/prometheus_swarm/database/transaction_evidence.py index 78be88f6..17de1d08 100644 --- a/agent-framework/prometheus_swarm/database/transaction_evidence.py +++ b/agent-framework/prometheus_swarm/database/transaction_evidence.py @@ -1,48 +1,34 @@ -from sqlalchemy import Column, String, DateTime, UUID, ForeignKey, UniqueConstraint -from sqlalchemy.orm import relationship -from sqlalchemy.sql import func -from .database import Base +"""Transaction Evidence Model.""" + +from datetime import datetime +from typing import Optional +from sqlmodel import SQLModel, Field, UniqueConstraint import uuid -class TransactionEvidence(Base): +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 = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True) # Transaction-specific fields - transaction_hash = Column(String, nullable=False) - transaction_type = Column(String, nullable=False) + transaction_hash: str = Field(index=True, unique=True) + transaction_type: str # Timestamp fields - created_at = Column(DateTime(timezone=True), server_default=func.now()) - updated_at = Column(DateTime(timezone=True), onupdate=func.now()) - - # Optional foreign key reference (adjust as needed) - # user_id = Column(UUID(as_uuid=True), ForeignKey('users.id'), nullable=True) - - # Uniqueness Constraints - __table_args__ = ( - # Ensure unique combination of transaction hash and type - UniqueConstraint('transaction_hash', 'transaction_type', name='uq_transaction_evidence'), - ) + created_at: datetime = Field(default_factory=datetime.utcnow) + updated_at: Optional[datetime] = None - def __repr__(self): - """ - String representation of the TransactionEvidence model. - - Returns: - str: A string describing the transaction evidence - """ - return f"" + # Optional additional fields can be added here @classmethod - def validate_transaction_evidence(cls, transaction_hash, transaction_type): + def validate_transaction_evidence(cls, transaction_hash: str, transaction_type: str): """ Validate transaction evidence before creation. @@ -59,4 +45,4 @@ def validate_transaction_evidence(cls, transaction_hash, transaction_type): if not transaction_type or not isinstance(transaction_type, str): raise ValueError("Transaction type must be a non-empty string") - # Add any additional validation logic here \ No newline at end of file + # Optional: Add more specific validation rules here \ No newline at end of file From 347074f39f20d05a8af463a8da3915ae61a02158 Mon Sep 17 00:00:00 2001 From: sbsmith09 Date: Tue, 6 May 2025 21:08:58 +0000 Subject: [PATCH 4/4] Update test file for TransactionEvidence model using SQLModel --- .../database/test_transaction_evidence.py | 58 +++++++++---------- 1 file changed, 27 insertions(+), 31 deletions(-) diff --git a/agent-framework/tests/unit/database/test_transaction_evidence.py b/agent-framework/tests/unit/database/test_transaction_evidence.py index 1c50d91b..764a1258 100644 --- a/agent-framework/tests/unit/database/test_transaction_evidence.py +++ b/agent-framework/tests/unit/database/test_transaction_evidence.py @@ -1,27 +1,23 @@ +"""Tests for Transaction Evidence Model.""" + import pytest -from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker -from prometheus_swarm.database.database import Base +from sqlmodel import create_engine, SQLModel, Session from prometheus_swarm.database.transaction_evidence import TransactionEvidence import uuid @pytest.fixture(scope='function') -def db_session(): +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) - Base.metadata.create_all(engine) - Session = sessionmaker(bind=engine) - session = Session() - - yield session - - session.close() - Base.metadata.drop_all(engine) + SQLModel.metadata.create_all(engine) + with Session(engine) as session: + yield session + SQLModel.metadata.drop_all(engine) -def test_create_transaction_evidence(db_session): +def test_create_transaction_evidence(session): """ Test creating a basic transaction evidence record. """ @@ -30,14 +26,14 @@ def test_create_transaction_evidence(db_session): transaction_type='transfer' ) - db_session.add(evidence) - db_session.commit() + 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(db_session): +def test_uniqueness_constraint(session): """ Test that a duplicate transaction evidence record is not allowed. """ @@ -46,17 +42,17 @@ def test_uniqueness_constraint(db_session): transaction_hash='unique_hash', transaction_type='transfer' ) - db_session.add(evidence1) - db_session.commit() + session.add(evidence1) + session.commit() - # Try to add a duplicate record (same hash and type) + # Try to add a duplicate record (same hash) with pytest.raises(Exception) as excinfo: evidence2 = TransactionEvidence( transaction_hash='unique_hash', - transaction_type='transfer' + transaction_type='another_transfer' ) - db_session.add(evidence2) - db_session.commit() + session.add(evidence2) + session.commit() assert 'UNIQUE constraint' in str(excinfo.value) @@ -72,25 +68,25 @@ def test_validate_transaction_evidence(): TransactionEvidence.validate_transaction_evidence('', 'transfer') with pytest.raises(ValueError, match="Transaction hash must be a non-empty string"): - TransactionEvidence.validate_transaction_evidence(None, 'transfer') + 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) + TransactionEvidence.validate_transaction_evidence('hash', None) # type: ignore -def test_transaction_evidence_repr(): +def test_transaction_evidence_properties(): """ - Test the string representation of TransactionEvidence. + Test additional properties of TransactionEvidence. """ evidence = TransactionEvidence( - id=uuid.uuid4(), transaction_hash='test_hash', transaction_type='test_type' ) - repr_str = repr(evidence) - assert 'TransactionEvidence' in repr_str - assert 'test_hash' in repr_str - assert 'test_type' in repr_str \ No newline at end of file + 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