test(conftest): add in_memory_db and task manager singleton reset fixtures; fix broken tests
- Replace conftest.py with two autouse fixtures: reset_graph_factory_singleton and reset_task_manager_singleton, plus in_memory_db fixture for tests requiring a real DB - Rewrite test_project_task_recovery.py to use ProjectManager + SQLite in-memory DB instead of the removed Project dataclass (same guarantee: active_task_id persists and defaults to None) - Fix db.py init_db() to import db_models before Base.metadata.create_all() so all ORM tables are registered and created on first startup Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
0b760eed0b
commit
9ed2412745
|
|
@ -19,6 +19,8 @@ def init_db(database_url: str) -> None:
|
|||
connect_args = {"check_same_thread": False} if database_url.startswith("sqlite") else {}
|
||||
_engine = create_engine(database_url, connect_args=connect_args, echo=False)
|
||||
_SessionLocal = sessionmaker(bind=_engine, autocommit=False, autoflush=False)
|
||||
# Import models so that all ORM classes register with Base.metadata before create_all
|
||||
from .models import db_models as _ # noqa: F401
|
||||
Base.metadata.create_all(_engine)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,12 +1,38 @@
|
|||
# backend/tests/conftest.py
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from backend.app.db import Base
|
||||
import backend.app.db as db_module
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_graph_factory_singleton():
|
||||
"""Reset the graph backend singleton before each test to avoid cross-test contamination."""
|
||||
"""Reset the graph backend singleton before each test."""
|
||||
yield
|
||||
try:
|
||||
import backend.app.graph.factory as fmod
|
||||
fmod._backend_instance = None
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_task_manager_singleton():
|
||||
"""Reset TaskManager singleton between tests."""
|
||||
from backend.app.models import task as task_module
|
||||
task_module.TaskManager._instance = None
|
||||
yield
|
||||
task_module.TaskManager._instance = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def in_memory_db():
|
||||
"""BD SQLite en memòria per a tests que necessiten BD."""
|
||||
db_module._engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
||||
db_module._SessionLocal = sessionmaker(bind=db_module._engine, autocommit=False, autoflush=False)
|
||||
Base.metadata.create_all(db_module._engine)
|
||||
yield db_module._engine
|
||||
Base.metadata.drop_all(db_module._engine)
|
||||
db_module._engine = None
|
||||
db_module._SessionLocal = None
|
||||
|
|
|
|||
|
|
@ -1,31 +1,56 @@
|
|||
# backend/tests/test_project_task_recovery.py
|
||||
"""
|
||||
Tests per a la garantia que active_task_id es persisteix i es recupera.
|
||||
Equivalent als tests originals sobre Project.to_dict/from_dict,
|
||||
ara adaptats a ProjectManager + BD SQLAlchemy.
|
||||
"""
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from backend.app.db import Base
|
||||
import backend.app.db as db_module
|
||||
import backend.app.models.db_models # registra tots els models ORM
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolated_db():
|
||||
"""BD SQLite en memòria per a cada test."""
|
||||
db_module._engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
|
||||
db_module._SessionLocal = sessionmaker(bind=db_module._engine, autocommit=False, autoflush=False)
|
||||
Base.metadata.create_all(db_module._engine)
|
||||
yield
|
||||
Base.metadata.drop_all(db_module._engine)
|
||||
db_module._engine = None
|
||||
db_module._SessionLocal = None
|
||||
|
||||
|
||||
def test_project_serializes_active_task_id():
|
||||
"""active_task_id is included in Project.to_dict()."""
|
||||
from app.models.project import Project, ProjectStatus
|
||||
p = Project(
|
||||
project_id="proj-1", name="Test",
|
||||
status=ProjectStatus.GRAPH_BUILDING,
|
||||
created_at="2026-01-01", updated_at="2026-01-01",
|
||||
active_task_id="task-abc-123",
|
||||
)
|
||||
assert p.to_dict()["active_task_id"] == "task-abc-123"
|
||||
"""active_task_id es pot desar i recuperar del ProjectManager."""
|
||||
from backend.app.models.project import ProjectManager
|
||||
proj = ProjectManager.create_project("Test")
|
||||
ProjectManager.save_project({
|
||||
"id": proj["id"],
|
||||
"active_task_id": "task-abc-123",
|
||||
})
|
||||
fetched = ProjectManager.get_project(proj["id"])
|
||||
assert fetched["active_task_id"] == "task-abc-123"
|
||||
|
||||
|
||||
def test_project_deserializes_active_task_id():
|
||||
"""Project.from_dict() restores active_task_id from JSON."""
|
||||
from app.models.project import Project
|
||||
data = {
|
||||
"project_id": "proj-1", "name": "Test", "status": "graph_building",
|
||||
"created_at": "2026-01-01", "updated_at": "2026-01-01",
|
||||
"active_task_id": "task-abc-123",
|
||||
}
|
||||
assert Project.from_dict(data).active_task_id == "task-abc-123"
|
||||
"""active_task_id es restaura correctament des de la BD."""
|
||||
from backend.app.models.project import ProjectManager
|
||||
proj = ProjectManager.create_project("Test")
|
||||
ProjectManager.save_project({
|
||||
"id": proj["id"],
|
||||
"active_task_id": "task-abc-456",
|
||||
})
|
||||
# Simulem que es torna a llegir (com si el servidor hagués reiniciat)
|
||||
fetched = ProjectManager.get_project(proj["id"])
|
||||
assert fetched["active_task_id"] == "task-abc-456"
|
||||
|
||||
|
||||
def test_project_active_task_id_defaults_none():
|
||||
"""active_task_id defaults to None for projects without it (backward compat)."""
|
||||
from app.models.project import Project
|
||||
data = {
|
||||
"project_id": "proj-1", "name": "Test", "status": "created",
|
||||
"created_at": "2026-01-01", "updated_at": "2026-01-01",
|
||||
}
|
||||
assert Project.from_dict(data).active_task_id is None
|
||||
"""active_task_id és None per a projectes creats sense task (compatibilitat enrere)."""
|
||||
from backend.app.models.project import ProjectManager
|
||||
proj = ProjectManager.create_project("Test")
|
||||
assert proj["active_task_id"] is None
|
||||
|
|
|
|||
Loading…
Reference in New Issue