7a0da7932b
Backwards Compatibility / Verify Encryption Constants (push) Waiting to run
Backwards Compatibility / PyPI Version Compatibility (push) Waiting to run
Backwards Compatibility / Database Migration Tests (push) Waiting to run
CodeQL Advanced / Analyze (javascript-typescript) (push) Waiting to run
CodeQL Advanced / Analyze (python) (push) Waiting to run
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Blocked by required conditions
Docker Tests (Consolidated) / detect-changes (push) Waiting to run
Docker Tests (Consolidated) / Build Test Image (push) Waiting to run
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Blocked by required conditions
Docker Tests (Consolidated) / Accessibility Tests (push) Blocked by required conditions
Docker Tests (Consolidated) / LLM Unit Tests (push) Blocked by required conditions
Docker Tests (Consolidated) / LLM Example Tests (push) Blocked by required conditions
Docker Tests (Consolidated) / Production Image Smoke Test (push) Blocked by required conditions
Docker Tests (Consolidated) / Infrastructure Tests (push) Blocked by required conditions
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Waiting to run
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
439 lines
14 KiB
Python
439 lines
14 KiB
Python
"""
|
|
Tests for download tracking models (DownloadTracker, DownloadDuplicates, DownloadAttempt).
|
|
|
|
Tests model structure, column constraints, default values, and __repr__ methods
|
|
WITHOUT requiring a real database -- uses direct attribute assignment on model instances.
|
|
"""
|
|
|
|
from sqlalchemy import Boolean, Integer, String, Text, UniqueConstraint
|
|
|
|
from local_deep_research.database.models.download_tracker import (
|
|
DownloadAttempt,
|
|
DownloadDuplicates,
|
|
DownloadTracker,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _col(model_cls, name):
|
|
"""Return a Column object from a model's __table__."""
|
|
return model_cls.__table__.columns[name]
|
|
|
|
|
|
def _make(model_cls, **kwargs):
|
|
"""Instantiate a model via its normal constructor (no DB session needed).
|
|
|
|
SQLAlchemy's declarative ``__init__`` accepts column names as keyword
|
|
arguments and properly initialises the instance state so that
|
|
instrumented attribute access (used by ``__repr__``) works correctly.
|
|
"""
|
|
return model_cls(**kwargs)
|
|
|
|
|
|
# ===========================================================================
|
|
# DownloadTracker
|
|
# ===========================================================================
|
|
|
|
|
|
class TestDownloadTrackerTablename:
|
|
"""Test 1: __tablename__ is 'download_tracker'."""
|
|
|
|
def test_tablename(self):
|
|
assert DownloadTracker.__tablename__ == "download_tracker"
|
|
|
|
|
|
class TestDownloadTrackerColumns:
|
|
"""Test 2: Column types and nullable constraints."""
|
|
|
|
def test_url_column_is_text_and_not_nullable(self):
|
|
col = _col(DownloadTracker, "url")
|
|
assert isinstance(col.type, Text)
|
|
assert col.nullable is False
|
|
|
|
def test_url_hash_is_string64_not_nullable_with_table_level_unique(self):
|
|
col = _col(DownloadTracker, "url_hash")
|
|
assert isinstance(col.type, String)
|
|
assert col.type.length == 64
|
|
assert col.nullable is False
|
|
# Uniqueness lives on a table-level UniqueConstraint named
|
|
# ``uq_download_tracker_url_hash`` (see model __table_args__) so the
|
|
# index lands inline in CREATE TABLE — required so SQLCipher accepts
|
|
# ``url_hash`` as a valid FK target for the child tables. Column-level
|
|
# ``unique=True``/``index=True`` would emit a separate CREATE UNIQUE
|
|
# INDEX which SQLCipher does not recognise as the FK target.
|
|
unique_cols = {
|
|
tuple(c.name for c in constraint.columns)
|
|
for constraint in DownloadTracker.__table__.constraints
|
|
if isinstance(constraint, UniqueConstraint)
|
|
}
|
|
assert ("url_hash",) in unique_cols
|
|
|
|
def test_first_resource_id_is_integer_not_nullable(self):
|
|
col = _col(DownloadTracker, "first_resource_id")
|
|
assert isinstance(col.type, Integer)
|
|
assert col.nullable is False
|
|
|
|
def test_first_resource_id_has_foreign_key(self):
|
|
col = _col(DownloadTracker, "first_resource_id")
|
|
fk_targets = [fk.target_fullname for fk in col.foreign_keys]
|
|
assert "research_resources.id" in fk_targets
|
|
|
|
def test_file_hash_is_nullable(self):
|
|
col = _col(DownloadTracker, "file_hash")
|
|
assert isinstance(col.type, String)
|
|
assert col.type.length == 64
|
|
assert col.nullable is True
|
|
assert col.index is True
|
|
|
|
def test_file_path_is_nullable_text(self):
|
|
col = _col(DownloadTracker, "file_path")
|
|
assert isinstance(col.type, Text)
|
|
assert col.nullable is True
|
|
|
|
def test_file_name_is_nullable_string255_indexed(self):
|
|
col = _col(DownloadTracker, "file_name")
|
|
assert isinstance(col.type, String)
|
|
assert col.type.length == 255
|
|
assert col.nullable is True
|
|
assert col.index is True
|
|
|
|
def test_file_size_is_nullable_integer(self):
|
|
col = _col(DownloadTracker, "file_size")
|
|
assert isinstance(col.type, Integer)
|
|
assert col.nullable is True
|
|
|
|
def test_is_downloaded_is_boolean_not_nullable_indexed(self):
|
|
col = _col(DownloadTracker, "is_downloaded")
|
|
assert isinstance(col.type, Boolean)
|
|
assert col.nullable is False
|
|
assert col.index is True
|
|
|
|
def test_is_accessible_is_boolean(self):
|
|
col = _col(DownloadTracker, "is_accessible")
|
|
assert isinstance(col.type, Boolean)
|
|
|
|
def test_downloaded_at_is_nullable(self):
|
|
col = _col(DownloadTracker, "downloaded_at")
|
|
assert col.nullable is True
|
|
|
|
def test_first_seen_is_not_nullable(self):
|
|
col = _col(DownloadTracker, "first_seen")
|
|
assert col.nullable is False
|
|
|
|
def test_last_checked_is_not_nullable(self):
|
|
col = _col(DownloadTracker, "last_checked")
|
|
assert col.nullable is False
|
|
|
|
def test_library_document_id_is_nullable_with_fk(self):
|
|
col = _col(DownloadTracker, "library_document_id")
|
|
# documents.id is String(36) UUID; column type matches PK type so
|
|
# SQLite doesn't silently coerce via type-affinity (which produced
|
|
# the original mismatched-FK bug, see #3697).
|
|
assert isinstance(col.type, String)
|
|
assert col.type.length == 36
|
|
assert col.nullable is True
|
|
fk_targets = [fk.target_fullname for fk in col.foreign_keys]
|
|
assert "documents.id" in fk_targets
|
|
|
|
def test_id_is_primary_key_autoincrement(self):
|
|
col = _col(DownloadTracker, "id")
|
|
assert col.primary_key is True
|
|
assert col.autoincrement is True
|
|
|
|
|
|
class TestDownloadTrackerReprDownloaded:
|
|
"""Test 3: __repr__ shows 'downloaded' when is_downloaded=True."""
|
|
|
|
def test_repr_downloaded(self):
|
|
obj = _make(
|
|
DownloadTracker,
|
|
url_hash="abcdef1234567890",
|
|
is_downloaded=True,
|
|
)
|
|
result = repr(obj)
|
|
assert "downloaded" in result
|
|
assert "not downloaded" not in result
|
|
|
|
|
|
class TestDownloadTrackerReprNotDownloaded:
|
|
"""Test 4: __repr__ shows 'not downloaded' when is_downloaded=False."""
|
|
|
|
def test_repr_not_downloaded(self):
|
|
obj = _make(
|
|
DownloadTracker,
|
|
url_hash="abcdef1234567890",
|
|
is_downloaded=False,
|
|
)
|
|
result = repr(obj)
|
|
assert "not downloaded" in result
|
|
|
|
|
|
class TestDownloadTrackerReprTruncation:
|
|
"""Test 5: __repr__ truncates url_hash to first 8 characters."""
|
|
|
|
def test_repr_truncates_url_hash(self):
|
|
full_hash = "a1b2c3d4e5f6g7h8i9j0"
|
|
obj = _make(
|
|
DownloadTracker,
|
|
url_hash=full_hash,
|
|
is_downloaded=False,
|
|
)
|
|
result = repr(obj)
|
|
assert "a1b2c3d4..." in result
|
|
# The full hash should NOT appear in the repr
|
|
assert full_hash not in result
|
|
|
|
|
|
class TestDownloadTrackerDefaultIsDownloaded:
|
|
"""Test 6: Default is_downloaded is False."""
|
|
|
|
def test_default_is_downloaded(self):
|
|
col = _col(DownloadTracker, "is_downloaded")
|
|
assert col.default is not None
|
|
assert col.default.arg is False
|
|
|
|
|
|
class TestDownloadTrackerDefaultIsAccessible:
|
|
"""Test 7: Default is_accessible is True."""
|
|
|
|
def test_default_is_accessible(self):
|
|
col = _col(DownloadTracker, "is_accessible")
|
|
assert col.default is not None
|
|
assert col.default.arg is True
|
|
|
|
|
|
# ===========================================================================
|
|
# DownloadDuplicates
|
|
# ===========================================================================
|
|
|
|
|
|
class TestDownloadDuplicatesTablename:
|
|
"""Test 8: __tablename__ is 'download_duplicates'."""
|
|
|
|
def test_tablename(self):
|
|
assert DownloadDuplicates.__tablename__ == "download_duplicates"
|
|
|
|
|
|
class TestDownloadDuplicatesUniqueConstraint:
|
|
"""Test 9: Has UniqueConstraint on (url_hash, resource_id)."""
|
|
|
|
def test_unique_constraint_exists(self):
|
|
table = DownloadDuplicates.__table__
|
|
unique_constraints = [
|
|
c
|
|
for c in table.constraints
|
|
if c.__class__.__name__ == "UniqueConstraint"
|
|
]
|
|
# Find the constraint named 'uix_url_resource'
|
|
matching = [
|
|
c for c in unique_constraints if c.name == "uix_url_resource"
|
|
]
|
|
assert len(matching) == 1
|
|
col_names = {col.name for col in matching[0].columns}
|
|
assert col_names == {"url_hash", "resource_id"}
|
|
|
|
|
|
class TestDownloadDuplicatesRepr:
|
|
"""Test 10: __repr__ shows url_hash and resource_id."""
|
|
|
|
def test_repr_content(self):
|
|
obj = _make(
|
|
DownloadDuplicates,
|
|
url_hash="deadbeef12345678",
|
|
resource_id=42,
|
|
)
|
|
result = repr(obj)
|
|
assert "deadbeef..." in result
|
|
assert "resource_id=42" in result
|
|
assert "DownloadDuplicates" in result
|
|
|
|
|
|
class TestDownloadDuplicatesCompositeIndex:
|
|
"""Test 11: Has composite index on (research_id, url_hash)."""
|
|
|
|
def test_composite_index_exists(self):
|
|
table = DownloadDuplicates.__table__
|
|
index_names = {idx.name for idx in table.indexes}
|
|
assert "idx_research_duplicates" in index_names
|
|
|
|
# Verify the columns in the index
|
|
target_index = next(
|
|
idx
|
|
for idx in table.indexes
|
|
if idx.name == "idx_research_duplicates"
|
|
)
|
|
col_names = [col.name for col in target_index.columns]
|
|
assert "research_id" in col_names
|
|
assert "url_hash" in col_names
|
|
|
|
|
|
class TestDownloadDuplicatesColumns:
|
|
"""Additional column validation for DownloadDuplicates."""
|
|
|
|
def test_url_hash_not_nullable_with_fk(self):
|
|
col = _col(DownloadDuplicates, "url_hash")
|
|
assert col.nullable is False
|
|
fk_targets = [fk.target_fullname for fk in col.foreign_keys]
|
|
assert "download_tracker.url_hash" in fk_targets
|
|
|
|
def test_resource_id_not_nullable_with_fk(self):
|
|
col = _col(DownloadDuplicates, "resource_id")
|
|
assert col.nullable is False
|
|
fk_targets = [fk.target_fullname for fk in col.foreign_keys]
|
|
assert "research_resources.id" in fk_targets
|
|
|
|
def test_research_id_not_nullable_indexed(self):
|
|
col = _col(DownloadDuplicates, "research_id")
|
|
assert isinstance(col.type, String)
|
|
assert col.type.length == 36
|
|
assert col.nullable is False
|
|
assert col.index is True
|
|
|
|
def test_added_at_not_nullable(self):
|
|
col = _col(DownloadDuplicates, "added_at")
|
|
assert col.nullable is False
|
|
|
|
|
|
# ===========================================================================
|
|
# DownloadAttempt
|
|
# ===========================================================================
|
|
|
|
|
|
class TestDownloadAttemptTablename:
|
|
"""Test 12: __tablename__ is 'download_attempts'."""
|
|
|
|
def test_tablename(self):
|
|
assert DownloadAttempt.__tablename__ == "download_attempts"
|
|
|
|
|
|
class TestDownloadAttemptReprSuccess:
|
|
"""Test 13: __repr__ shows 'success' when succeeded=True."""
|
|
|
|
def test_repr_success(self):
|
|
obj = _make(
|
|
DownloadAttempt,
|
|
attempt_number=1,
|
|
succeeded=True,
|
|
status_code=200,
|
|
error_type=None,
|
|
)
|
|
result = repr(obj)
|
|
assert "success" in result
|
|
assert "failed" not in result
|
|
assert "attempt=1" in result
|
|
|
|
|
|
class TestDownloadAttemptReprFailedWithStatusCode:
|
|
"""Test 14: __repr__ shows 'failed (404)' when succeeded=False, status_code=404."""
|
|
|
|
def test_repr_failed_status_code(self):
|
|
obj = _make(
|
|
DownloadAttempt,
|
|
attempt_number=2,
|
|
succeeded=False,
|
|
status_code=404,
|
|
error_type=None,
|
|
)
|
|
result = repr(obj)
|
|
assert "failed (404)" in result
|
|
assert "attempt=2" in result
|
|
|
|
|
|
class TestDownloadAttemptReprFailedWithErrorType:
|
|
"""Test 15: __repr__ shows 'failed (timeout)' when no status_code, error_type='timeout'."""
|
|
|
|
def test_repr_failed_error_type(self):
|
|
obj = _make(
|
|
DownloadAttempt,
|
|
attempt_number=3,
|
|
succeeded=False,
|
|
status_code=None,
|
|
error_type="timeout",
|
|
)
|
|
result = repr(obj)
|
|
assert "failed (timeout)" in result
|
|
assert "attempt=3" in result
|
|
|
|
|
|
class TestDownloadAttemptReprFailedNoInfo:
|
|
"""Test 16: __repr__ shows 'failed (None)' when no status_code or error_type."""
|
|
|
|
def test_repr_failed_none(self):
|
|
obj = _make(
|
|
DownloadAttempt,
|
|
attempt_number=1,
|
|
succeeded=False,
|
|
status_code=None,
|
|
error_type=None,
|
|
)
|
|
result = repr(obj)
|
|
assert "failed (None)" in result
|
|
|
|
|
|
class TestDownloadAttemptDefaultSucceeded:
|
|
"""Test 17: Default succeeded is False."""
|
|
|
|
def test_default_succeeded(self):
|
|
col = _col(DownloadAttempt, "succeeded")
|
|
assert col.default is not None
|
|
assert col.default.arg is False
|
|
|
|
|
|
class TestDownloadAttemptColumns:
|
|
"""Additional column validation for DownloadAttempt."""
|
|
|
|
def test_url_hash_not_nullable_with_fk(self):
|
|
col = _col(DownloadAttempt, "url_hash")
|
|
assert col.nullable is False
|
|
assert col.index is True
|
|
fk_targets = [fk.target_fullname for fk in col.foreign_keys]
|
|
assert "download_tracker.url_hash" in fk_targets
|
|
|
|
def test_attempt_number_not_nullable(self):
|
|
col = _col(DownloadAttempt, "attempt_number")
|
|
assert isinstance(col.type, Integer)
|
|
assert col.nullable is False
|
|
|
|
def test_status_code_nullable(self):
|
|
col = _col(DownloadAttempt, "status_code")
|
|
assert isinstance(col.type, Integer)
|
|
assert col.nullable is True
|
|
|
|
def test_error_type_nullable_string100(self):
|
|
col = _col(DownloadAttempt, "error_type")
|
|
assert isinstance(col.type, String)
|
|
assert col.type.length == 100
|
|
assert col.nullable is True
|
|
|
|
def test_error_message_nullable_text(self):
|
|
col = _col(DownloadAttempt, "error_message")
|
|
assert isinstance(col.type, Text)
|
|
assert col.nullable is True
|
|
|
|
def test_attempted_at_not_nullable(self):
|
|
col = _col(DownloadAttempt, "attempted_at")
|
|
assert col.nullable is False
|
|
|
|
def test_duration_ms_nullable(self):
|
|
col = _col(DownloadAttempt, "duration_ms")
|
|
assert isinstance(col.type, Integer)
|
|
assert col.nullable is True
|
|
|
|
def test_succeeded_not_nullable(self):
|
|
col = _col(DownloadAttempt, "succeeded")
|
|
assert isinstance(col.type, Boolean)
|
|
assert col.nullable is False
|
|
|
|
def test_bytes_downloaded_nullable(self):
|
|
col = _col(DownloadAttempt, "bytes_downloaded")
|
|
assert isinstance(col.type, Integer)
|
|
assert col.nullable is True
|
|
|
|
def test_id_primary_key(self):
|
|
col = _col(DownloadAttempt, "id")
|
|
assert col.primary_key is True
|