""" Comprehensive tests for file integrity and report SQLAlchemy models. Tests cover structural inspection (columns, types, constraints, defaults, indexes, foreign keys) and __repr__ output for: - FileIntegrityRecord - FileVerificationFailure - Report - ReportSection All tests run WITHOUT a real database by inspecting the SQLAlchemy `__table__` metadata and by instantiating model objects with keyword args. """ import pytest from sqlalchemy import Boolean, Float, Integer, String, Text from sqlalchemy import JSON as SA_JSON from sqlalchemy_utc import UtcDateTime from local_deep_research.database.models.file_integrity import ( FileIntegrityRecord, FileVerificationFailure, ) from local_deep_research.database.models.reports import ( Report, ReportSection, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _col(model, name): """Return a column object from the model's table metadata.""" return model.__table__.columns[name] def _col_names(model): """Return the set of all column names for a model.""" return set(model.__table__.columns.keys()) def _pk_col_names(model): """Return the set of primary key column names.""" return {c.name for c in model.__table__.primary_key.columns} def _fk_targets(model, col_name): """Return the set of foreign key target strings for a column.""" return {fk.target_fullname for fk in _col(model, col_name).foreign_keys} def _has_index(model, col_name): """Check whether a column participates in at least one index.""" for index in model.__table__.indexes: if col_name in {c.name for c in index.columns}: return True return False def _has_unique_constraint(model, col_name): """Check whether a column has a unique constraint (column-level or table-level).""" col = _col(model, col_name) if col.unique: return True for uc in model.__table__.constraints: if hasattr(uc, "columns") and col_name in {c.name for c in uc.columns}: if getattr(uc, "_pending_colargs", None) is not None: continue from sqlalchemy import UniqueConstraint if isinstance(uc, UniqueConstraint): return True return False # ============================================================================ # FileIntegrityRecord # ============================================================================ class TestFileIntegrityRecordTableName: """Verify the __tablename__ value.""" def test_tablename(self): assert FileIntegrityRecord.__tablename__ == "file_integrity_records" class TestFileIntegrityRecordColumns: """Verify that all expected columns exist.""" EXPECTED_COLUMNS = { "id", "file_path", "file_type", "checksum", "algorithm", "file_size", "file_mtime", "verify_on_load", "allow_modifications", "total_verifications", "last_verified_at", "last_verification_passed", "consecutive_successes", "consecutive_failures", "created_at", "updated_at", "related_entity_type", "related_entity_id", } def test_all_expected_columns_present(self): actual = _col_names(FileIntegrityRecord) missing = self.EXPECTED_COLUMNS - actual assert not missing, f"Missing columns: {missing}" def test_no_unexpected_columns(self): actual = _col_names(FileIntegrityRecord) extra = actual - self.EXPECTED_COLUMNS assert not extra, f"Unexpected columns: {extra}" class TestFileIntegrityRecordColumnTypes: """Verify the SQLAlchemy column type for each column.""" @pytest.mark.parametrize( "col_name, expected_type", [ ("id", Integer), ("file_path", Text), ("file_type", String), ("checksum", String), ("algorithm", String), ("file_size", Integer), ("file_mtime", Float), ("verify_on_load", Boolean), ("allow_modifications", Boolean), ("total_verifications", Integer), ("last_verified_at", UtcDateTime), ("last_verification_passed", Boolean), ("consecutive_successes", Integer), ("consecutive_failures", Integer), ("created_at", UtcDateTime), ("updated_at", UtcDateTime), ("related_entity_type", String), ("related_entity_id", Integer), ], ) def test_column_type(self, col_name, expected_type): col = _col(FileIntegrityRecord, col_name) assert isinstance(col.type, expected_type), ( f"Column '{col_name}' type is {type(col.type).__name__}, " f"expected {expected_type.__name__}" ) class TestFileIntegrityRecordStringLengths: """Verify length limits on String columns.""" @pytest.mark.parametrize( "col_name, expected_length", [ ("file_type", 50), ("checksum", 64), ("algorithm", 20), ("related_entity_type", 50), ], ) def test_string_length(self, col_name, expected_length): col = _col(FileIntegrityRecord, col_name) assert col.type.length == expected_length class TestFileIntegrityRecordNullable: """Verify nullable constraints on each column.""" @pytest.mark.parametrize( "col_name, expected_nullable", [ ("id", False), ("file_path", False), ("file_type", False), ("checksum", False), ("algorithm", True), ("file_size", True), ("file_mtime", True), ("verify_on_load", True), ("allow_modifications", True), ("total_verifications", True), ("last_verified_at", True), ("last_verification_passed", True), ("consecutive_successes", True), ("consecutive_failures", True), ("created_at", True), ("updated_at", True), ("related_entity_type", True), ("related_entity_id", True), ], ) def test_nullable(self, col_name, expected_nullable): col = _col(FileIntegrityRecord, col_name) assert col.nullable is expected_nullable, ( f"Column '{col_name}' nullable={col.nullable}, " f"expected {expected_nullable}" ) class TestFileIntegrityRecordDefaults: """Verify default values set on columns.""" def test_algorithm_default(self): col = _col(FileIntegrityRecord, "algorithm") assert col.default is not None assert col.default.arg == "sha256" def test_verify_on_load_default(self): col = _col(FileIntegrityRecord, "verify_on_load") assert col.default is not None assert col.default.arg is True def test_allow_modifications_default(self): col = _col(FileIntegrityRecord, "allow_modifications") assert col.default is not None assert col.default.arg is False def test_total_verifications_default(self): col = _col(FileIntegrityRecord, "total_verifications") assert col.default is not None assert col.default.arg == 0 def test_last_verification_passed_default(self): col = _col(FileIntegrityRecord, "last_verification_passed") assert col.default is not None assert col.default.arg is True def test_consecutive_successes_default(self): col = _col(FileIntegrityRecord, "consecutive_successes") assert col.default is not None assert col.default.arg == 0 def test_consecutive_failures_default(self): col = _col(FileIntegrityRecord, "consecutive_failures") assert col.default is not None assert col.default.arg == 0 def test_created_at_has_default(self): col = _col(FileIntegrityRecord, "created_at") assert col.default is not None def test_updated_at_has_onupdate(self): col = _col(FileIntegrityRecord, "updated_at") assert col.onupdate is not None class TestFileIntegrityRecordPrimaryKey: """Verify primary key configuration.""" def test_primary_key_is_id(self): assert _pk_col_names(FileIntegrityRecord) == {"id"} def test_id_autoincrement(self): col = _col(FileIntegrityRecord, "id") # autoincrement is True or "auto" (depending on SQLAlchemy version) assert col.autoincrement in (True, "auto") class TestFileIntegrityRecordIndexes: """Verify indexes exist on expected columns.""" def test_file_path_index(self): assert _has_index(FileIntegrityRecord, "file_path") def test_file_type_index(self): assert _has_index(FileIntegrityRecord, "file_type") class TestFileIntegrityRecordUniqueConstraints: """Verify unique constraints on expected columns.""" def test_file_path_unique(self): assert _has_unique_constraint(FileIntegrityRecord, "file_path") def test_checksum_not_unique(self): col = _col(FileIntegrityRecord, "checksum") assert not col.unique class TestFileIntegrityRecordRepr: """Verify __repr__ output format.""" def test_repr_basic(self): record = FileIntegrityRecord( id=42, file_path="/data/index.faiss", file_type="faiss_index", total_verifications=7, ) expected = ( "" ) assert repr(record) == expected def test_repr_none_values(self): record = FileIntegrityRecord() expected = ( "" ) assert repr(record) == expected def test_repr_empty_strings(self): record = FileIntegrityRecord( id=0, file_path="", file_type="", total_verifications=0, ) expected = "" assert repr(record) == expected class TestFileIntegrityRecordRelationship: """Verify the verification_failures relationship is declared.""" def test_has_verification_failures_relationship(self): mapper = FileIntegrityRecord.__mapper__ assert "verification_failures" in mapper.relationships # ============================================================================ # FileVerificationFailure # ============================================================================ class TestFileVerificationFailureTableName: """Verify the __tablename__ value.""" def test_tablename(self): assert ( FileVerificationFailure.__tablename__ == "file_verification_failures" ) class TestFileVerificationFailureColumns: """Verify that all expected columns exist.""" EXPECTED_COLUMNS = { "id", "file_record_id", "verified_at", "expected_checksum", "actual_checksum", "file_size", "failure_reason", } def test_all_expected_columns_present(self): actual = _col_names(FileVerificationFailure) missing = self.EXPECTED_COLUMNS - actual assert not missing, f"Missing columns: {missing}" def test_no_unexpected_columns(self): actual = _col_names(FileVerificationFailure) extra = actual - self.EXPECTED_COLUMNS assert not extra, f"Unexpected columns: {extra}" class TestFileVerificationFailureColumnTypes: """Verify the SQLAlchemy column type for each column.""" @pytest.mark.parametrize( "col_name, expected_type", [ ("id", Integer), ("file_record_id", Integer), ("verified_at", UtcDateTime), ("expected_checksum", String), ("actual_checksum", String), ("file_size", Integer), ("failure_reason", Text), ], ) def test_column_type(self, col_name, expected_type): col = _col(FileVerificationFailure, col_name) assert isinstance(col.type, expected_type) class TestFileVerificationFailureStringLengths: """Verify length limits on String columns.""" @pytest.mark.parametrize( "col_name, expected_length", [ ("expected_checksum", 64), ("actual_checksum", 64), ], ) def test_string_length(self, col_name, expected_length): col = _col(FileVerificationFailure, col_name) assert col.type.length == expected_length class TestFileVerificationFailureNullable: """Verify nullable constraints on each column.""" @pytest.mark.parametrize( "col_name, expected_nullable", [ ("id", False), ("file_record_id", False), ("verified_at", True), ("expected_checksum", False), ("actual_checksum", True), ("file_size", True), ("failure_reason", False), ], ) def test_nullable(self, col_name, expected_nullable): col = _col(FileVerificationFailure, col_name) assert col.nullable is expected_nullable class TestFileVerificationFailureDefaults: """Verify default values set on columns.""" def test_verified_at_has_default(self): col = _col(FileVerificationFailure, "verified_at") assert col.default is not None class TestFileVerificationFailurePrimaryKey: """Verify primary key configuration.""" def test_primary_key_is_id(self): assert _pk_col_names(FileVerificationFailure) == {"id"} def test_id_autoincrement(self): col = _col(FileVerificationFailure, "id") assert col.autoincrement in (True, "auto") class TestFileVerificationFailureForeignKeys: """Verify foreign key references.""" def test_file_record_id_fk(self): targets = _fk_targets(FileVerificationFailure, "file_record_id") assert "file_integrity_records.id" in targets class TestFileVerificationFailureIndexes: """Verify indexes on expected columns.""" def test_file_record_id_index(self): assert _has_index(FileVerificationFailure, "file_record_id") class TestFileVerificationFailureRepr: """Verify __repr__ output format.""" def test_repr_basic(self): failure = FileVerificationFailure( id=5, file_record_id=42, failure_reason="checksum_mismatch", verified_at="2025-01-15T10:30:00", ) expected = ( "" ) assert repr(failure) == expected def test_repr_none_values(self): failure = FileVerificationFailure() expected = ( "" ) assert repr(failure) == expected class TestFileVerificationFailureRelationship: """Verify the file_record relationship is declared.""" def test_has_file_record_relationship(self): mapper = FileVerificationFailure.__mapper__ assert "file_record" in mapper.relationships # ============================================================================ # Report # ============================================================================ class TestReportTableName: """Verify the __tablename__ value.""" def test_tablename(self): assert Report.__tablename__ == "reports" class TestReportColumns: """Verify that all expected columns exist.""" EXPECTED_COLUMNS = { "id", "research_task_id", "title", "subtitle", "abstract", "content", "format", "template", "style", "language", "word_count", "section_count", "reference_count", "image_count", "generation_params", "generation_model", "generation_time_seconds", "version", "is_draft", "created_at", "updated_at", "published_at", } def test_all_expected_columns_present(self): actual = _col_names(Report) missing = self.EXPECTED_COLUMNS - actual assert not missing, f"Missing columns: {missing}" def test_no_unexpected_columns(self): actual = _col_names(Report) extra = actual - self.EXPECTED_COLUMNS assert not extra, f"Unexpected columns: {extra}" class TestReportColumnTypes: """Verify the SQLAlchemy column type for each column.""" @pytest.mark.parametrize( "col_name, expected_type", [ ("id", Integer), ("research_task_id", Integer), ("title", String), ("subtitle", String), ("abstract", Text), ("content", Text), ("format", String), ("template", String), ("style", String), ("language", String), ("word_count", Integer), ("section_count", Integer), ("reference_count", Integer), ("image_count", Integer), ("generation_params", SA_JSON), ("generation_model", String), ("generation_time_seconds", Float), ("version", Integer), ("is_draft", Boolean), ("created_at", UtcDateTime), ("updated_at", UtcDateTime), ("published_at", UtcDateTime), ], ) def test_column_type(self, col_name, expected_type): col = _col(Report, col_name) assert isinstance(col.type, expected_type), ( f"Column '{col_name}' type is {type(col.type).__name__}, " f"expected {expected_type.__name__}" ) class TestReportStringLengths: """Verify length limits on String columns.""" @pytest.mark.parametrize( "col_name, expected_length", [ ("title", 500), ("subtitle", 500), ("format", 50), ("template", 100), ("style", 100), ("language", 10), ("generation_model", 100), ], ) def test_string_length(self, col_name, expected_length): col = _col(Report, col_name) assert col.type.length == expected_length class TestReportNullable: """Verify nullable constraints on columns. Note: Columns without explicit nullable=False default to True in SQLAlchemy. """ @pytest.mark.parametrize( "col_name, expected_nullable", [ ("id", False), ("research_task_id", True), ("title", True), ("subtitle", True), ("abstract", True), ("content", True), ("format", True), ("template", True), ("style", True), ("language", True), ("word_count", True), ("section_count", True), ("reference_count", True), ("image_count", True), ("generation_params", True), ("generation_model", True), ("generation_time_seconds", True), ("version", True), ("is_draft", True), ("created_at", True), ("updated_at", True), ("published_at", True), ], ) def test_nullable(self, col_name, expected_nullable): col = _col(Report, col_name) assert col.nullable is expected_nullable class TestReportDefaults: """Verify default values set on columns.""" def test_format_default(self): col = _col(Report, "format") assert col.default is not None assert col.default.arg == "markdown" def test_language_default(self): col = _col(Report, "language") assert col.default is not None assert col.default.arg == "en" def test_version_default(self): col = _col(Report, "version") assert col.default is not None assert col.default.arg == 1 def test_is_draft_default(self): col = _col(Report, "is_draft") assert col.default is not None assert col.default.arg is False def test_created_at_has_default(self): col = _col(Report, "created_at") assert col.default is not None def test_updated_at_has_default(self): col = _col(Report, "updated_at") assert col.default is not None def test_updated_at_has_onupdate(self): col = _col(Report, "updated_at") assert col.onupdate is not None class TestReportPrimaryKey: """Verify primary key configuration.""" def test_primary_key_is_id(self): assert _pk_col_names(Report) == {"id"} class TestReportForeignKeys: """Verify foreign key references.""" def test_research_task_id_fk(self): targets = _fk_targets(Report, "research_task_id") assert "research_tasks.id" in targets def test_research_task_id_fk_ondelete_cascade(self): fks = _col(Report, "research_task_id").foreign_keys for fk in fks: assert fk.ondelete == "CASCADE" class TestReportRepr: """Verify __repr__ output format.""" def test_repr_basic(self): report = Report( title="Climate Change Analysis", format="markdown", is_draft=False, ) expected = "" assert repr(report) == expected def test_repr_draft(self): report = Report( title="Draft Report", format="html", is_draft=True, ) expected = "" assert repr(report) == expected def test_repr_none_values(self): report = Report() expected = "" assert repr(report) == expected def test_repr_title_with_special_characters(self): report = Report( title="Analysis of O'Brien's Data", format="pdf", is_draft=False, ) expected = "" assert repr(report) == expected class TestReportRelationships: """Verify relationships are declared on the mapper.""" def test_has_research_task_relationship(self): mapper = Report.__mapper__ assert "research_task" in mapper.relationships def test_has_sections_relationship(self): mapper = Report.__mapper__ assert "sections" in mapper.relationships # ============================================================================ # ReportSection # ============================================================================ class TestReportSectionTableName: """Verify the __tablename__ value.""" def test_tablename(self): assert ReportSection.__tablename__ == "report_sections" class TestReportSectionColumns: """Verify that all expected columns exist.""" EXPECTED_COLUMNS = { "id", "report_id", "title", "subtitle", "content", "section_order", "section_type", "section_level", "parent_section_id", "references", "citations", "auto_generated", "edited", "created_at", "updated_at", } def test_all_expected_columns_present(self): actual = _col_names(ReportSection) missing = self.EXPECTED_COLUMNS - actual assert not missing, f"Missing columns: {missing}" def test_no_unexpected_columns(self): actual = _col_names(ReportSection) extra = actual - self.EXPECTED_COLUMNS assert not extra, f"Unexpected columns: {extra}" class TestReportSectionColumnTypes: """Verify the SQLAlchemy column type for each column.""" @pytest.mark.parametrize( "col_name, expected_type", [ ("id", Integer), ("report_id", Integer), ("title", String), ("subtitle", String), ("content", Text), ("section_order", Integer), ("section_type", String), ("section_level", Integer), ("parent_section_id", Integer), ("references", SA_JSON), ("citations", SA_JSON), ("auto_generated", Boolean), ("edited", Boolean), ("created_at", UtcDateTime), ("updated_at", UtcDateTime), ], ) def test_column_type(self, col_name, expected_type): col = _col(ReportSection, col_name) assert isinstance(col.type, expected_type), ( f"Column '{col_name}' type is {type(col.type).__name__}, " f"expected {expected_type.__name__}" ) class TestReportSectionStringLengths: """Verify length limits on String columns.""" @pytest.mark.parametrize( "col_name, expected_length", [ ("title", 500), ("subtitle", 500), ("section_type", 50), ], ) def test_string_length(self, col_name, expected_length): col = _col(ReportSection, col_name) assert col.type.length == expected_length class TestReportSectionNullable: """Verify nullable constraints on columns.""" @pytest.mark.parametrize( "col_name, expected_nullable", [ ("id", False), ("report_id", True), ("title", True), ("subtitle", True), ("content", True), ("section_order", False), ("section_type", True), ("section_level", True), ("parent_section_id", True), ("references", True), ("citations", True), ("auto_generated", True), ("edited", True), ("created_at", True), ("updated_at", True), ], ) def test_nullable(self, col_name, expected_nullable): col = _col(ReportSection, col_name) assert col.nullable is expected_nullable class TestReportSectionDefaults: """Verify default values set on columns.""" def test_section_level_default(self): col = _col(ReportSection, "section_level") assert col.default is not None assert col.default.arg == 1 def test_auto_generated_default(self): col = _col(ReportSection, "auto_generated") assert col.default is not None assert col.default.arg is True def test_edited_default(self): col = _col(ReportSection, "edited") assert col.default is not None assert col.default.arg is False def test_created_at_has_default(self): col = _col(ReportSection, "created_at") assert col.default is not None def test_updated_at_has_default(self): col = _col(ReportSection, "updated_at") assert col.default is not None def test_updated_at_has_onupdate(self): col = _col(ReportSection, "updated_at") assert col.onupdate is not None class TestReportSectionPrimaryKey: """Verify primary key configuration.""" def test_primary_key_is_id(self): assert _pk_col_names(ReportSection) == {"id"} class TestReportSectionForeignKeys: """Verify foreign key references.""" def test_report_id_fk(self): targets = _fk_targets(ReportSection, "report_id") assert "reports.id" in targets def test_report_id_fk_ondelete_cascade(self): fks = _col(ReportSection, "report_id").foreign_keys for fk in fks: assert fk.ondelete == "CASCADE" def test_parent_section_id_fk(self): targets = _fk_targets(ReportSection, "parent_section_id") assert "report_sections.id" in targets class TestReportSectionRepr: """Verify __repr__ output format.""" def test_repr_basic(self): section = ReportSection( title="Introduction", section_order=1, section_type="introduction", ) expected = "" assert repr(section) == expected def test_repr_none_values(self): section = ReportSection() expected = "" assert repr(section) == expected def test_repr_high_order(self): section = ReportSection( title="Appendix B", section_order=99, section_type="appendix", ) expected = ( "" ) assert repr(section) == expected def test_repr_references_section(self): section = ReportSection( title="References", section_order=10, section_type="references", ) expected = ( "" ) assert repr(section) == expected class TestReportSectionRelationships: """Verify relationships are declared on the mapper.""" def test_has_report_relationship(self): mapper = ReportSection.__mapper__ assert "report" in mapper.relationships def test_has_subsections_relationship(self): mapper = ReportSection.__mapper__ assert "subsections" in mapper.relationships # ============================================================================ # Cross-model consistency checks # ============================================================================ class TestCrossModelConsistency: """Verify consistency across related models.""" def test_file_integrity_and_failure_fk_alignment(self): """The FK in FileVerificationFailure must point to FileIntegrityRecord's primary key table.""" fk_targets = _fk_targets(FileVerificationFailure, "file_record_id") assert f"{FileIntegrityRecord.__tablename__}.id" in fk_targets def test_report_section_fk_points_to_report(self): """The FK in ReportSection must point to Report's table.""" fk_targets = _fk_targets(ReportSection, "report_id") assert f"{Report.__tablename__}.id" in fk_targets def test_report_section_self_referential_fk(self): """The parent_section_id FK must reference ReportSection's own table.""" fk_targets = _fk_targets(ReportSection, "parent_section_id") assert f"{ReportSection.__tablename__}.id" in fk_targets def test_all_models_share_base(self): """All four models should inherit from the same Base.""" for model_cls in [ FileIntegrityRecord, FileVerificationFailure, Report, ReportSection, ]: from local_deep_research.database.models.base import Base assert issubclass(model_cls, Base) def test_all_tablenames_unique(self): """No two models should share the same __tablename__.""" names = [ FileIntegrityRecord.__tablename__, FileVerificationFailure.__tablename__, Report.__tablename__, ReportSection.__tablename__, ] assert len(names) == len(set(names))