my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""Regression tests for [C12.a.cognee]: CogneeImporter._extract_direct
#2	read rows out of Cognee's SQLite metadata using `conn.row_factory =
#3	sqlite3.Row` and then called `row.get("id", "")` etc. sqlite3.Row does
#4	not support `.get()` — bracket access only — so each row raised
#5	AttributeError. The surrounding `except Exception: pass` swallowed it,
#6	so direct cognee imports silently returned zero rows even when the
#7	data was present.
#8
#9	Same pattern as the latent fact_recall bug surfaced by C12.a; this is
#10	the adjacent occurrence in the importer surface.
#11	"""
#12
#13	import sqlite3
#14
#15	import pytest
#16
#17	from mnemosyne.core.importers.cognee import CogneeImporter
#18
#19
#20	def _make_cognee_db(tmp_path):
#21	"""Build a minimal Cognee-shaped SQLite DB with one data_chunks row."""
#22	data_dir = tmp_path / "cognee-data"
#23	data_dir.mkdir()
#24	db_path = data_dir / "cognee_db"
#25	conn = sqlite3.connect(str(db_path))
#26	try:
#27	conn.execute("""
#28	CREATE TABLE data_chunks (
#29	id TEXT PRIMARY KEY,
#30	document_id TEXT,
#31	text TEXT,
#32	content TEXT,
#33	created_at TEXT
#34	)
#35	""")
#36	conn.execute("""
#37	INSERT INTO data_chunks (id, document_id, text, content, created_at)
#38	VALUES (?, ?, ?, ?, ?)
#39	""", (
#40	"chunk-1",
#41	"doc-1",
#42	"Alice was born in Boston.",
#43	None,
#44	"2026-05-09T00:00:00",
#45	))
#46	conn.commit()
#47	finally:
#48	conn.close()
#49	return data_dir
#50
#51
#52	class TestCogneeDirectImport:
#53
#54	def test_extract_direct_returns_rows_from_data_chunks(self, tmp_path):
#55	"""The bug surface: pre-fix, this returned [] silently because
#56	row.get on sqlite3.Row raised AttributeError and the broad
#57	except swallowed it."""
#58	data_dir = _make_cognee_db(tmp_path)
#59	importer = CogneeImporter(
#60	data_dir=str(data_dir),
#61	direct_db=True,
#62	)
#63	items = importer._extract_direct()
#64	assert items, (
#65	"_extract_direct returned empty despite seeded data_chunks "
#66	"row — the row.get on sqlite3.Row crash is masked by the "
#67	"broad except"
#68	)
#69	assert len(items) == 1
#70	item = items[0]
#71	assert item["content"] == "Alice was born in Boston."
#72	assert item["source"] == "cognee_direct"
#73	assert item["metadata"]["chunk_id"] == "chunk-1"
#74	assert item["metadata"]["document_id"] == "doc-1"
#75	assert item["timestamp"] == "2026-05-09T00:00:00"
#76
#77	def test_extract_direct_handles_null_text_falls_back_to_content(self, tmp_path):
#78	"""Existing fallback: row['text'] or row['content'] or ''. Make
#79	sure it survives the dict conversion."""
#80	data_dir = tmp_path / "cognee-data"
#81	data_dir.mkdir()
#82	db_path = data_dir / "cognee_db"
#83	conn = sqlite3.connect(str(db_path))
#84	try:
#85	conn.execute("""
#86	CREATE TABLE data_chunks (
#87	id TEXT PRIMARY KEY,
#88	document_id TEXT,
#89	text TEXT,
#90	content TEXT,
#91	created_at TEXT
#92	)
#93	""")
#94	conn.execute("""
#95	INSERT INTO data_chunks (id, document_id, text, content, created_at)
#96	VALUES (?, ?, ?, ?, ?)
#97	""", (
#98	"chunk-2",
#99	"doc-2",
#100	None,
#101	"Backup content text",
#102	"2026-05-09T00:00:01",
#103	))
#104	conn.commit()
#105	finally:
#106	conn.close()
#107
#108	importer = CogneeImporter(
#109	data_dir=str(data_dir),
#110	direct_db=True,
#111	)
#112	items = importer._extract_direct()
#113	assert items
#114	assert items[0]["content"] == "Backup content text"
#115
#116	def test_extract_direct_returns_empty_when_db_missing(self, tmp_path):
#117	"""Defensive: missing cognee_db should produce [] (not raise),
#118	which is the existing contract the broad except provides."""
#119	data_dir = tmp_path / "no-cognee"
#120	data_dir.mkdir()
#121	importer = CogneeImporter(
#122	data_dir=str(data_dir),
#123	direct_db=True,
#124	)
#125	items = importer._extract_direct()
#126	assert items == []
#127

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public