89 lines
4.6 KiB
Python
89 lines
4.6 KiB
Python
"""PostgreSQL-only regressions; never connects using the production environment."""
|
|
from datetime import datetime, timezone
|
|
|
|
import psycopg
|
|
import pytest
|
|
|
|
from btcdata_archive import sqlite_archive
|
|
from btcdata_archive.schema import TABLES
|
|
from btcdata_archive.weeks import WEEK
|
|
|
|
pytestmark = pytest.mark.integration
|
|
START = datetime(2026, 2, 16, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_ledger_progress_preserves_verification_before_insert_check(postgres_infra):
|
|
pg, admin = postgres_infra.pg, postgres_infra.admin
|
|
pg.lock()
|
|
key, version, checksum = 'test/week.sqlite', 'verified-version', 'a' * 64
|
|
counts = {'binance_price': 3, 'strike_price': 0, 'kraken_minute': 0}
|
|
pg.ledger(START, 'exporting', key)
|
|
pg.ledger(START, 'verified', key, version, checksum, counts)
|
|
# These progress-only calls used to fail INSERT's CHECK before ON CONFLICT.
|
|
pg.ledger(START, 'pruning', key)
|
|
pg.ledger(START, 'pruning', key, deletions={'binance_price': 2})
|
|
pg.ledger(START, 'complete', key, deletions={'binance_price': 3})
|
|
row = admin.execute('SELECT status,object_version_id,file_sha256,row_counts,deletion_counts,completion_time IS NOT NULL FROM btcdata_archive.weeks WHERE week_start=%s', (START,)).fetchone()
|
|
assert row == ('complete', version, checksum, counts, {'binance_price': 3}, True)
|
|
pg.ledger(START, 'failed', key, error='test interruption')
|
|
pg.ledger(START, 'exporting', key)
|
|
pg.ledger(START, 'verified', key, 'new-version', 'b' * 64, counts)
|
|
pg.ledger(START, 'pruning', key)
|
|
assert admin.execute('SELECT object_version_id,file_sha256,completion_time FROM btcdata_archive.weeks').fetchone() == ('new-version', 'b' * 64, None)
|
|
|
|
|
|
def test_ledger_still_rejects_unverified_pruning(postgres_infra):
|
|
pg = postgres_infra.pg
|
|
pg.lock()
|
|
with pytest.raises(psycopg.errors.CheckViolation):
|
|
pg.ledger(START, 'pruning', 'test/week.sqlite')
|
|
pg.ledger(START, 'exporting', 'test/week.sqlite')
|
|
with pytest.raises(psycopg.errors.CheckViolation):
|
|
pg.ledger(START, 'complete', 'test/week.sqlite')
|
|
assert postgres_infra.admin.execute('SELECT status FROM btcdata_archive.weeks').fetchone() == ('exporting',)
|
|
|
|
|
|
def test_typed_pruning_changed_rows_and_recovery(postgres_infra, tmp_path):
|
|
pg, admin, config = postgres_infra.pg, postgres_infra.admin, postgres_infra.config
|
|
pg.lock()
|
|
pg.check_schema()
|
|
admin.execute('INSERT INTO binance_price VALUES (1,1.00000001,%s)', (START,))
|
|
admin.execute('INSERT INTO strike_price VALUES (1,%s::jsonb,%s,NULL)', ('{"a":12345678901234567890.12345678901234567890}', START))
|
|
admin.execute('INSERT INTO kraken_minute VALUES (%s,1.00000001,2,3,4,5,1234567890123456.12345678,7,%s)', (int(START.timestamp()), START.replace(microsecond=123456)))
|
|
path = tmp_path / 'verified.sqlite'
|
|
sqlite_archive.create(path)
|
|
key = 'test/week.sqlite'
|
|
|
|
def export_and_record(version):
|
|
pg.ledger(START, 'exporting', key)
|
|
pg.export(path, START)
|
|
meta = sqlite_archive.finalize(path, START, config.pg['dbname'], START, 2)
|
|
pg.ledger(START, 'verified', key, version, sqlite_archive.file_hash(path),
|
|
{name: stats['count'] for name, stats in meta['tables'].items()})
|
|
pg.ledger(START, 'pruning', key)
|
|
|
|
def progress(deleted):
|
|
pg.ledger(START, 'pruning', key, deletions=deleted)
|
|
|
|
export_and_record('test-v1')
|
|
# Source mutation and late insertion must remain; unchanged rows can be deleted.
|
|
admin.execute('UPDATE binance_price SET price=9 WHERE id=1')
|
|
admin.execute("INSERT INTO strike_price VALUES (2,'null'::jsonb,%s,NULL)", (START,))
|
|
with pytest.raises(RuntimeError, match='reconciliation required'):
|
|
pg.prune(path, START, START + WEEK, progress)
|
|
assert admin.execute('SELECT id FROM binance_price').fetchall() == [(1,)]
|
|
assert admin.execute('SELECT id FROM strike_price').fetchall() == [(2,)]
|
|
assert admin.execute('SELECT count(*) FROM kraken_minute').fetchone() == (0,)
|
|
pg.ledger(START, 'failed', key, error='test reconciliation')
|
|
export_and_record('test-v2')
|
|
deleted = pg.prune(path, START, START + WEEK, progress)
|
|
pg.ledger(START, 'complete', key, deletions=deleted)
|
|
assert all(pg.counts(START, START + WEEK)[t.name] == 0 for t in TABLES)
|
|
db = sqlite_archive.connect(path, readonly=True)
|
|
try:
|
|
assert db.execute('SELECT id,price FROM binance_price').fetchall() == [(1, '9.00000000')]
|
|
assert db.execute('SELECT id,rates,quote_sats FROM strike_price ORDER BY id').fetchall() == [(1, '{"a":12345678901234567890.12345678901234567890}', None), (2, 'null', None)]
|
|
assert db.execute('SELECT count(*) FROM kraken_minute').fetchone() == (1,)
|
|
finally:
|
|
db.close()
|