mirror of
https://github.com/aljazceru/nutshell.git
synced 2025-12-20 10:34:20 +01:00
464c0e0ea4
* make backups before migrations * database tests * postgres db backup with location string * ignore version in pg_dump and throw warning * install latest pg_dump * install latest * pg update? * remove test from github * skip for postgres on github actions
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
import datetime
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from cashu.core import db
|
|
from cashu.core.db import Connection, timestamp_now
|
|
from cashu.core.migrations import backup_database
|
|
from cashu.core.settings import settings
|
|
from cashu.mint.ledger import Ledger
|
|
from tests.helpers import is_github_actions, is_postgres
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(
|
|
is_github_actions and is_postgres,
|
|
reason=(
|
|
"Fails on GitHub Actions because pg_dump is not the same version as postgres"
|
|
),
|
|
)
|
|
async def test_backup_db_migration(ledger: Ledger):
|
|
settings.db_backup_path = "./test_data/backups/"
|
|
filepath = await backup_database(ledger.db, 999)
|
|
assert os.path.exists(filepath)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timestamp_now(ledger: Ledger):
|
|
ts = timestamp_now(ledger.db)
|
|
if ledger.db.type == db.SQLITE:
|
|
assert isinstance(ts, str)
|
|
assert int(ts) <= time.time()
|
|
elif ledger.db.type in {db.POSTGRES, db.COCKROACH}:
|
|
assert isinstance(ts, str)
|
|
datetime.datetime.strptime(ts, "%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_connection(ledger: Ledger):
|
|
async with ledger.db.connect() as conn:
|
|
assert isinstance(conn, Connection)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_db_tables(ledger: Ledger):
|
|
async with ledger.db.connect() as conn:
|
|
if ledger.db.type == db.SQLITE:
|
|
tables_res = await conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table';"
|
|
)
|
|
elif ledger.db.type in {db.POSTGRES, db.COCKROACH}:
|
|
tables_res = await conn.execute(
|
|
"SELECT table_name FROM information_schema.tables WHERE table_schema ="
|
|
" 'public';"
|
|
)
|
|
tables = [t[0] for t in await tables_res.fetchall()]
|
|
tables_expected = [
|
|
"dbversions",
|
|
"keysets",
|
|
"proofs_used",
|
|
"proofs_pending",
|
|
"melt_quotes",
|
|
"mint_quotes",
|
|
"mint_pubkeys",
|
|
"promises",
|
|
]
|
|
for table in tables_expected:
|
|
assert table in tables
|