170 lines
7.0 KiB
Python
170 lines
7.0 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
from decimal import Decimal
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from btcdata_archive import sqlite_archive as archive
|
|
from btcdata_archive.cli import parser, main
|
|
from btcdata_archive.config import Config, load_env_file
|
|
from btcdata_archive.errors import SafetyError
|
|
from btcdata_archive.schema import TABLES, Digest, encode_row, decode_row, canonical_json
|
|
from btcdata_archive.weeks import boundaries, week_start, weeks, timestamp
|
|
|
|
UTC = timezone.utc
|
|
START = datetime(2026, 8, 31, tzinfo=UTC)
|
|
|
|
|
|
@pytest.mark.parametrize('instant,anchor,cutoff', [
|
|
('2026-10-05T00:00:00+00:00', '2026-10-05', '2026-09-07'),
|
|
('2026-10-04T23:59:59+00:00', '2026-09-28', '2026-08-31'),
|
|
('2027-01-01T12:00:00+00:00', '2026-12-28', '2026-11-30'),
|
|
('2024-02-29T12:00:00+00:00', '2024-02-26', '2024-01-29'),
|
|
('2026-03-29T03:30:00+02:00', '2026-03-23', '2026-02-23'),
|
|
])
|
|
def test_boundaries(instant, anchor, cutoff):
|
|
result = boundaries(datetime.fromisoformat(instant))
|
|
assert [d.date().isoformat() for d in result] == [anchor, cutoff]
|
|
|
|
|
|
def test_week_intervals():
|
|
assert list(weeks(START + timedelta(days=2), START + timedelta(days=21))) == [START, START + timedelta(days=7), START + timedelta(days=14)]
|
|
assert list(weeks(None, START)) == []
|
|
assert list(weeks(START, START)) == []
|
|
with pytest.raises(ValueError):
|
|
week_start(datetime(2026, 1, 1))
|
|
assert TABLES[0].retention == TABLES[1].retention == 'recorded_at'
|
|
assert TABLES[2].retention == 'ts'
|
|
|
|
|
|
def test_exact_roundtrips(tmp_path):
|
|
path = tmp_path / 'week.sqlite'
|
|
archive.create(path)
|
|
values = [
|
|
(2**63 - 1, Decimal('1234567890.12345678'), START.replace(microsecond=123456)),
|
|
(2**63 - 2, '{"z":null,"b":12345678901234567890.12345678901234567890,"a":[1,true]}', START, None),
|
|
(int(START.timestamp()), *[Decimal('0.00000001')] * 6, 0, START.replace(microsecond=999999)),
|
|
]
|
|
db = archive.connect(path)
|
|
for table, row in zip(TABLES, values):
|
|
encoded = encode_row(table, row)
|
|
archive.upsert(db, table, [encoded])
|
|
assert encode_row(table, decode_row(table, encoded)) == encoded
|
|
db.commit()
|
|
db.close()
|
|
meta = archive.finalize(path, START, 'btcprod', START, 1)
|
|
assert all(s['count'] == 1 for s in meta['tables'].values())
|
|
db = archive.connect(path, readonly=True)
|
|
rates = db.execute('SELECT rates, quote_sats FROM strike_price').fetchone()
|
|
assert rates[1] is None
|
|
assert '12345678901234567890.12345678901234567890' in rates[0]
|
|
assert json.loads(rates[0])['z'] is None
|
|
db.close()
|
|
assert not list(tmp_path.glob('*-wal'))
|
|
assert timestamp(START.replace(microsecond=1)).endswith('.000001Z')
|
|
|
|
|
|
@pytest.mark.parametrize('price', ['1.000000001', '10000000000.00000000'])
|
|
def test_decimal_type_limits(tmp_path, price):
|
|
path = tmp_path / 'week.sqlite'
|
|
archive.create(path)
|
|
db = archive.connect(path)
|
|
archive.upsert(db, TABLES[0], [(1, price, timestamp(START))])
|
|
db.commit()
|
|
db.close()
|
|
with pytest.raises(SafetyError, match='precision/scale'):
|
|
archive.finalize(path, START, 'btcprod', START, 2)
|
|
|
|
|
|
def test_json_and_digest():
|
|
assert canonical_json('{"b": 1.234567890123456789, "a": null}') == '{"a":null,"b":1.234567890123456789}'
|
|
a, b = Digest(), Digest()
|
|
a.add((1, 'a,b', None))
|
|
b.add((1, 'a', 'b'))
|
|
assert a.result() != b.result()
|
|
|
|
|
|
def test_empty_and_corruption(tmp_path):
|
|
path = tmp_path / 'week.sqlite'
|
|
archive.create(path)
|
|
assert all(v['count'] == 0 for v in archive.finalize(path, START, 'btcprod', START, 2)['tables'].values())
|
|
with pytest.raises(ValueError, match='metadata'):
|
|
archive.validate(path, START, 'other', 2)
|
|
db = archive.connect(path)
|
|
db.execute('ALTER TABLE strike_price ADD COLUMN extra TEXT')
|
|
db.commit()
|
|
db.close()
|
|
with pytest.raises(ValueError, match='schema'):
|
|
archive.validate(path, START, 'btcprod', 2)
|
|
path.write_bytes(b'corrupt')
|
|
with pytest.raises(sqlite3.DatabaseError):
|
|
archive.validate(path, START, 'btcprod', 2)
|
|
|
|
|
|
def test_outside_week_and_checksum(tmp_path):
|
|
path = tmp_path / 'week.sqlite'
|
|
archive.create(path)
|
|
db = archive.connect(path)
|
|
archive.upsert(db, TABLES[0], [(1, '1.00000000', timestamp(START))])
|
|
db.commit()
|
|
db.close()
|
|
archive.finalize(path, START, 'btcprod', START, 2)
|
|
db = archive.connect(path)
|
|
db.execute("UPDATE binance_price SET price='2.00000000'")
|
|
db.commit()
|
|
db.close()
|
|
with pytest.raises(ValueError, match='checksum'):
|
|
archive.validate(path, START, 'btcprod', 2)
|
|
db = archive.connect(path)
|
|
db.execute('UPDATE binance_price SET recorded_at=?', (timestamp(START + timedelta(days=7)),))
|
|
db.commit()
|
|
db.close()
|
|
with pytest.raises(ValueError, match='outside week'):
|
|
archive.finalize(path, START, 'btcprod', START, 2)
|
|
|
|
|
|
def test_literal_env_file(tmp_path, monkeypatch):
|
|
path = tmp_path / '.env'
|
|
path.write_text("# comment\nTEST_ARCHIVE_VALUE='space # $value'\nTEST_ARCHIVE_KEEP=file\nTEST_ARCHIVE_EMPTY=\nTEST_ARCHIVE_LITERAL='$(touch should-not-exist)'\n")
|
|
for key in ('TEST_ARCHIVE_VALUE', 'TEST_ARCHIVE_KEEP', 'TEST_ARCHIVE_EMPTY', 'TEST_ARCHIVE_LITERAL'):
|
|
monkeypatch.delenv(key, raising=False)
|
|
# Register environment restoration even for variables created by the loader.
|
|
monkeypatch.setenv(key, '')
|
|
monkeypatch.delenv(key)
|
|
monkeypatch.setenv('TEST_ARCHIVE_KEEP', 'environment')
|
|
load_env_file(path)
|
|
import os
|
|
assert os.environ['TEST_ARCHIVE_VALUE'] == 'space # $value'
|
|
assert os.environ['TEST_ARCHIVE_KEEP'] == 'environment'
|
|
assert os.environ['TEST_ARCHIVE_EMPTY'] == ''
|
|
assert os.environ['TEST_ARCHIVE_LITERAL'] == '$(touch should-not-exist)'
|
|
path.write_text("TEST_ARCHIVE_VALUE='private-unclosed-value\n")
|
|
with pytest.raises(SafetyError) as error:
|
|
load_env_file(path)
|
|
assert 'private-unclosed-value' not in str(error.value)
|
|
|
|
|
|
@pytest.mark.parametrize('args', [['run'], ['run', '--mode', 'other'], ['run', '--mode', 'verify-only', '--max-weeks', '0'], ['run', '--mode', 'verify-only', '--max-weeks', '-1']])
|
|
def test_cli_rejects(args):
|
|
with pytest.raises(SystemExit):
|
|
parser().parse_args(args)
|
|
|
|
|
|
def test_config_and_safe_error(monkeypatch, capsys):
|
|
monkeypatch.delenv('ARCHIVE_ENV_FILE', raising=False)
|
|
for key in ('PGHOST','PGUSER','PGPASSWORD','S3_ENDPOINT_URL','AWS_ACCESS_KEY_ID','AWS_SECRET_ACCESS_KEY'):
|
|
monkeypatch.delenv(key, raising=False)
|
|
assert main(['run', '--mode', 'verify-only']) == 1
|
|
assert 'missing configuration' in capsys.readouterr().out
|
|
for key in ('PGHOST','PGUSER','PGPASSWORD'):
|
|
monkeypatch.setenv(key, 'test')
|
|
assert Config.from_env(need_s3=False).endpoint is None
|
|
with pytest.raises(SafetyError):
|
|
Config.from_env()
|
|
monkeypatch.setenv('S3_ENDPOINT_URL', 'http://user:password@example.com')
|
|
monkeypatch.setenv('AWS_ACCESS_KEY_ID', 'test')
|
|
monkeypatch.setenv('AWS_SECRET_ACCESS_KEY', 'test')
|
|
with pytest.raises(SafetyError):
|
|
Config.from_env()
|