From 5a7a60d2995f107a69ea577b4dc38bf4ddddd8c9 Mon Sep 17 00:00:00 2001 From: Djeex Date: Wed, 26 Aug 2026 15:43:40 +0200 Subject: [PATCH] CI/CD hardening: lint, secret scan, coverage gate, auto CVE-fix PRs, GHCR + GitHub mirror publishing (#32) - release changelog: commits rendered as description (link), divider lines dropped - gitleaks secret scan and hadolint on every push/PR - ruff lint/format gate (Python repos) with a pytest --cov-fail-under gate - scheduled CRITICAL Trivy failures attempt an apk upgrade rebuild and open a follow-up PR if it clears the finding, instead of just failing red - images also published to ghcr.io/djeex/ - a matching GitHub Release is created on the GitHub mirror, with a notice pointing back to this repo as the source of truth --- .gitea/workflows/ci.yml | 104 ++++++++++++++++++++++++++++++++++++-- Dockerfile | 9 +++- app/env_config.py | 73 ++++++++++++++------------ app/gpu_checker.py | 35 ++++++++----- app/main.py | 10 ++-- app/notifier.py | 62 ++++++++++++++++------- ruff.toml | 13 +++++ tests/test_env_config.py | 89 +++++++++++++++++++------------- tests/test_gpu_checker.py | 48 ++++++++++++------ tests/test_notifier.py | 48 +++++++++++++----- 10 files changed, 356 insertions(+), 135 deletions(-) create mode 100644 ruff.toml diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index ddb37c5..3c2e5c0 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -18,6 +18,21 @@ jobs: fetch-depth: 0 persist-credentials: false + - name: Scan for secrets + run: | + # docker cp, not a build COPY: a repo's own .dockerignore (e.g. one that + # excludes .git for prod builds) would otherwise silently give an empty, + # falsely-clean scan. + CID=$(docker create zricethezav/gitleaks:v8.30.1 detect --source=/repo --no-banner -v) + docker cp . "$CID:/repo" + docker start -a "$CID" + STATUS=$? + docker rm "$CID" > /dev/null + exit $STATUS + + - name: Lint Dockerfile with hadolint + run: docker run --rm -i hadolint/hadolint:v2.15.1-alpine hadolint --failure-threshold error - < Dockerfile + - name: Build Docker image run: | docker build -t nvidia-stock-bot:ci . 2>&1 | tee build.log @@ -46,7 +61,10 @@ jobs: - name: Run unit tests run: | docker build --target test -t nvidia-stock-bot:test . - docker run --rm nvidia-stock-bot:test pytest -v + docker run --rm nvidia-stock-bot:test pytest -v --cov=. --cov-report=term-missing --cov-fail-under=75 + + - name: Lint with ruff + run: docker build --target lint -t nvidia-stock-bot:lint . - name: Check deprecation warnings run: | @@ -60,12 +78,53 @@ jobs: fi - name: Scan with Trivy (critical - blocking) + id: trivy_critical + continue-on-error: true run: | docker run --rm \ -e DOCKER_HOST=tcp://dockerhost:2375 \ --add-host=dockerhost:host-gateway \ aquasec/trivy:0.74.0 image --exit-code 1 --severity CRITICAL nvidia-stock-bot:ci + - name: Handle CRITICAL findings + if: steps.trivy_critical.outcome == 'failure' + run: | + if [ "${{ github.event_name }}" != "schedule" ]; then + echo "::error::CRITICAL vulnerabilities found, failing the build." + exit 1 + fi + + echo "Scheduled scan found CRITICAL vulnerabilities — attempting an automatic apk upgrade + rescan." + + sed -i '/^FROM .* AS base$/a RUN apk upgrade --no-cache' Dockerfile + docker build -t nvidia-stock-bot:remediated . + + if docker run --rm \ + -e DOCKER_HOST=tcp://dockerhost:2375 \ + --add-host=dockerhost:host-gateway \ + aquasec/trivy:0.74.0 image --exit-code 1 --severity CRITICAL nvidia-stock-bot:remediated; then + echo "apk upgrade clears the CRITICAL finding(s) — opening a PR for review." + + BRANCH="auto/cve-fix-$(date +%Y%m%d)-$(echo "${{ github.sha }}" | cut -c1-7)" + git config user.name "nvidia-stock-bot-ci" + git config user.email "ci@git.djeex.fr" + git checkout -b "$BRANCH" + git add Dockerfile + git commit -m "Auto-remediate CRITICAL CVE via apk upgrade" + git config --unset-all http.https://git.djeex.fr/.extraheader || true + git push "https://Djeex:${{ secrets.CI_PUSH_TOKEN }}@git.djeex.fr/Djeex/nvidia-stock-bot.git" "HEAD:$BRANCH" + + PR_JSON=$(curl -s -X POST \ + -H "Authorization: token ${{ secrets.CI_PUSH_TOKEN }}" \ + -H "Content-Type: application/json" \ + -d "$(jq -n --arg head "$BRANCH" '{title: "🔒 Auto: remediate CRITICAL CVE via apk upgrade", head: $head, base: "main", body: "Opened automatically by the scheduled CVE scan. An `apk upgrade --no-cache` cleared the CRITICAL Trivy finding(s) in a rebuild — review the diff and merge to publish the fix."}')" \ + "https://git.djeex.fr/api/v1/repos/Djeex/nvidia-stock-bot/pulls") + echo "PR API response: $(echo "$PR_JSON" | jq -r '.html_url // .message // "unknown"')" + else + echo "::error::apk upgrade does not clear the CRITICAL finding(s) — no automatic fix available, needs manual review." + exit 1 + fi + - name: Scan with Trivy (high - informative) run: | docker run --rm \ @@ -130,6 +189,16 @@ jobs: docker push "$IMAGE:$MINOR_TAG" docker push "$IMAGE:$VERSION" + GHCR_IMAGE=ghcr.io/djeex/nvidia-stock-bot + echo "${{ secrets.GH_TOKEN }}" | docker login ghcr.io -u Djeex --password-stdin + + docker tag nvidia-stock-bot:ci "$GHCR_IMAGE:latest" + docker tag nvidia-stock-bot:ci "$GHCR_IMAGE:$MINOR_TAG" + docker tag nvidia-stock-bot:ci "$GHCR_IMAGE:$VERSION" + docker push "$GHCR_IMAGE:latest" + docker push "$GHCR_IMAGE:$MINOR_TAG" + docker push "$GHCR_IMAGE:$VERSION" + TRIGGER_MSG=$(git log -1 --format=%s "${{ github.sha }}") PR_NUM=$(echo "$TRIGGER_MSG" | grep -oE '#[0-9]+' | head -1 | tr -d '#' || true) @@ -156,7 +225,7 @@ jobs: fi REPO_URL="https://git.djeex.fr/Djeex/nvidia-stock-bot" - COMMIT_LIST=$(git log --no-merges --format="- [%h](${REPO_URL}/commit/%H) %s" "$BASE_REF".."${{ github.sha }}") + COMMIT_LIST=$(git log --no-merges --format="- %s ([%h](${REPO_URL}/commit/%H))" "$BASE_REF".."${{ github.sha }}") SHORT_SHA=$(echo "${{ github.sha }}" | cut -c1-7) SOURCE_LINE="[${SHORT_SHA}](${REPO_URL}/commit/${{ github.sha }})" @@ -166,10 +235,8 @@ jobs: BODY=$(cat </dev/null || true + fi diff --git a/Dockerfile b/Dockerfile index 58adedc..035c5cd 100644 --- a/Dockerfile +++ b/Dockerfile @@ -11,13 +11,20 @@ RUN pip install --no-cache-dir -r requirements.txt FROM base AS test -RUN pip install --no-cache-dir pytest==9.1.1 +RUN pip install --no-cache-dir pytest==9.1.1 pytest-cov==7.1.0 COPY pytest.ini /app/pytest.ini COPY /tests/ /app/tests/ CMD ["pytest", "-v"] +FROM base AS lint + +RUN pip install --no-cache-dir ruff==0.16.4 +COPY ruff.toml /app/ruff.toml +COPY /tests/ /app/tests/ +RUN ruff check . && ruff format --check . + FROM base RUN addgroup -g 911 nvbot && adduser -D -u 911 -G nvbot nvbot diff --git a/app/env_config.py b/app/env_config.py index c71fbea..23f75c7 100644 --- a/app/env_config.py +++ b/app/env_config.py @@ -1,11 +1,13 @@ +import json +import logging import os import re -import logging -import json import sys # Read version from VERSION file -with open(os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION"), "r", encoding="utf-8") as f: +with open( + os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION"), encoding="utf-8" +) as f: VERSION = f.read().strip() # Logger setup @@ -23,13 +25,13 @@ logging.info("=" * 60) # Env variables try: - DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL'] - DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free') - DISCORD_ROLES = os.environ.get('DISCORD_ROLES') - COUNTRY = os.environ.get('COUNTRY') or 'US' - REFRESH_TIME = int(os.environ.get('REFRESH_TIME') or 30) - TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true' - PRODUCT_NAMES = os.environ['PRODUCT_NAMES'] + DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"] + DISCORD_SERVER_NAME = os.environ.get("DISCORD_SERVER_NAME", "Shared for free") + DISCORD_ROLES = os.environ.get("DISCORD_ROLES") + COUNTRY = os.environ.get("COUNTRY") or "US" + REFRESH_TIME = int(os.environ.get("REFRESH_TIME") or 30) + TEST_MODE = os.environ.get("TEST_MODE", "False").lower() == "true" + PRODUCT_NAMES = os.environ["PRODUCT_NAMES"] # Errors and warning except KeyError as e: @@ -49,32 +51,32 @@ if not DISCORD_WEBHOOK_URL: logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.") sys.exit(1) -PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(',')] +PRODUCT_NAMES = [name.strip() for name in PRODUCT_NAMES.split(",")] # Role mapping DISCORD_ROLE_MAP = {} if not DISCORD_ROLES or not DISCORD_ROLES.strip(): logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.") for name in PRODUCT_NAMES: - DISCORD_ROLE_MAP[name] = '@everyone' + DISCORD_ROLE_MAP[name] = "@everyone" else: - roles = [r.strip() if r.strip() else '@everyone' for r in DISCORD_ROLES.split(',')] + roles = [r.strip() if r.strip() else "@everyone" for r in DISCORD_ROLES.split(",")] if len(roles) != len(PRODUCT_NAMES): logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.") sys.exit(1) for name, role in zip(PRODUCT_NAMES, roles): - if role != '@everyone' and not re.match(r'^<@&\d{17,20}>$', role): + if role != "@everyone" and not re.match(r"^<@&\d{17,20}>$", role): logging.error(f"❌ Invalid DISCORD_ROLE format for {name}: {role}") sys.exit(1) DISCORD_ROLE_MAP[name] = role # Masked webhook in terminal -match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL) +match = re.search(r"/(\d+)/(.*)", DISCORD_WEBHOOK_URL) if match: webhook_id = match.group(1) webhook_token = match.group(2) - masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10 - masked_webhook_token = webhook_token[:len(webhook_token) - 120] + '*' * 10 + masked_webhook_id = webhook_id[: len(webhook_id) - 10] + "*" * 10 + masked_webhook_token = webhook_token[: len(webhook_token) - 120] + "*" * 10 wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}" else: wh_masked_url = "[Invalid webhook URL]" @@ -90,29 +92,33 @@ HEADERS = { "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", - "Sec-Ch-Ua": "\"Google Chrome\";v=\"131\", \"Chromium\";v=\"131\", \"Not.A/Brand\";v=\"24\"", - "Sec-Ch-Ua-Platform": "\"macOS\"", + "Sec-Ch-Ua": '"Google Chrome";v="131", "Chromium";v="131", "Not.A/Brand";v="24"', + "Sec-Ch-Ua-Platform": '"macOS"', "Cache-Control": "no-cache, no-store, must-revalidate", "Pragma": "no-cache", - "Expires": "0" + "Expires": "0", } # Load country setting and localization config country_code = os.environ.get("COUNTRY", "US").upper() try: - with open("localization.json", "r", encoding="utf-8") as f: + with open("localization.json", encoding="utf-8") as f: localization_config = json.load(f) except FileNotFoundError: logging.error("❌ localization.json file not found.") sys.exit(1) # Find country entry -country_entry = next((entry for entry in localization_config if entry["country_code"].upper() == country_code), None) +country_entry = next( + (entry for entry in localization_config if entry["country_code"].upper() == country_code), None +) if not country_entry: logging.warning(f"⚠️ Country '{country_code}' not found in localization.json. Defaulting to US.") - country_entry = next((entry for entry in localization_config if entry["country_code"].upper() == "US"), None) + country_entry = next( + (entry for entry in localization_config if entry["country_code"].upper() == "US"), None + ) if not country_entry: logging.error("❌ US fallback not found in localization.json.") sys.exit(1) @@ -124,7 +130,7 @@ currency = country_entry["currency"] # Load language file try: - with open("languages.json", "r", encoding="utf-8") as f: + with open("languages.json", encoding="utf-8") as f: loc_lang = json.load(f) except FileNotFoundError: logging.error("❌ languages.json file not found.") @@ -141,9 +147,15 @@ if not loc: # Ensure all required keys are present required_keys = [ - "in_stock_title", "out_of_stock_title", "sku_change_title", - "buy_now", "price", "time", "footer", - "sku_description", "imminent_drop" + "in_stock_title", + "out_of_stock_title", + "sku_change_title", + "buy_now", + "price", + "time", + "footer", + "sku_description", + "imminent_drop", ] missing_keys = [key for key in required_keys if key not in loc] fallback = loc_lang.get("en", {}) @@ -159,17 +171,16 @@ for key in missing_keys: locale = full_lang_code.lower() API_URL_SKU = os.getenv( "API_URL_SKU", - f"https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale={locale}&Manufacturer=Nvidia" + f"https://api.nvidia.partners/edge/product/search?page=1&limit=100&locale={locale}&Manufacturer=Nvidia", ) API_URL_STOCK = os.getenv( - "API_URL_STOCK", - f"https://api.store.nvidia.com/partner/v1/feinventory?locale={locale}&skus=" + "API_URL_STOCK", f"https://api.store.nvidia.com/partner/v1/feinventory?locale={locale}&skus=" ) PRODUCT_URL = os.getenv( "PRODUCT_URL", - f"https://marketplace.nvidia.com/{locale}/consumer/graphics-cards/?locale={locale}&page=1&limit=12&manufacturer=NVIDIA" + f"https://marketplace.nvidia.com/{locale}/consumer/graphics-cards/?locale={locale}&page=1&limit=12&manufacturer=NVIDIA", ) # Public constants diff --git a/app/gpu_checker.py b/app/gpu_checker.py index 75713d4..b8fd047 100644 --- a/app/gpu_checker.py +++ b/app/gpu_checker.py @@ -1,21 +1,28 @@ -import requests import logging import time -from env_config import HEADERS, PRODUCT_NAMES, API_URL_SKU, API_URL_STOCK, PRODUCT_URL -from notifier import send_discord_notification, send_out_of_stock_notification, send_sku_change_notification + +import requests from requests.adapters import HTTPAdapter, Retry +from env_config import API_URL_SKU, API_URL_STOCK, HEADERS, PRODUCT_NAMES, PRODUCT_URL +from notifier import ( + send_discord_notification, + send_out_of_stock_notification, + send_sku_change_notification, +) + # HTTP session session = requests.Session() retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) -session.mount('https://', HTTPAdapter(max_retries=retries)) +session.mount("https://", HTTPAdapter(max_retries=retries)) session.headers.update(HEADERS) -# Keeping memory of last run +# Keeping memory of last run last_sku_dict = {} global_stock_status_dict = {} first_run_dict = {name: True for name in PRODUCT_NAMES} + # Stock check function def check_rtx_50_founders(): global last_sku_dict, global_stock_status_dict, first_run_dict @@ -24,7 +31,7 @@ def check_rtx_50_founders(): try: cache_buster = int(time.time() * 1000) sku_url = f"{API_URL_SKU}&_t={cache_buster}" - + response = session.get(sku_url, timeout=10) logging.info(f"SKU API response: {response.status_code}") response.raise_for_status() @@ -32,9 +39,9 @@ def check_rtx_50_founders(): except requests.exceptions.RequestException as e: logging.error(f"SKU API error: {e}") return - + # Checking productSKU and productUPC for all GPU set in PRODUCT_NAME - all_products = data['searchedProducts']['productDetails'] + all_products = data["searchedProducts"]["productDetails"] for product_name in PRODUCT_NAMES: product_details = None @@ -47,8 +54,8 @@ def check_rtx_50_founders(): logging.warning(f"⚠️ No product with GPU '{product_name}' found.") continue - product_sku = product_details['productSKU'] - product_upc = product_details.get('productUPC', "") + product_sku = product_details["productSKU"] + product_upc = product_details.get("productUPC", "") if not isinstance(product_upc, list): product_upc = [product_upc] @@ -60,7 +67,7 @@ def check_rtx_50_founders(): last_sku_dict[product_name] = product_sku first_run_dict[product_name] = False - + # Check product availability in API_URL_STOCK for each SKU cache_buster = int(time.time() * 1000) api_stock_url = f"{API_URL_STOCK}{product_sku}&_t={cache_buster}" @@ -80,8 +87,8 @@ def check_rtx_50_founders(): products_price = "Price not available" if isinstance(products, list) and len(products) > 0: for p in products: - price = p.get("price", 'Price not available') - if price != 'Price not available': + price = p.get("price", "Price not available") + if price != "Price not available": products_price = price break else: @@ -93,7 +100,7 @@ def check_rtx_50_founders(): is_active = p.get("is_active") == "true" if is_active and any(upc.upper() in gpu_name for upc in product_upc): found_in_stock.add(gpu_name) - + # Comparing previous state and notify for upc in product_upc: upc_upper = upc.upper() diff --git a/app/main.py b/app/main.py index eeae05d..7c8d8c8 100644 --- a/app/main.py +++ b/app/main.py @@ -1,17 +1,20 @@ -import time import logging import signal import sys -from gpu_checker import check_rtx_50_founders +import time + from env_config import REFRESH_TIME +from gpu_checker import check_rtx_50_founders + # Signal handler function def handle_exit(signum, frame): logging.info(f"🛑 Received signal {signum}. Exiting gracefully...") sys.exit(0) + # Register signal handlers -signal.signal(signal.SIGINT, handle_exit) # Ctrl+C +signal.signal(signal.SIGINT, handle_exit) # Ctrl+C signal.signal(signal.SIGTERM, handle_exit) # docker stop / kill -15 if __name__ == "__main__": @@ -24,4 +27,3 @@ if __name__ == "__main__": except KeyboardInterrupt: logging.info("🛑 Script interrupted by user (KeyboardInterrupt). Exiting gracefully.") sys.exit(0) - \ No newline at end of file diff --git a/app/notifier.py b/app/notifier.py index ad668c0..e040a14 100644 --- a/app/notifier.py +++ b/app/notifier.py @@ -1,15 +1,29 @@ -import time import logging +import time + import requests + from env_config import ( - DISCORD_WEBHOOK_URL, DISCORD_SERVER_NAME, DISCORD_ROLE_MAP, TEST_MODE, currency, - in_stock_title, out_of_stock_title, sku_change_title, - buy_now, price_label, time_label, footer, sku_description, imminent_drop + DISCORD_ROLE_MAP, + DISCORD_SERVER_NAME, + DISCORD_WEBHOOK_URL, + TEST_MODE, + buy_now, + currency, + footer, + imminent_drop, + in_stock_title, + out_of_stock_title, + price_label, + sku_change_title, + sku_description, + time_label, ) AVATAR = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/ds_wh_pp.jpg" THUMBNAIL = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg" + # In stock def send_discord_notification(gpu_name, product_link, products_price): timestamp = int(time.time()) @@ -24,17 +38,20 @@ def send_discord_notification(gpu_name, product_link, products_price): "author": {"name": "Nvidia Founder Editions"}, "fields": [ {"name": price_label, "value": f"`{currency}{products_price}`", "inline": True}, - {"name": time_label, "value": f" ", "inline": True} + {"name": time_label, "value": f" ", "inline": True}, ], "description": buy_now.format(product_link=product_link), - "footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR} + "footer": { + "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), + "icon_url": AVATAR, + }, } payload = { "content": DISCORD_ROLE_MAP.get(gpu_name, "@everyone"), "username": "NviBot", "avatar_url": AVATAR, - "embeds": [embed] + "embeds": [embed], } try: @@ -46,6 +63,7 @@ def send_discord_notification(gpu_name, product_link, products_price): except Exception as e: logging.error(f"🚨 Error sending webhook: {e}") + # Out of stock def send_out_of_stock_notification(gpu_name, product_link, products_price): timestamp = int(time.time()) @@ -59,15 +77,16 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price): "thumbnail": {"url": THUMBNAIL}, "url": product_link, "author": {"name": "Nvidia Founder Editions"}, - "footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR}, - "fields": [{"name": time_label, "value": f" ", "inline": True}] + "footer": { + "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), + "icon_url": AVATAR, + }, + "fields": [ + {"name": time_label, "value": f" ", "inline": True} + ], } - payload = { - "username": "NviBot", - "avatar_url": AVATAR, - "embeds": [embed] - } + payload = {"username": "NviBot", "avatar_url": AVATAR, "embeds": [embed]} try: response = requests.post(DISCORD_WEBHOOK_URL, json=payload) @@ -78,6 +97,7 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price): except Exception as e: logging.error(f"🚨 Error sending webhook: {e}") + # SKU change def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link): timestamp = int(time.time()) @@ -90,15 +110,20 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link): "url": product_link, "description": sku_description.format(old_sku=old_sku, new_sku=new_sku), "color": 16776960, - "footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR}, - "fields": [{"name": time_label, "value": f" ", "inline": True}] + "footer": { + "text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), + "icon_url": AVATAR, + }, + "fields": [ + {"name": time_label, "value": f" ", "inline": True} + ], } payload = { - "content": imminent_drop.format(DISCORD_ROLE=DISCORD_ROLE_MAP.get(gpu_name, '@everyone')), + "content": imminent_drop.format(DISCORD_ROLE=DISCORD_ROLE_MAP.get(gpu_name, "@everyone")), "username": "NviBot", "avatar_url": AVATAR, - "embeds": [embed] + "embeds": [embed], } try: @@ -109,4 +134,3 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link): logging.error(f"❌ Webhook error: {response.status_code} - {response.text}") except Exception as e: logging.error(f"🚨 Error sending webhook: {e}") - \ No newline at end of file diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 0000000..d78ee69 --- /dev/null +++ b/ruff.toml @@ -0,0 +1,13 @@ +line-length = 100 + +[lint] +select = ["E", "F", "I", "UP", "B"] +# E501: handled by the formatter. B905: zip() without strict= in env_config.py — +# app-logic change, left for the user to decide (see feedback-no-app-logic-changes). +ignore = ["E501", "B905"] + +[lint.isort] +# Pinned explicitly: auto-detection of first-party modules differs between the +# host (app/ subdir + .git present) and the Docker lint stage (flattened to /app, +# no .git) — without this, import-sort results silently diverge between the two. +known-first-party = ["env_config", "gpu_checker", "notifier", "main"] diff --git a/tests/test_env_config.py b/tests/test_env_config.py index c2100a6..45a9c47 100644 --- a/tests/test_env_config.py +++ b/tests/test_env_config.py @@ -38,62 +38,83 @@ def test_missing_product_names_exits(monkeypatch): def test_default_role_map_is_everyone(monkeypatch): - cfg = _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090, RTX 5080", - }) + cfg = _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090, RTX 5080", + }, + ) assert cfg.DISCORD_ROLE_MAP == {"RTX 5090": "@everyone", "RTX 5080": "@everyone"} def test_role_count_mismatch_exits(monkeypatch): with pytest.raises(SystemExit): - _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090, RTX 5080", - "DISCORD_ROLES": "<@&123456789012345678>", - }) + _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090, RTX 5080", + "DISCORD_ROLES": "<@&123456789012345678>", + }, + ) def test_invalid_role_format_exits(monkeypatch): with pytest.raises(SystemExit): - _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090", - "DISCORD_ROLES": "not-a-role", - }) + _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090", + "DISCORD_ROLES": "not-a-role", + }, + ) def test_valid_role_format_accepted(monkeypatch): - cfg = _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090", - "DISCORD_ROLES": "<@&123456789012345678>", - }) + cfg = _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090", + "DISCORD_ROLES": "<@&123456789012345678>", + }, + ) assert cfg.DISCORD_ROLE_MAP["RTX 5090"] == "<@&123456789012345678>" def test_unknown_country_falls_back_to_us(monkeypatch): - cfg = _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090", - "COUNTRY": "ZZ", - }) + cfg = _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090", + "COUNTRY": "ZZ", + }, + ) assert cfg.currency == "$" def test_known_country_currency(monkeypatch): - cfg = _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090", - "COUNTRY": "GB", - }) + cfg = _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090", + "COUNTRY": "GB", + }, + ) assert cfg.currency == "£" def test_refresh_time_invalid_exits(monkeypatch): with pytest.raises(SystemExit): - _reload_env_config(monkeypatch, { - "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, - "PRODUCT_NAMES": "RTX 5090", - "REFRESH_TIME": "not-a-number", - }) + _reload_env_config( + monkeypatch, + { + "DISCORD_WEBHOOK_URL": VALID_WEBHOOK, + "PRODUCT_NAMES": "RTX 5090", + "REFRESH_TIME": "not-a-number", + }, + ) diff --git a/tests/test_gpu_checker.py b/tests/test_gpu_checker.py index 51258b9..eb225de 100644 --- a/tests/test_gpu_checker.py +++ b/tests/test_gpu_checker.py @@ -6,9 +6,7 @@ PRODUCT_NAME = "RTX 5090 Founders Edition" SKU_PAYLOAD = { "searchedProducts": { - "productDetails": [ - {"gpu": PRODUCT_NAME, "productSKU": "SKU-1", "productUPC": "ABC123"} - ] + "productDetails": [{"gpu": PRODUCT_NAME, "productSKU": "SKU-1", "productUPC": "ABC123"}] } } @@ -52,9 +50,15 @@ def _queue_responses(monkeypatch, checker, *payloads): def test_transition_to_in_stock_sends_notification(monkeypatch): checker = _import_gpu_checker(monkeypatch) calls = [] - monkeypatch.setattr(checker, "send_discord_notification", lambda *a: calls.append(("in_stock", a))) - monkeypatch.setattr(checker, "send_out_of_stock_notification", lambda *a: calls.append(("out_of_stock", a))) - monkeypatch.setattr(checker, "send_sku_change_notification", lambda *a: calls.append(("sku_change", a))) + monkeypatch.setattr( + checker, "send_discord_notification", lambda *a: calls.append(("in_stock", a)) + ) + monkeypatch.setattr( + checker, "send_out_of_stock_notification", lambda *a: calls.append(("out_of_stock", a)) + ) + monkeypatch.setattr( + checker, "send_sku_change_notification", lambda *a: calls.append(("sku_change", a)) + ) _queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True)) checker.check_rtx_50_founders() @@ -67,9 +71,15 @@ def test_transition_to_in_stock_sends_notification(monkeypatch): def test_transition_to_out_of_stock_sends_notification(monkeypatch): checker = _import_gpu_checker(monkeypatch) calls = [] - monkeypatch.setattr(checker, "send_discord_notification", lambda *a: calls.append(("in_stock", a))) - monkeypatch.setattr(checker, "send_out_of_stock_notification", lambda *a: calls.append(("out_of_stock", a))) - monkeypatch.setattr(checker, "send_sku_change_notification", lambda *a: calls.append(("sku_change", a))) + monkeypatch.setattr( + checker, "send_discord_notification", lambda *a: calls.append(("in_stock", a)) + ) + monkeypatch.setattr( + checker, "send_out_of_stock_notification", lambda *a: calls.append(("out_of_stock", a)) + ) + monkeypatch.setattr( + checker, "send_sku_change_notification", lambda *a: calls.append(("sku_change", a)) + ) _queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True)) checker.check_rtx_50_founders() @@ -84,9 +94,15 @@ def test_transition_to_out_of_stock_sends_notification(monkeypatch): def test_no_duplicate_notification_while_still_in_stock(monkeypatch): checker = _import_gpu_checker(monkeypatch) calls = [] - monkeypatch.setattr(checker, "send_discord_notification", lambda *a: calls.append(("in_stock", a))) - monkeypatch.setattr(checker, "send_out_of_stock_notification", lambda *a: calls.append(("out_of_stock", a))) - monkeypatch.setattr(checker, "send_sku_change_notification", lambda *a: calls.append(("sku_change", a))) + monkeypatch.setattr( + checker, "send_discord_notification", lambda *a: calls.append(("in_stock", a)) + ) + monkeypatch.setattr( + checker, "send_out_of_stock_notification", lambda *a: calls.append(("out_of_stock", a)) + ) + monkeypatch.setattr( + checker, "send_sku_change_notification", lambda *a: calls.append(("sku_change", a)) + ) _queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True)) checker.check_rtx_50_founders() @@ -101,7 +117,9 @@ def test_sku_change_triggers_notification_after_first_run(monkeypatch): sku_change_calls = [] monkeypatch.setattr(checker, "send_discord_notification", lambda *a: None) monkeypatch.setattr(checker, "send_out_of_stock_notification", lambda *a: None) - monkeypatch.setattr(checker, "send_sku_change_notification", lambda *a: sku_change_calls.append(a)) + monkeypatch.setattr( + checker, "send_sku_change_notification", lambda *a: sku_change_calls.append(a) + ) _queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(False)) checker.check_rtx_50_founders() @@ -109,9 +127,7 @@ def test_sku_change_triggers_notification_after_first_run(monkeypatch): changed_payload = { "searchedProducts": { - "productDetails": [ - {"gpu": PRODUCT_NAME, "productSKU": "SKU-2", "productUPC": "ABC123"} - ] + "productDetails": [{"gpu": PRODUCT_NAME, "productSKU": "SKU-2", "productUPC": "ABC123"}] } } _queue_responses(monkeypatch, checker, changed_payload, _stock_payload(False)) diff --git a/tests/test_notifier.py b/tests/test_notifier.py index e8fdf58..49f12ad 100644 --- a/tests/test_notifier.py +++ b/tests/test_notifier.py @@ -44,7 +44,9 @@ def test_in_stock_notification_posts_expected_payload(monkeypatch): monkeypatch.setattr(notifier.requests, "post", fake_post) - notifier.send_discord_notification("RTX 5090 Founders Edition", "https://example.com/buy", "1999") + notifier.send_discord_notification( + "RTX 5090 Founders Edition", "https://example.com/buy", "1999" + ) assert captured["url"] == notifier.DISCORD_WEBHOOK_URL assert captured["json"]["content"] == "@everyone" @@ -54,7 +56,9 @@ def test_in_stock_notification_posts_expected_payload(monkeypatch): def test_discord_notification_survives_http_error(monkeypatch): notifier = _import_notifier(monkeypatch, test_mode="False") - monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=500, text="boom")) + monkeypatch.setattr( + notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=500, text="boom") + ) notifier.send_discord_notification("RTX 5090 Founders Edition", "https://example.com", "1999") @@ -76,7 +80,9 @@ def test_out_of_stock_test_mode_skips_network_call(monkeypatch): calls = [] monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: calls.append((a, k))) - notifier.send_out_of_stock_notification("RTX 5090 Founders Edition", "https://example.com", "1999") + notifier.send_out_of_stock_notification( + "RTX 5090 Founders Edition", "https://example.com", "1999" + ) assert calls == [] @@ -91,17 +97,23 @@ def test_out_of_stock_notification_posts_on_success(monkeypatch): monkeypatch.setattr(notifier.requests, "post", fake_post) - notifier.send_out_of_stock_notification("RTX 5090 Founders Edition", "https://example.com/buy", "1999") + notifier.send_out_of_stock_notification( + "RTX 5090 Founders Edition", "https://example.com/buy", "1999" + ) assert captured["json"]["embeds"][0]["url"] == "https://example.com/buy" def test_out_of_stock_notification_survives_http_error(monkeypatch): notifier = _import_notifier(monkeypatch, test_mode="False") - monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=500, text="boom")) + monkeypatch.setattr( + notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=500, text="boom") + ) # Should not raise even though the webhook call "fails" - notifier.send_out_of_stock_notification("RTX 5090 Founders Edition", "https://example.com", "1999") + notifier.send_out_of_stock_notification( + "RTX 5090 Founders Edition", "https://example.com", "1999" + ) def test_out_of_stock_notification_survives_connection_error(monkeypatch): @@ -112,7 +124,9 @@ def test_out_of_stock_notification_survives_connection_error(monkeypatch): monkeypatch.setattr(notifier.requests, "post", raise_error) - notifier.send_out_of_stock_notification("RTX 5090 Founders Edition", "https://example.com", "1999") + notifier.send_out_of_stock_notification( + "RTX 5090 Founders Edition", "https://example.com", "1999" + ) def test_sku_change_test_mode_skips_network_call(monkeypatch): @@ -120,16 +134,22 @@ def test_sku_change_test_mode_skips_network_call(monkeypatch): calls = [] monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: calls.append((a, k))) - notifier.send_sku_change_notification("RTX 5090 Founders Edition", "old-sku", "new-sku", "https://example.com") + notifier.send_sku_change_notification( + "RTX 5090 Founders Edition", "old-sku", "new-sku", "https://example.com" + ) assert calls == [] def test_sku_change_notification_survives_http_error(monkeypatch): notifier = _import_notifier(monkeypatch, test_mode="False") - monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=500, text="boom")) + monkeypatch.setattr( + notifier.requests, "post", lambda *a, **k: FakeResponse(status_code=500, text="boom") + ) - notifier.send_sku_change_notification("RTX 5090 Founders Edition", "old-sku", "new-sku", "https://example.com") + notifier.send_sku_change_notification( + "RTX 5090 Founders Edition", "old-sku", "new-sku", "https://example.com" + ) def test_sku_change_notification_survives_connection_error(monkeypatch): @@ -140,11 +160,15 @@ def test_sku_change_notification_survives_connection_error(monkeypatch): monkeypatch.setattr(notifier.requests, "post", raise_error) - notifier.send_sku_change_notification("RTX 5090 Founders Edition", "old-sku", "new-sku", "https://example.com") + notifier.send_sku_change_notification( + "RTX 5090 Founders Edition", "old-sku", "new-sku", "https://example.com" + ) def test_sku_change_notification_mentions_role_and_skus(monkeypatch): - notifier = _import_notifier(monkeypatch, test_mode="False", discord_roles="<@&123456789012345678>") + notifier = _import_notifier( + monkeypatch, test_mode="False", discord_roles="<@&123456789012345678>" + ) captured = {} def fake_post(url, json=None, **kwargs):