added responses mock

requirements-dev.txt

@@ -2,3 +2,4 @@
 pytest==7.2.0
 pytest-cov==4.0.0
 pytest-postgresql==4.1.1
+pytest-responses==0.5.1

src/config.py

@@ -10,6 +10,7 @@ def __init__(self):
             "DEBUG", "").lower() in ('true', '1', 't') else False
         self.symbols: list[str] = json.loads(environ.get("SYMBOLS", "[]"))
         self.start_time: str = environ.get("START_DATE", "")
+        self.end_time: str = environ.get("END_DATE", "NOW")
         self.sleep_time: int = int(environ.get("SLEEP_TIME", 60))
         self.timescaledb: TimescaleDB = TimescaleDB(
             host=environ.get("TIMESCALE_HOST", "localhost"),

src/runner.py

@@ -6,18 +6,23 @@
 import sys
 from time import sleep
 from worker import Worker
+from http.client import HTTPConnection
 
 
 def main() -> NoReturn:
     config = Config()
 
-    logging.basicConfig(
-        format='%(asctime)s %(levelname)-8s %(message)s',
-        level=logging.DEBUG if config.debug else logging.INFO,
-        datefmt='%Y-%m-%d %H:%M:%S',
-        handlers=[
-            logging.StreamHandler()
-        ])
+    HTTPConnection.debuglevel = 0
+    for handler in logging.root.handlers:
+        logging.root.removeHandler(handler)
+    logging.root.setLevel(logging.DEBUG if config.debug else logging.INFO)
+    stream_handler = logging.StreamHandler()
+    stream_formatter = logging.Formatter(
+        fmt='%(asctime)s %(levelname)-8s %(message)s'
+    )
+    stream_handler.setFormatter(stream_formatter)
+    logging.root.addHandler(stream_handler)
+
     logger = logging.getLogger('cryptodb')
 
     try:

src/worker.py

@@ -26,7 +26,7 @@ def run(self):
 
             if start_timestamp and first_time and start_timestamp < datetime.timestamp(first_time):
                 self.logger.warning(
-                    "Start time befor first timestamp in database")
+                    "Start time before first timestamp in database")
                 self.logger.warning(
                     "Recreating table {}".format(symbol.lower()))
                 self.config.timescaledb.createTable(
@@ -37,7 +37,7 @@ def run(self):
             if last_time:
                 start_time = str(last_time)
             klines_df = self.config.binance.get_klines(
-                symbol, startTime=start_time)
+                symbol, startTime=start_time, endTime=self.config.end_time)
             self.logger.debug(klines_df)
             self.config.timescaledb.write(
                 self.config.database, symbol, klines_df)

tests/conftest.py

@@ -1,4 +1,6 @@
 import pytest
+import requests
+import json
 from os import environ
 import pickle
 import pandas as pd
@@ -7,8 +9,8 @@
 from config import Config
 
 
-@pytest.fixture
-def timescaleDB():
+@pytest.fixture(scope='function')
+def timescaledb():
     return TimescaleDB(
         host=environ.get("TEST_TIMESCALE_HOST", "localhost"),
         port=environ.get("TEST_TIMESCALE_PORT", 5432),
@@ -17,37 +19,70 @@ def timescaleDB():
     )
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def database():
     return environ.get("TEST_DATABASE", "binance")
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def symbol():
     return environ.get("TEST_SYMBOL", "BTCUSDT")
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def binance() -> Binance:
     return Binance()
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def test_data():
     with open('tests/test_data.pkl', 'rb') as file:
         data = pickle.load(file)
     return data
 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def test_dataframe():
     return pd.read_pickle('tests/test_dataframe.pkl')
 
 
-@pytest.fixture
-def config(timescaleDB, database, symbol):
+@pytest.fixture(scope='function')
+def binance_ping_response(responses):
+    responses.add(
+        responses.GET,
+        "https://api.binance.com/api/v3/ping",
+        body="{}",
+        status=200
+    )
+
+
+@pytest.fixture(scope='function')
+def binance_get_klines_response(responses, test_data: list[list]):
+    def request_callback(request: requests.PreparedRequest):
+        start = int(request.params.get("startTime"))
+        end = int(request.params.get("endTime"))
+        limit = int(request.params.get("limit"))
+        response_data = []
+        for row in test_data:
+            if len(response_data) > limit:
+                break
+            if row[0] >= start and row[0] <= end:
+                response_data.append(row)
+        return (200, {}, json.dumps(response_data))
+
+    responses.add_callback(
+        responses.GET,
+        "https://api.binance.com/api/v3/klines",
+        callback=request_callback,
+        content_type="text/plain"
+    )
+
+
+@pytest.fixture(scope='function')
+def config(timescaledb, binance, database, symbol):
     config = Config()
     config.symbols = [symbol]
     config.database = database
-    config.timescaledb = timescaleDB
+    config.timescaledb = timescaledb
+    config.binance = binance
     return config

tests/test_binance.py

@@ -6,3 +6,14 @@ def test_make_dataframe(binance: Binance, test_data: list):
     df = binance.make_dataframe(test_data)
     assert isinstance(df, pd.DataFrame)
     assert any(col in df.columns for col in columns[:, 0])
+
+
+def test_binance_ping(binance: Binance, responses, binance_ping_response):
+    binance.client.ping()
+    assert len(responses.calls) == 2
+
+
+def test_binance_get_klines(binance: Binance, symbol: str, test_dataframe: pd.DataFrame, responses, binance_ping_response, binance_get_klines_response):
+    df = binance.get_klines(symbol, '24.12.2022')
+    assert len(responses.calls) == 6
+    assert df.equals(test_dataframe)

tests/test_config.py

@@ -10,6 +10,7 @@ def environment() -> dict:
         "DEBUG": "true",
         "SYMBOLS": '["BTCUSDT","ETHUSDT"]',
         "START_DATE": "01.01.2021",
+        "END_DATE": "01.02.2021",
         "SLEEP_TIME": "30",
         "TIMESCALE_HOST": "postgres",
         "TIMESCALE_PORT": "1234",
@@ -26,6 +27,7 @@ def assert_config():
         "debug": True,
         "symbols": ["BTCUSDT", "ETHUSDT"],
         "start_time": "01.01.2021",
+        "end_time": "01.02.2021",
         "sleep_time": 30,
         "timescaledb": TimescaleDB(
             host="postgres",

tests/test_timescaledb.py

@@ -2,31 +2,31 @@
 from sinks.timescaledb import TimescaleDB
 
 
-def test_database(timescaleDB: TimescaleDB, database: str, symbol: str, test_dataframe: pd.DataFrame):
-    timescaleDB.dropDatabase(database)
+def test_database(timescaledb: TimescaleDB, database: str, symbol: str, test_dataframe: pd.DataFrame):
+    timescaledb.dropDatabase(database)
 
-    timescaleDB.createDatabase(database)
-    timescaleDB.createDatabase(database, False)
-    timescaleDB.createDatabase(database, True)
+    timescaledb.createDatabase(database)
+    timescaledb.createDatabase(database, False)
+    timescaledb.createDatabase(database, True)
 
-    timescaleDB.createTable(database, symbol, False)
-    timescaleDB.createTable(database, symbol, False)
-    timescaleDB.createTable(database, symbol, True)
-    timescaleDB.dropTable(database, symbol)
-    timescaleDB.createTable(database, symbol, False)
+    timescaledb.createTable(database, symbol, False)
+    timescaledb.createTable(database, symbol, False)
+    timescaledb.createTable(database, symbol, True)
+    timescaledb.dropTable(database, symbol)
+    timescaledb.createTable(database, symbol, False)
 
-    timescaleDB.write(database, symbol, test_dataframe)
-    first_ts = timescaleDB.getFirstTimestamp(database, symbol)
+    timescaledb.write(database, symbol, test_dataframe)
+    first_ts = timescaledb.getFirstTimestamp(database, symbol)
     assert pd.Timestamp(test_dataframe['time'].values[0]).tz_localize(
         'UTC') == pd.to_datetime(first_ts)
 
-    last_ts = timescaleDB.getLastTimestamp(database, symbol)
+    last_ts = timescaledb.getLastTimestamp(database, symbol)
     assert pd.Timestamp(test_dataframe['time'].values[-1]
                         ).tz_localize('UTC') == pd.to_datetime(last_ts)
 
-    timescaleDB.deleteLastTimestamp(database, symbol)
-    last_ts = timescaleDB.getLastTimestamp(database, symbol)
+    timescaledb.deleteLastTimestamp(database, symbol)
+    last_ts = timescaledb.getLastTimestamp(database, symbol)
     assert pd.Timestamp(test_dataframe['time'].values[-2]
                         ).tz_localize('UTC') == pd.to_datetime(last_ts)
 
-    timescaleDB.dropDatabase(database)
+    timescaledb.dropDatabase(database)

tests/test_worker.py

@@ -1,50 +1,33 @@
-from typing import Generator
-import pandas as pd
+from datetime import datetime, timezone
 from worker import Worker
 from config import Config
-from sources import Binance
 
 
-class OfflineBinance(Binance):
-    def __init__(self, data: pd.DataFrame, api_key: str = None, api_secret: str = None):
-        super().__init__(api_key, api_secret)
-        self._data = data
-        self._data_generator = self._make_data_generator()
-        self._next_data = next(self._data_generator, None)
-        self._has_next = True
-
-    def _make_data_generator(self) -> Generator[pd.DataFrame, None, None]:
-        chunk_size = 100
-        pos = 0
-        while pos < len(self._data):
-            yield self._data.iloc[pos:pos+chunk_size]
-            pos = pos + chunk_size
-
-    def get_klines(self, symbol: str, startTime: str, endTime: str = 'NOW', interval: str = '1m') -> pd.DataFrame:
-        data = self._next_data
-        self._next_data = next(self._data_generator, None)
-        if self._next_data is None:
-            self._has_next = False
-        return data
-
-
-def test_worker(config: Config):
-    df = pd.read_pickle('tests/test_dataframe.pkl')
-
-    config.timescaledb.createDatabase(config.database)
+def test_worker(config: Config, responses, binance_ping_response, binance_get_klines_response):
+    config.timescaledb.createDatabase(config.database, True)
 
     # Test with subset
-    config.start_time = "25.12.2021"
-    config.binance = OfflineBinance(data=df[df['time'].dt.day == 25])
+    config.start_time = "25.12.2022"
+    config.end_time = "26.12.2022"
     worker = Worker(config)
-    while config.binance._has_next:
-        worker.run()
+    worker.run()
+
+    assert len(responses.calls) == 4
+    assert config.timescaledb.getFirstTimestamp(
+        config.database, config.symbols[0]) == datetime(2022, 12, 25, tzinfo=timezone.utc)
+    assert config.timescaledb.getLastTimestamp(
+        config.database, config.symbols[0]) == datetime(2022, 12, 26, tzinfo=timezone.utc)
 
     # Use earlier start time and newer data
-    config.start_time = "24.12.2021"
-    config.binance = OfflineBinance(data=df)
+    config.start_time = "24.12.2022"
+    config.end_time = "NOW"
     worker = Worker(config)
-    while config.binance._has_next:
-        worker.run()
+    worker.run()
+
+    assert len(responses.calls) == 10
+    assert config.timescaledb.getFirstTimestamp(
+        config.database, config.symbols[0]) == datetime(2022, 12, 24, tzinfo=timezone.utc)
+    assert config.timescaledb.getLastTimestamp(
+        config.database, config.symbols[0]) == datetime(2022, 12, 26, 17, 38, tzinfo=timezone.utc)
 
     config.timescaledb.dropDatabase(config.database)