7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
1026 lines
31 KiB
Python
1026 lines
31 KiB
Python
"""
|
|
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 = (
|
|
"<FileIntegrityRecord(id=42, "
|
|
"path=/data/index.faiss, "
|
|
"type=faiss_index, "
|
|
"verifications=7)>"
|
|
)
|
|
assert repr(record) == expected
|
|
|
|
def test_repr_none_values(self):
|
|
record = FileIntegrityRecord()
|
|
expected = (
|
|
"<FileIntegrityRecord(id=None, "
|
|
"path=None, "
|
|
"type=None, "
|
|
"verifications=None)>"
|
|
)
|
|
assert repr(record) == expected
|
|
|
|
def test_repr_empty_strings(self):
|
|
record = FileIntegrityRecord(
|
|
id=0,
|
|
file_path="",
|
|
file_type="",
|
|
total_verifications=0,
|
|
)
|
|
expected = "<FileIntegrityRecord(id=0, path=, type=, verifications=0)>"
|
|
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 = (
|
|
"<FileVerificationFailure(id=5, "
|
|
"file_record_id=42, "
|
|
"reason=checksum_mismatch, "
|
|
"verified_at=2025-01-15T10:30:00)>"
|
|
)
|
|
assert repr(failure) == expected
|
|
|
|
def test_repr_none_values(self):
|
|
failure = FileVerificationFailure()
|
|
expected = (
|
|
"<FileVerificationFailure(id=None, "
|
|
"file_record_id=None, "
|
|
"reason=None, "
|
|
"verified_at=None)>"
|
|
)
|
|
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 = "<Report(title='Climate Change Analysis', format='markdown', draft=False)>"
|
|
assert repr(report) == expected
|
|
|
|
def test_repr_draft(self):
|
|
report = Report(
|
|
title="Draft Report",
|
|
format="html",
|
|
is_draft=True,
|
|
)
|
|
expected = "<Report(title='Draft Report', format='html', draft=True)>"
|
|
assert repr(report) == expected
|
|
|
|
def test_repr_none_values(self):
|
|
report = Report()
|
|
expected = "<Report(title='None', format='None', draft=None)>"
|
|
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 = "<Report(title='Analysis of O'Brien's Data', format='pdf', draft=False)>"
|
|
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 = "<ReportSection(title='Introduction', order=1, type='introduction')>"
|
|
assert repr(section) == expected
|
|
|
|
def test_repr_none_values(self):
|
|
section = ReportSection()
|
|
expected = "<ReportSection(title='None', order=None, type='None')>"
|
|
assert repr(section) == expected
|
|
|
|
def test_repr_high_order(self):
|
|
section = ReportSection(
|
|
title="Appendix B",
|
|
section_order=99,
|
|
section_type="appendix",
|
|
)
|
|
expected = (
|
|
"<ReportSection(title='Appendix B', order=99, type='appendix')>"
|
|
)
|
|
assert repr(section) == expected
|
|
|
|
def test_repr_references_section(self):
|
|
section = ReportSection(
|
|
title="References",
|
|
section_order=10,
|
|
section_type="references",
|
|
)
|
|
expected = (
|
|
"<ReportSection(title='References', order=10, type='references')>"
|
|
)
|
|
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))
|