Navegador

navegador / tests / test_migrations.py

Source Blame History 126 lines

0838495…	lmata	1	"""Tests for navegador.graph.migrations — schema versioning and migration."""
0838495…	lmata	2
0838495…	lmata	3	from unittest.mock import MagicMock, call
0838495…	lmata	4
0838495…	lmata	5	import pytest
0838495…	lmata	6
0838495…	lmata	7	from navegador.graph.migrations import (
0838495…	lmata	8	CURRENT_SCHEMA_VERSION,
0838495…	lmata	9	_migrations,
0838495…	lmata	10	get_schema_version,
0838495…	lmata	11	migrate,
0838495…	lmata	12	needs_migration,
0838495…	lmata	13	set_schema_version,
0838495…	lmata	14	)
0838495…	lmata	15
0838495…	lmata	16
0838495…	lmata	17	def _mock_store(version=None):
0838495…	lmata	18	store = MagicMock()
0838495…	lmata	19	if version is None:
0838495…	lmata	20	store.query.return_value = MagicMock(result_set=[])
0838495…	lmata	21	else:
0838495…	lmata	22	store.query.return_value = MagicMock(result_set=[[version]])
0838495…	lmata	23	return store
0838495…	lmata	24
0838495…	lmata	25
0838495…	lmata	26	# ── get_schema_version ───────────────────────────────────────────────────────
0838495…	lmata	27
0838495…	lmata	28	class TestGetSchemaVersion:
0838495…	lmata	29	def test_returns_zero_for_empty_graph(self):
0838495…	lmata	30	store = _mock_store(version=None)
0838495…	lmata	31	assert get_schema_version(store) == 0
0838495…	lmata	32
0838495…	lmata	33	def test_returns_zero_for_null_version(self):
0838495…	lmata	34	store = MagicMock()
0838495…	lmata	35	store.query.return_value = MagicMock(result_set=[[None]])
0838495…	lmata	36	assert get_schema_version(store) == 0
0838495…	lmata	37
0838495…	lmata	38	def test_returns_stored_version(self):
0838495…	lmata	39	store = _mock_store(version=2)
0838495…	lmata	40	assert get_schema_version(store) == 2
0838495…	lmata	41
0838495…	lmata	42
0838495…	lmata	43	# ── set_schema_version ──────────────────────────────────────────────────────
0838495…	lmata	44
0838495…	lmata	45	class TestSetSchemaVersion:
0838495…	lmata	46	def test_calls_query_with_merge(self):
0838495…	lmata	47	store = MagicMock()
0838495…	lmata	48	set_schema_version(store, 3)
0838495…	lmata	49	store.query.assert_called_once()
0838495…	lmata	50	cypher = store.query.call_args[0][0]
0838495…	lmata	51	assert "MERGE" in cypher
0838495…	lmata	52	assert store.query.call_args[0][1]["version"] == 3
0838495…	lmata	53
0838495…	lmata	54
0838495…	lmata	55	# ── needs_migration ──────────────────────────────────────────────────────────
0838495…	lmata	56
0838495…	lmata	57	class TestNeedsMigration:
0838495…	lmata	58	def test_true_when_behind(self):
0838495…	lmata	59	store = _mock_store(version=0)
0838495…	lmata	60	assert needs_migration(store) is True
0838495…	lmata	61
0838495…	lmata	62	def test_false_when_current(self):
0838495…	lmata	63	store = _mock_store(version=CURRENT_SCHEMA_VERSION)
0838495…	lmata	64	assert needs_migration(store) is False
0838495…	lmata	65
0838495…	lmata	66
0838495…	lmata	67	# ── migrate ──────────────────────────────────────────────────────────────────
0838495…	lmata	68
0838495…	lmata	69	class TestMigrate:
0838495…	lmata	70	def test_applies_all_migrations_from_zero(self):
0838495…	lmata	71	call_log = []
0838495…	lmata	72
0838495…	lmata	73	def track_query(cypher, params=None):
0838495…	lmata	74	call_log.append(cypher)
0838495…	lmata	75	result = MagicMock()
0838495…	lmata	76	# get_schema_version query returns no rows initially
0838495…	lmata	77	if "Meta" in cypher and "RETURN" in cypher:
0838495…	lmata	78	result.result_set = []
0838495…	lmata	79	else:
0838495…	lmata	80	result.result_set = []
0838495…	lmata	81	return result
0838495…	lmata	82
0838495…	lmata	83	store = MagicMock()
0838495…	lmata	84	store.query.side_effect = track_query
0838495…	lmata	85
0838495…	lmata	86	applied = migrate(store)
0838495…	lmata	87	assert applied == list(range(1, CURRENT_SCHEMA_VERSION + 1))
0838495…	lmata	88
0838495…	lmata	89	def test_no_op_when_already_current(self):
0838495…	lmata	90	store = _mock_store(version=CURRENT_SCHEMA_VERSION)
0838495…	lmata	91	applied = migrate(store)
0838495…	lmata	92	assert applied == []
0838495…	lmata	93
0838495…	lmata	94	def test_raises_on_missing_migration(self):
0838495…	lmata	95	# Temporarily remove a migration to trigger the RuntimeError
0838495…	lmata	96	saved = _migrations.pop(0)
0838495…	lmata	97	try:
0838495…	lmata	98	store = _mock_store(version=None)
0838495…	lmata	99	with pytest.raises(RuntimeError, match="No migration registered"):
0838495…	lmata	100	migrate(store)
0838495…	lmata	101	finally:
0838495…	lmata	102	_migrations[0] = saved
0838495…	lmata	103
0838495…	lmata	104
0838495…	lmata	105	# ── migrations registry ─────────────────────────────────────────────────────
0838495…	lmata	106
0838495…	lmata	107	class TestMigrationsRegistry:
0838495…	lmata	108	def test_has_migration_for_each_version(self):
0838495…	lmata	109	for v in range(CURRENT_SCHEMA_VERSION):
0838495…	lmata	110	assert v in _migrations, f"Missing migration for version {v} -> {v + 1}"
0838495…	lmata	111
0838495…	lmata	112	def test_current_version_is_positive(self):
0838495…	lmata	113	assert CURRENT_SCHEMA_VERSION > 0
0838495…	lmata	114
0838495…	lmata	115	def test_migration_0_to_1_runs(self):
0838495…	lmata	116	store = MagicMock()
0838495…	lmata	117	store.query.return_value = MagicMock(result_set=[])
0838495…	lmata	118	_migrations[0](store)
0838495…	lmata	119
0838495…	lmata	120	def test_migration_1_to_2_sets_content_hash(self):
0838495…	lmata	121	store = MagicMock()
0838495…	lmata	122	store.query.return_value = MagicMock(result_set=[])
0838495…	lmata	123	_migrations[1](store)
0838495…	lmata	124	store.query.assert_called_once()
0838495…	lmata	125	cypher = store.query.call_args[0][0]
0838495…	lmata	126	assert "content_hash" in cypher

Navegador

Keyboard Shortcuts