You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
66 lines
2.2 KiB
66 lines
2.2 KiB
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
# Add current directory to path so we can import adapters
|
|
sys.path.append(os.getcwd())
|
|
|
|
from adapters.ccxt_shim.breeze_ccxt import BreezeCCXT
|
|
from freqtrade.exceptions import OperationalException
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("smoke_auth")
|
|
|
|
|
|
def run_smoke_auth():
|
|
logger.info("Starting Auth & Rate Limiter Smoke Test")
|
|
|
|
# Check if credentials are present (mocked for this test)
|
|
api_key = os.environ.get("BREEZE_API_KEY")
|
|
api_secret = os.environ.get("BREEZE_API_SECRET")
|
|
session_token = os.environ.get("BREEZE_SESSION_TOKEN")
|
|
|
|
if not all([api_key, api_secret, session_token]):
|
|
logger.error(
|
|
"FAILED: Missing BREEZE_API_KEY, BREEZE_API_SECRET, or BREEZE_SESSION_TOKEN in environment"
|
|
)
|
|
return
|
|
|
|
# 1. Verify Session Initialization
|
|
try:
|
|
exchange = BreezeCCXT({})
|
|
if exchange.breeze is not None:
|
|
logger.info("PASSED: Breeze object initialized via ENV")
|
|
# In a real scenario we'd check if generate_session was called,
|
|
# but since we can't easily mock the SDK's internal state here without more complexity,
|
|
# we assume the log message from BreezeCCXT.__init__ is correct.
|
|
print("AUTH OK")
|
|
else:
|
|
logger.error("FAILED: Breeze object NOT initialized")
|
|
return
|
|
except Exception:
|
|
logger.exception("P19: FAILED: Initialization error")
|
|
return
|
|
|
|
# 2. Test Rate Limiter (100/min)
|
|
logger.info("Testing Rate Limiter (firing >100 requests)")
|
|
try:
|
|
for i in range(101):
|
|
# simulate 101 requests quickly
|
|
exchange.rate_limiter.check_and_record()
|
|
logger.error("FAILED: Rate limiter did NOT block 101st request")
|
|
except OperationalException as e:
|
|
if "100/min" in str(e):
|
|
logger.info(f"PASSED: Rate limiter blocked 101st request: {e}")
|
|
else:
|
|
logger.error(f"FAILED: Unexpected OperationalException: {e}")
|
|
except Exception:
|
|
logger.exception("P19: FAILED: Unexpected error in Rate Limiter test")
|
|
|
|
logger.info("SMOKE TEST FINISHED SUCCESSFULLY")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_smoke_auth()
|