from __future__ import annotations import state as state_module from errors import ConnectorError from state import CheckpointPolicy, apply_table_resync_requests, normalize_state, set_snapshot_tso from tests.pytest_compat import pytest def test_checkpoint_policy_fires_on_elapsed_time(monkeypatch): timestamps = iter([100.0, 105.0]) monkeypatch.setattr(state_module.time, "monotonic", lambda: next(timestamps)) policy = CheckpointPolicy(rows=100, seconds=5) assert policy.record_row() is True def test_set_snapshot_tso_is_stable(workspace): state = normalize_state({}, sync_task_id="task-001") assert set_snapshot_tso(state, 100) is True assert set_snapshot_tso(state, 100) is False assert state.snapshot["snapshot_tso"] == 100 def test_normalize_state_accepts_target_incremental_scan(workspace): state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": { "checkpoint_ts": "100", "scan": { "high_watermark": "200", "current_file": "increment/test/worker/120/2026-05-25/CDC000001.csv", "next_row": 3, }, "ddl_watermark": {"`test`.`worker`": 120}, }, }, sync_task_id="task-001", ) assert state.checkpoint_ts == 100 assert state.scan is not None assert state.scan.high_watermark == 200 assert state.scan.current_file == "increment/test/worker/120/2026-05-25/CDC000001.csv" assert state.scan.next_row == 3 assert state.to_dict()["incremental"]["scan"] == { "high_watermark": "200", "current_file": "increment/test/worker/120/2026-05-25/CDC000001.csv", "next_row": 3, } def test_normalize_state_accepts_global_snapshot_cursor(workspace): state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": { "done": False, "snapshot_tso": "100", "truncate_done": True, "current_file": "snapshot/test.worker.000000001.csv", "next_row": 3, }, "incremental": {"checkpoint_ts": "0", "ddl_watermark": {}}, }, sync_task_id="task-001", ) assert state.to_dict()["snapshot"] == { "done": False, "snapshot_tso": "100", "truncate_done": True, "current_file": "snapshot/test.worker.000000001.csv", "next_row": 3, } def test_normalize_state_rejects_missing_snapshot_done_for_non_empty_state(workspace): with pytest.raises(ConnectorError, match=r"state\.snapshot must include done"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"snapshot_tso": "100"}, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, }, sync_task_id="task-001", ) def test_normalize_state_rejects_completed_snapshot_cursor(workspace): with pytest.raises(ConnectorError, match=r"completed state\.snapshot must not include current_file"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": { "done": True, "snapshot_tso": "100", "current_file": "snapshot/test.worker.000000001.csv", }, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, }, sync_task_id="task-001", ) def test_normalize_state_rejects_snapshot_next_row_without_current_file(workspace): with pytest.raises(ConnectorError, match=r"state\.snapshot\.next_row must be 0"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": { "done": False, "snapshot_tso": "100", "current_file": None, "next_row": 1, }, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, }, sync_task_id="task-001", ) def test_normalize_state_rejects_resync_next_row_without_current_file(workspace): with pytest.raises(ConnectorError, match=r"state\.resync\.next_row must be 0"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": ["`test`.`worker`"], "snapshot_tso": "150", "snapshot_prefix": "snapshot/table-resync-001", "current_file": None, "next_row": 1, }, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, }, sync_task_id="task-001", ) def test_normalize_state_rejects_missing_sync_task_id_for_non_empty_state(workspace): with pytest.raises(ConnectorError, match="sync_task_id"): normalize_state( {"version": 1, "snapshot": {}}, sync_task_id="task-001", ) def test_normalize_state_rejects_scan_path_escape(workspace): with pytest.raises(ConnectorError, match="normalized relative object path"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": { "checkpoint_ts": "100", "scan": { "high_watermark": "200", "current_file": "../increment/test/worker/120/CDC000001.csv", "next_row": 0, }, "ddl_watermark": {}, }, }, sync_task_id="task-001", ) def test_normalize_state_rejects_resync_snapshot_prefix_escape(workspace): with pytest.raises(ConnectorError, match="normalized relative object path"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": ["`test`.`worker`"], "snapshot_prefix": "../snapshot/table-resync-001", "snapshot_tso": "150", }, }, sync_task_id="task-001", ) def test_normalize_state_rejects_non_numeric_resync_next_row(workspace): with pytest.raises(ConnectorError, match=r"state\.resync\.next_row must be non-negative"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": ["`test`.`worker`"], "snapshot_prefix": "snapshot/table-resync-001", "snapshot_tso": "150", "next_row": "not-a-number", }, }, sync_task_id="task-001", ) def test_normalize_state_rejects_prototype_incremental_fields(workspace): with pytest.raises(ConnectorError, match=r"state\.incremental contains unknown fields"): normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": { "last_seen_checkpoint_ts": 200, "last_processed_commit_ts": 121, "commit_ts_by_table": {"test.worker": 121}, "ddl_watermark": {}, "streams": {}, "dml": {}, }, }, sync_task_id="task-001", ) def test_normalize_state_preserves_ddl_watermark_without_catalog(workspace): state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "incremental": { "checkpoint_ts": "200", "ddl_watermark": {"`test`.`worker`": 120, "`other`.`table`": 9}, }, }, sync_task_id="task-001", ) assert state.incremental["ddl_watermark"] == {"`test`.`worker`": 120, "`other`.`table`": 9} assert state.needs_checkpoint is False def test_apply_table_resync_request_preserves_resync_snapshot_cursor(workspace): state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": ["`test`.`worker`"], "snapshot_prefix": "snapshot", "snapshot_tso": "150", "truncate_done": True, "current_file": "snapshot/test.worker.000000001.csv", "next_row": 10, }, "incremental": { "checkpoint_ts": "200", "ddl_watermark": {"`test`.`worker`": 120}, }, }, sync_task_id="task-001", ) assert apply_table_resync_requests(state) is True assert state.resync is not None assert state.resync.truncate_done is True assert state.resync.current_file == "snapshot/test.worker.000000001.csv" assert state.resync.next_row == 10 assert state.incremental["ddl_watermark"] == {} def test_apply_table_resync_request_matches_quoted_schema_with_dot(workspace): state = normalize_state( { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": ["`a.b`.`c`"], "snapshot_prefix": "snapshot", "snapshot_tso": "150", }, "incremental": { "checkpoint_ts": "200", "ddl_watermark": {"`a.b`.`c`": 120, "`a`.`b.c`": 130}, }, }, sync_task_id="task-001", ) assert apply_table_resync_requests(state) is True assert state.resync is not None assert state.resync.tables == ["`a.b`.`c`"] assert state.incremental["ddl_watermark"] == {"`a`.`b.c`": 130} def test_normalize_state_parses_resync_table_names(workspace): def raw_state(tables): return { "version": 1, "sync_task_id": "task-001", "snapshot": {"done": True, "snapshot_tso": "100"}, "resync": { "request_id": "table-resync-001", "phase": "snapshot", "tables": tables, "snapshot_prefix": "snapshot", "snapshot_tso": "150", }, "incremental": {"checkpoint_ts": "200", "ddl_watermark": {}}, } state = normalize_state( raw_state(["test.worker", "`a.b`.`c`"]), sync_task_id="task-001", ) assert state.resync is not None assert state.resync.tables == ["`test`.`worker`", "`a.b`.`c`"] with pytest.raises(ConnectorError, match="duplicate table"): normalize_state( raw_state(["`test`.`worker`", "test.worker"]), sync_task_id="task-001", ) with pytest.raises(ConnectorError, match="schema/table separator"): normalize_state( raw_state(["te.st.t.t"]), sync_task_id="task-001", ) def test_normalize_state_rejects_unknown_fields(workspace): with pytest.raises(ConnectorError, match=r"state\.snapshot contains unknown fields"): normalize_state( {"version": 1, "sync_task_id": "task-001", "snapshot": {"unexpected": True}}, sync_task_id="task-001", )