wehub-resource-sync
751 lines
28 KiB
Python
751 lines
28 KiB
Python
"""
|
|
Unit tests for the open_notebook.domain module.
|
|
|
|
This test suite focuses on validation logic, business rules, and data structures
|
|
that can be tested without database mocking.
|
|
"""
|
|
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from types import ModuleType, SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from api.podcast_service import PodcastService
|
|
from open_notebook.ai.models import ModelManager
|
|
from open_notebook.domain.base import RecordModel
|
|
from open_notebook.domain.content_settings import ContentSettings
|
|
from open_notebook.domain.notebook import (
|
|
Asset,
|
|
Note,
|
|
Notebook,
|
|
Source,
|
|
SourceInsight,
|
|
)
|
|
from open_notebook.domain.transformation import Transformation
|
|
from open_notebook.exceptions import InvalidInputError
|
|
from open_notebook.podcasts.models import EpisodeProfile, SpeakerProfile
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 1: RecordModel Singleton Pattern
|
|
# ============================================================================
|
|
|
|
|
|
class TestRecordModelSingleton:
|
|
"""Test suite for RecordModel singleton behavior."""
|
|
|
|
def test_recordmodel_singleton_behavior(self):
|
|
"""Test that same instance is returned for same record_id."""
|
|
|
|
class TestRecord(RecordModel):
|
|
record_id = "test:singleton"
|
|
value: int = 0
|
|
|
|
# Clear any existing instance
|
|
TestRecord.clear_instance()
|
|
|
|
# Create first instance
|
|
instance1 = TestRecord(value=42)
|
|
assert instance1.value == 42
|
|
|
|
# Create second instance - should return same object
|
|
instance2 = TestRecord(value=99)
|
|
assert instance1 is instance2
|
|
assert instance2.value == 99 # Value was updated
|
|
|
|
# Cleanup
|
|
TestRecord.clear_instance()
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 2: ModelManager Instance Isolation
|
|
# ============================================================================
|
|
|
|
|
|
class TestModelManager:
|
|
"""Test suite for ModelManager instance behavior."""
|
|
|
|
def test_model_manager_instance_isolation(self):
|
|
"""Test that each ModelManager instance is independent (not a singleton)."""
|
|
manager1 = ModelManager()
|
|
manager2 = ModelManager()
|
|
|
|
# Each instance should be independent (not a singleton)
|
|
assert manager1 is not manager2
|
|
assert id(manager1) != id(manager2)
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 3: Notebook Domain Logic
|
|
# ============================================================================
|
|
|
|
|
|
class TestNotebookDomain:
|
|
"""Test suite for Notebook validation and business rules."""
|
|
|
|
def test_notebook_name_validation(self):
|
|
"""Test empty/whitespace names are rejected."""
|
|
# Empty name should raise error
|
|
with pytest.raises(InvalidInputError, match="Notebook name cannot be empty"):
|
|
Notebook(name="", description="Test")
|
|
|
|
# Whitespace-only name should raise error
|
|
with pytest.raises(InvalidInputError, match="Notebook name cannot be empty"):
|
|
Notebook(name=" ", description="Test")
|
|
|
|
# Valid name should work
|
|
notebook = Notebook(name="Valid Name", description="Test")
|
|
assert notebook.name == "Valid Name"
|
|
|
|
def test_notebook_archived_flag(self):
|
|
"""Test archived flag defaults to False."""
|
|
notebook = Notebook(name="Test", description="Test")
|
|
assert notebook.archived is False
|
|
|
|
notebook_archived = Notebook(name="Test", description="Test", archived=True)
|
|
assert notebook_archived.archived is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notebook_get_context_includes_source_full_text(self):
|
|
"""Test notebook context includes full source content for podcasts."""
|
|
notebook = Notebook(id="notebook:test", name="Test", description="Test")
|
|
sources = [
|
|
Source(
|
|
id="source:first",
|
|
title="First Source",
|
|
full_text="First source full text for podcast generation.",
|
|
),
|
|
Source(
|
|
id="source:second",
|
|
title="Second Source",
|
|
full_text="Second source full text for podcast generation.",
|
|
),
|
|
]
|
|
get_sources_calls = []
|
|
|
|
async def fake_get_sources(self, include_full_text=False):
|
|
get_sources_calls.append(include_full_text)
|
|
return sources
|
|
|
|
async def fake_get_notes(self, include_content=False):
|
|
return []
|
|
|
|
async def fake_get_for_sources(cls, source_ids):
|
|
return {sid: [] for sid in source_ids}
|
|
|
|
with (
|
|
patch.object(Notebook, "get_sources", new=fake_get_sources),
|
|
patch.object(Notebook, "get_notes", new=fake_get_notes),
|
|
patch.object(
|
|
SourceInsight, "get_for_sources", new=classmethod(fake_get_for_sources)
|
|
),
|
|
):
|
|
context = await notebook.get_context()
|
|
|
|
assert get_sources_calls == [True]
|
|
assert "## Source: First Source" in context
|
|
assert "First source full text for podcast generation." in context
|
|
assert "## Source: Second Source" in context
|
|
assert "Second source full text for podcast generation." in context
|
|
assert "Notebook(id=" not in context
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notebook_get_context_includes_note_content(self):
|
|
"""Test notebook context includes linked note content."""
|
|
notebook = Notebook(id="notebook:test", name="Test", description="Test")
|
|
notes = [
|
|
Note(
|
|
id="note:first",
|
|
title="Research Note",
|
|
content="Important notebook note for the podcast.",
|
|
)
|
|
]
|
|
get_notes_calls = []
|
|
|
|
async def fake_get_sources(self, include_full_text=False):
|
|
return []
|
|
|
|
async def fake_get_notes(self, include_content=False):
|
|
get_notes_calls.append(include_content)
|
|
return notes
|
|
|
|
with (
|
|
patch.object(Notebook, "get_sources", new=fake_get_sources),
|
|
patch.object(Notebook, "get_notes", new=fake_get_notes),
|
|
):
|
|
context = await notebook.get_context()
|
|
|
|
assert get_notes_calls == [True]
|
|
assert "## Note: Research Note" in context
|
|
assert "Important notebook note for the podcast." in context
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notebook_get_context_returns_empty_string_without_content(self):
|
|
"""Test notebooks with no source or note content produce empty context."""
|
|
notebook = Notebook(id="notebook:test", name="Test", description="Test")
|
|
|
|
async def fake_get_sources(self, include_full_text=False):
|
|
return []
|
|
|
|
async def fake_get_notes(self, include_content=False):
|
|
return []
|
|
|
|
with (
|
|
patch.object(Notebook, "get_sources", new=fake_get_sources),
|
|
patch.object(Notebook, "get_notes", new=fake_get_notes),
|
|
):
|
|
assert await notebook.get_context() == ""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notebook_get_context_propagates_source_errors(self):
|
|
"""Test source context failures are not swallowed by notebook context."""
|
|
notebook = Notebook(id="notebook:test", name="Test", description="Test")
|
|
source = Source(id="source:first", title="First Source")
|
|
|
|
async def fake_get_sources(self, include_full_text=False):
|
|
return [source]
|
|
|
|
async def fake_get_notes(self, include_content=False):
|
|
return []
|
|
|
|
async def fake_get_for_sources(cls, source_ids):
|
|
return {sid: [] for sid in source_ids}
|
|
|
|
async def fake_get_context(self, context_size="short", insights=None):
|
|
raise RuntimeError("source context failed")
|
|
|
|
with (
|
|
patch.object(Notebook, "get_sources", new=fake_get_sources),
|
|
patch.object(Notebook, "get_notes", new=fake_get_notes),
|
|
patch.object(
|
|
SourceInsight, "get_for_sources", new=classmethod(fake_get_for_sources)
|
|
),
|
|
patch.object(Source, "get_context", new=fake_get_context),
|
|
):
|
|
with pytest.raises(RuntimeError, match="source context failed"):
|
|
await notebook.get_context()
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 4: Source Domain
|
|
# ============================================================================
|
|
|
|
|
|
class TestSourceDomain:
|
|
"""Test suite for Source domain model."""
|
|
|
|
def test_source_command_field_parsing(self):
|
|
"""Test RecordID parsing for command field."""
|
|
# Test with string command
|
|
source = Source(title="Test", command="command:123")
|
|
assert source.command is not None
|
|
|
|
# Test with None command
|
|
source2 = Source(title="Test", command=None)
|
|
assert source2.command is None
|
|
|
|
# Test command is included in save data prep
|
|
source3 = Source(id="source:123", title="Test", command="command:456")
|
|
save_data = source3._prepare_save_data()
|
|
assert "command" in save_data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_delete_cleans_up_file(self):
|
|
"""Test that deleting a source removes the associated file."""
|
|
# Create a temporary file
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as tmp_file:
|
|
tmp_file.write(b"Test content")
|
|
tmp_path = Path(tmp_file.name)
|
|
|
|
try:
|
|
# Create source with file asset
|
|
source = Source(
|
|
id="source:test_delete",
|
|
title="Test Source",
|
|
asset=Asset(file_path=str(tmp_path)),
|
|
)
|
|
|
|
# Verify file exists
|
|
assert tmp_path.exists()
|
|
|
|
# Mock the parent delete method to avoid database operations
|
|
with patch.object(
|
|
Source.__bases__[0], "delete", new_callable=AsyncMock
|
|
) as mock_delete:
|
|
mock_delete.return_value = True
|
|
|
|
# Delete the source
|
|
result = await source.delete()
|
|
|
|
# Verify parent delete was called
|
|
mock_delete.assert_called_once()
|
|
assert result is True
|
|
|
|
# Verify file was deleted
|
|
assert not tmp_path.exists()
|
|
|
|
finally:
|
|
# Cleanup in case test fails
|
|
if tmp_path.exists():
|
|
tmp_path.unlink()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_delete_without_file(self):
|
|
"""Test that deleting a source without a file doesn't fail."""
|
|
# Create source without file asset
|
|
source = Source(id="source:test_no_file", title="Test Source", asset=None)
|
|
|
|
# Mock the parent delete method
|
|
with patch.object(
|
|
Source.__bases__[0], "delete", new_callable=AsyncMock
|
|
) as mock_delete:
|
|
mock_delete.return_value = True
|
|
|
|
# Delete should complete without error
|
|
result = await source.delete()
|
|
assert result is True
|
|
mock_delete.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_delete_continues_on_file_error(self):
|
|
"""Test that source deletion continues even if file deletion fails."""
|
|
# Create source with non-existent file
|
|
source = Source(
|
|
id="source:test_missing_file",
|
|
title="Test Source",
|
|
asset=Asset(file_path="/nonexistent/path/file.txt"),
|
|
)
|
|
|
|
# Mock the parent delete method
|
|
with patch.object(
|
|
Source.__bases__[0], "delete", new_callable=AsyncMock
|
|
) as mock_delete:
|
|
mock_delete.return_value = True
|
|
|
|
# Delete should complete even though file doesn't exist
|
|
result = await source.delete()
|
|
assert result is True
|
|
mock_delete.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vectorize_raises_valueerror_when_no_text(self):
|
|
"""Test that vectorize() raises ValueError (not DatabaseOperationError) for empty text."""
|
|
source = Source(id="source:test_empty", title="Test", full_text=None)
|
|
with pytest.raises(ValueError, match="has no text to vectorize"):
|
|
await source.vectorize()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vectorize_raises_valueerror_when_empty_string(self):
|
|
"""Test that vectorize() raises ValueError for empty string."""
|
|
source = Source(id="source:test_empty_str", title="Test", full_text="")
|
|
with pytest.raises(ValueError, match="has no text to vectorize"):
|
|
await source.vectorize()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vectorize_raises_valueerror_when_whitespace_only(self):
|
|
"""Test that vectorize() raises ValueError for whitespace-only text."""
|
|
source = Source(id="source:test_ws", title="Test", full_text=" \n\t ")
|
|
with pytest.raises(ValueError, match="has no text to vectorize"):
|
|
await source.vectorize()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_vectorize_submits_command_with_valid_text(self):
|
|
"""Test that vectorize() submits embed_source command when text is valid."""
|
|
source = Source(id="source:test_valid", title="Test", full_text="Real content")
|
|
with patch(
|
|
"open_notebook.domain.notebook.submit_command", return_value="command:123"
|
|
) as mock_submit:
|
|
result = await source.vectorize()
|
|
mock_submit.assert_called_once_with(
|
|
"open_notebook",
|
|
"embed_source",
|
|
{"source_id": "source:test_valid"},
|
|
)
|
|
assert result == "command:123"
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 5: Note Domain
|
|
# ============================================================================
|
|
|
|
|
|
class TestNoteDomain:
|
|
"""Test suite for Note validation."""
|
|
|
|
def test_note_content_validation(self):
|
|
"""Test empty content is rejected."""
|
|
# None content is allowed
|
|
note = Note(title="Test", content=None)
|
|
assert note.content is None
|
|
|
|
# Non-empty content is valid
|
|
note2 = Note(title="Test", content="Valid content")
|
|
assert note2.content == "Valid content"
|
|
|
|
# Empty string should raise error
|
|
with pytest.raises(InvalidInputError, match="Note content cannot be empty"):
|
|
Note(title="Test", content="")
|
|
|
|
# Whitespace-only should raise error
|
|
with pytest.raises(InvalidInputError, match="Note content cannot be empty"):
|
|
Note(title="Test", content=" ")
|
|
|
|
def test_note_content_for_embedding(self):
|
|
"""Test notes can hold content for embedding.
|
|
|
|
Note: Embedding is now handled via command submission in Note.save(),
|
|
not via needs_embedding() method. This test verifies basic content handling.
|
|
"""
|
|
note = Note(title="Test", content="Test content")
|
|
assert note.content == "Test content"
|
|
|
|
# Test with None content - valid, no embedding will be submitted
|
|
note2 = Note(title="Test", content=None)
|
|
assert note2.content is None
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 6: Podcast Domain Validation
|
|
# ============================================================================
|
|
|
|
|
|
class TestPodcastDomain:
|
|
"""Test suite for Podcast domain validation."""
|
|
|
|
def test_speaker_profile_validation(self):
|
|
"""Test speaker profile validates count and required fields."""
|
|
# Test invalid - no speakers
|
|
with pytest.raises(ValidationError):
|
|
SpeakerProfile(
|
|
name="Test",
|
|
speakers=[],
|
|
)
|
|
|
|
# Test invalid - too many speakers (> 4)
|
|
with pytest.raises(ValidationError):
|
|
SpeakerProfile(
|
|
name="Test",
|
|
speakers=[{"name": f"Speaker{i}"} for i in range(5)],
|
|
)
|
|
|
|
# Test invalid - missing required fields
|
|
with pytest.raises(ValidationError):
|
|
SpeakerProfile(
|
|
name="Test",
|
|
speakers=[
|
|
{"name": "Speaker 1"}
|
|
], # Missing voice_id, backstory, personality
|
|
)
|
|
|
|
# Test valid - single speaker with all fields
|
|
profile = SpeakerProfile(
|
|
name="Test",
|
|
speakers=[
|
|
{
|
|
"name": "Host",
|
|
"voice_id": "voice123",
|
|
"backstory": "A friendly host",
|
|
"personality": "Enthusiastic and welcoming",
|
|
}
|
|
],
|
|
)
|
|
assert len(profile.speakers) == 1
|
|
assert profile.speakers[0]["name"] == "Host"
|
|
|
|
|
|
class TestPodcastService:
|
|
"""Test suite for podcast service notebook content resolution."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_submit_generation_job_uses_notebook_context_content(self):
|
|
"""Test notebook podcast jobs submit real source content, not model repr."""
|
|
notebook = Notebook(id="notebook:test", name="Test", description="Test")
|
|
sources = [
|
|
Source(
|
|
id="source:first",
|
|
title="First Source",
|
|
full_text="First source full text for submitted podcast content.",
|
|
),
|
|
Source(
|
|
id="source:second",
|
|
title="Second Source",
|
|
full_text="Second source full text for submitted podcast content.",
|
|
),
|
|
]
|
|
submitted_args = {}
|
|
|
|
async def fake_get_sources(self, include_full_text=False):
|
|
return sources
|
|
|
|
async def fake_get_notes(self, include_content=False):
|
|
return []
|
|
|
|
async def fake_get_for_sources(cls, source_ids):
|
|
return {sid: [] for sid in source_ids}
|
|
|
|
def fake_submit_command(app_name, command_name, command_args):
|
|
submitted_args.update(command_args)
|
|
return "command:podcast"
|
|
|
|
fake_commands_module = ModuleType("commands.podcast_commands")
|
|
# The real commands/__init__.py runs `from .podcast_commands import
|
|
# generate_podcast_command` when the package is imported, so the fake
|
|
# submodule must expose that name or the package import fails before the
|
|
# patched submit_command is reached.
|
|
# setattr: dynamic module attribute mypy can't know about
|
|
setattr(fake_commands_module, "generate_podcast_command", lambda *a, **k: None)
|
|
|
|
with (
|
|
patch.object(
|
|
EpisodeProfile, "get_by_name", new=AsyncMock(return_value=object())
|
|
),
|
|
patch.object(
|
|
SpeakerProfile,
|
|
"get_by_name",
|
|
new=AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
id="speaker_profile:speakers", name="Speakers"
|
|
)
|
|
),
|
|
),
|
|
patch.object(Notebook, "get", new=AsyncMock(return_value=notebook)),
|
|
patch.object(Notebook, "get_sources", new=fake_get_sources),
|
|
patch.object(Notebook, "get_notes", new=fake_get_notes),
|
|
patch.object(
|
|
SourceInsight, "get_for_sources", new=classmethod(fake_get_for_sources)
|
|
),
|
|
patch("api.podcast_service.submit_command", new=fake_submit_command),
|
|
patch.dict(
|
|
sys.modules, {"commands.podcast_commands": fake_commands_module}
|
|
),
|
|
):
|
|
job_id = await PodcastService.submit_generation_job(
|
|
episode_profile_name="Episode",
|
|
speaker_profile_name="Speakers",
|
|
episode_name="Episode Name",
|
|
notebook_id="notebook:test",
|
|
)
|
|
|
|
assert job_id == "command:podcast"
|
|
content = submitted_args["content"]
|
|
assert "First source full text for submitted podcast content." in content
|
|
assert "Second source full text for submitted podcast content." in content
|
|
assert "Notebook(id=" not in content
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 7: Transformation Domain
|
|
# ============================================================================
|
|
|
|
|
|
class TestTransformationDomain:
|
|
"""Test suite for Transformation domain model."""
|
|
|
|
def test_transformation_creation(self):
|
|
"""Test transformation model creation."""
|
|
transform = Transformation(
|
|
name="summarize",
|
|
title="Summarize Content",
|
|
description="Creates a summary",
|
|
prompt="Summarize the following text: {content}",
|
|
apply_default=True,
|
|
)
|
|
|
|
assert transform.name == "summarize"
|
|
assert transform.apply_default is True
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 8: Content Settings
|
|
# ============================================================================
|
|
|
|
|
|
class TestContentSettings:
|
|
"""Test suite for ContentSettings defaults."""
|
|
|
|
def teardown_method(self):
|
|
# ContentSettings is a RecordModel singleton; drop any instance these
|
|
# tests created so a non-default value can't leak into other tests.
|
|
ContentSettings.clear_instance()
|
|
|
|
def test_content_settings_defaults(self):
|
|
"""Test ContentSettings has proper defaults."""
|
|
settings = ContentSettings()
|
|
|
|
assert settings.record_id == "open_notebook:content_settings"
|
|
assert settings.default_content_processing_engine_doc == "auto"
|
|
assert settings.default_embedding_option == "ask"
|
|
assert settings.auto_delete_files == "yes"
|
|
assert settings.youtube_preferred_languages is not None
|
|
assert len(settings.youtube_preferred_languages) > 0
|
|
|
|
def test_content_settings_accepts_crawl4ai_url_engine(self):
|
|
"""crawl4ai is a valid URL processing engine (content-core 2.x)."""
|
|
settings = ContentSettings(default_content_processing_engine_url="crawl4ai")
|
|
assert settings.default_content_processing_engine_url == "crawl4ai"
|
|
|
|
def test_docling_ocr_defaults_on(self):
|
|
"""OCR is on by default (matches content-core's docling_ocr default)."""
|
|
settings = ContentSettings()
|
|
assert settings.docling_ocr is True
|
|
assert ContentSettings(docling_ocr=False).docling_ocr is False
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE 9: Episode Profile Validation
|
|
# ============================================================================
|
|
|
|
|
|
class TestEpisodeProfile:
|
|
"""Test suite for EpisodeProfile validation."""
|
|
|
|
def test_episode_profile_segment_validation(self):
|
|
"""Test segment count validation (3-20)."""
|
|
# Test invalid - too few segments
|
|
with pytest.raises(
|
|
ValidationError, match="Number of segments must be between 3 and 20"
|
|
):
|
|
EpisodeProfile(
|
|
name="Test",
|
|
speaker_config="default",
|
|
default_briefing="Test briefing",
|
|
num_segments=2,
|
|
)
|
|
|
|
# Test invalid - too many segments
|
|
with pytest.raises(
|
|
ValidationError, match="Number of segments must be between 3 and 20"
|
|
):
|
|
EpisodeProfile(
|
|
name="Test",
|
|
speaker_config="default",
|
|
default_briefing="Test briefing",
|
|
num_segments=21,
|
|
)
|
|
|
|
# Test valid segment count
|
|
profile = EpisodeProfile(
|
|
name="Test",
|
|
speaker_config="default",
|
|
default_briefing="Test briefing",
|
|
num_segments=5,
|
|
)
|
|
assert profile.num_segments == 5
|
|
|
|
|
|
# ============================================================================
|
|
# TEST SUITE: Credential flexible config bag (#875)
|
|
# ============================================================================
|
|
|
|
|
|
class TestCredentialConfigBag:
|
|
"""Provider-specific extras (num_ctx) round-trip through the flexible
|
|
`config` object instead of a dedicated SCHEMAFULL column."""
|
|
|
|
def test_prepare_save_data_packs_num_ctx_into_config(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
cred = Credential(name="Local Ollama", provider="ollama", num_ctx=16384)
|
|
data = cred._prepare_save_data()
|
|
|
|
assert data["config"] == {"num_ctx": 16384}
|
|
assert "num_ctx" not in data # not a top-level column anymore
|
|
|
|
def test_prepare_save_data_config_none_when_no_extras(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
cred = Credential(name="OpenAI", provider="openai")
|
|
data = cred._prepare_save_data()
|
|
|
|
assert data["config"] is None
|
|
assert "num_ctx" not in data
|
|
|
|
def test_db_row_with_config_lifts_num_ctx_to_top_level(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
# Simulates a row read back from the DB
|
|
cred = Credential(
|
|
name="Local Ollama",
|
|
provider="ollama",
|
|
config={"num_ctx": 8192},
|
|
)
|
|
|
|
assert cred.num_ctx == 8192 # mirrored from config onto the convenience field
|
|
|
|
def test_num_ctx_round_trips_through_save_and_load(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
original = Credential(name="Local Ollama", provider="ollama", num_ctx=4096)
|
|
persisted = original._prepare_save_data()
|
|
# Rebuild from the persisted shape (as the DB would return it)
|
|
reloaded = Credential(
|
|
name=persisted["name"],
|
|
provider=persisted["provider"],
|
|
config=persisted["config"],
|
|
)
|
|
|
|
assert reloaded.num_ctx == 4096
|
|
assert reloaded.to_esperanto_config()["num_ctx"] == 4096
|
|
|
|
def test_null_config_loads_without_extras(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
cred = Credential(name="OpenAI", provider="openai", config=None)
|
|
|
|
assert cred.num_ctx is None
|
|
assert cred.config is None
|
|
assert cred._prepare_save_data()["config"] is None
|
|
|
|
def test_unmapped_config_keys_are_preserved_on_save(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
# A newer version may have written config keys this model doesn't map.
|
|
# They must survive a load/save round-trip rather than be clobbered
|
|
# (repo_update MERGE replaces the whole config object).
|
|
cred = Credential(
|
|
name="Local Ollama",
|
|
provider="ollama",
|
|
config={"num_ctx": 8192, "future_option": "keep-me"},
|
|
)
|
|
assert cred.num_ctx == 8192
|
|
|
|
data = cred._prepare_save_data()
|
|
assert data["config"] == {"num_ctx": 8192, "future_option": "keep-me"}
|
|
|
|
def test_clearing_num_ctx_keeps_other_config_keys(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
cred = Credential(
|
|
name="Local Ollama",
|
|
provider="ollama",
|
|
config={"num_ctx": 8192, "future_option": "keep-me"},
|
|
)
|
|
cred.num_ctx = None # user clears the override
|
|
|
|
data = cred._prepare_save_data()
|
|
assert data["config"] == {"future_option": "keep-me"}
|
|
|
|
def test_mirrored_num_ctx_is_validated_as_int(self):
|
|
from open_notebook.domain.credential import Credential
|
|
|
|
# A value coming from the flexible config bag is routed through normal
|
|
# Pydantic field validation, not set raw.
|
|
cred = Credential(
|
|
name="Local Ollama", provider="ollama", config={"num_ctx": "8192"}
|
|
)
|
|
assert cred.num_ctx == 8192
|
|
assert isinstance(cred.num_ctx, int)
|
|
|
|
# A non-coercible value is rejected rather than silently flowing through.
|
|
with pytest.raises(ValidationError):
|
|
Credential(
|
|
name="Local Ollama", provider="ollama", config={"num_ctx": "not-an-int"}
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|