97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
import datetime
|
|
import pandas as pd
|
|
from database import make_query, make_insert, Database, HDict, SQLite
|
|
from exchangeinterface import ExchangeInterface
|
|
from passlib.hash import bcrypt
|
|
from sqlalchemy import create_engine, text
|
|
import config
|
|
|
|
|
|
def test():
|
|
# un_hashed_pass = 'password'
|
|
hasher = bcrypt.using(rounds=13)
|
|
# hashed_pass = hasher.hash(un_hashed_pass)
|
|
# print(f'password: {un_hashed_pass}')
|
|
# print(f'hashed pass: {hashed_pass}')
|
|
# print(f" right pass: {hasher.verify('password', hashed_pass)}")
|
|
# print(f" wrong pass: {hasher.verify('passWord', hashed_pass)}")
|
|
engine = create_engine("sqlite:///" + config.DB_FILE, echo=True)
|
|
with engine.connect() as conn:
|
|
default_user = pd.read_sql_query(sql=text("SELECT * FROM users WHERE user_name = 'guest'"), con=conn)
|
|
# hashed_password = default_user.password.values[0]
|
|
# print(f" verify pass: {hasher.verify('password', hashed_password)}")
|
|
username = default_user.user_name.values[0]
|
|
print(username)
|
|
|
|
|
|
def test_make_query():
|
|
values = {'first_field': 'first_value'}
|
|
item = 'market_id'
|
|
table = 'markets'
|
|
q_str = make_query(item=item, table=table, values=values)
|
|
print(f'\nWith one indexing field: {q_str}')
|
|
|
|
values = {'first_field': 'first_value', 'second_field': 'second_value'}
|
|
q_str = make_query(item=item, table=table, values=values)
|
|
print(f'\nWith two indexing fields: {q_str}')
|
|
assert q_str is not None
|
|
|
|
|
|
def test_make_insert():
|
|
table = 'markets'
|
|
values = {'first_field': 'first_value'}
|
|
q_str = make_insert(table=table, values=values)
|
|
print(f'\nWith one indexing field: {q_str}')
|
|
|
|
values = {'first_field': 'first_value', 'second_field': 'second_value'}
|
|
q_str = make_insert(table=table, values=values)
|
|
print(f'\nWith two indexing fields: {q_str}')
|
|
assert q_str is not None
|
|
|
|
|
|
def test__table_exists():
|
|
exchanges = ExchangeInterface()
|
|
d_obj = Database(exchanges)
|
|
exists = d_obj._table_exists('BTC/USD_5m_alpaca')
|
|
print(f'\nExists - Should be true: {exists}')
|
|
assert exists
|
|
exists = d_obj._table_exists('BTC/USD_5m_alpina')
|
|
print(f'Doesnt exist - should be false: {exists}')
|
|
assert not exists
|
|
|
|
|
|
def test_get_from_static_table():
|
|
exchanges = ExchangeInterface()
|
|
d_obj = Database(exchanges)
|
|
market_id = d_obj._fetch_market_id('BTC/USD', 'alpaca')
|
|
e_id = d_obj._fetch_exchange_id('alpaca')
|
|
print(f'market id: {market_id}')
|
|
assert market_id > 0
|
|
print(f'exchange_name ID: {e_id}')
|
|
assert e_id == 4
|
|
|
|
|
|
def test_populate_table():
|
|
"""
|
|
Populates a database table with records from the exchange_name.
|
|
:param table_name: str - The name of the table in the database.
|
|
:param start_time: datetime - The starting time to fetch the records from.
|
|
:param end_time: datetime - The end time to get the records until.
|
|
:return: pdDataframe: - The data that was downloaded.
|
|
"""
|
|
exchanges = ExchangeInterface()
|
|
d_obj = Database(exchanges)
|
|
d_obj._populate_table(table_name='BTC/USD_2h_alpaca',
|
|
start_time=datetime.datetime(year=2023, month=3, day=27, hour=6, minute=0))
|
|
|
|
|
|
def test_get_records_since():
|
|
exchanges = ExchangeInterface()
|
|
d_obj = Database(exchanges)
|
|
records = d_obj.get_records_since(table_name='BTC/USD_15m_alpaca',
|
|
st=datetime.datetime(year=2023, month=3, day=27, hour=1, minute=0),
|
|
et=datetime.datetime.utcnow(),
|
|
rl=15)
|
|
print(records)
|
|
assert records is not None
|