6 Commits
Author SHA1 Message Date
gitea-ci-bot 753750530d chore: bump version to 4.0.4 [skip ci] 2026-08-22 21:14:15 +00:00
Djeex f4f41abbe8 Merge pull request 'Add Gitea Actions CI/CD pipeline' (#27) from dev-ci into main
Reviewed-on: #27
2026-08-22 23:13:45 +02:00
Djeex 7fb3663097 Cover notifier.py error paths (TEST_MODE, HTTP error, connection error)
Brings app/notifier.py from 73% to 100% test coverage by exercising the
three notification functions' TEST_MODE early-return, non-204 response,
and requests exception branches — the exact paths a requests upgrade
regression would break silently.
2026-08-22 23:08:33 +02:00
Djeex 530aeba667 Add Gitea Actions CI/CD pipeline
Bring nvidia-stock-bot up to the same CI/CD maturity as adguard-cidre:
pytest suite (19 tests), multi-stage Dockerfile with a dedicated test
stage, pinned requests dependency, Gitea Actions workflow (build, smoke
test, unit tests, deprecation check, Trivy critical/high scan, version
bump + registry publish + release), and a Renovate config for automated
dependency updates.
2026-08-22 22:54:18 +02:00
Djeex 10290fd28d Merge pull request 'Better versionning' (#25) from fixed-cache into main
Reviewed-on: #25
2025-12-05 16:55:42 +01:00
Djeex 78ee9ac9da Merge pull request 'fixed cache issues' (#24) from fixed-cache into main
Reviewed-on: #24
2025-12-05 16:26:57 +01:00
13 changed files with 642 additions and 75 deletions
+147
View File
@@ -0,0 +1,147 @@
name: CI
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
build-and-scan:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v7
with:
fetch-depth: 0
persist-credentials: false
- name: Build image
run: docker build -t nvidia-stock-bot:ci .
- name: Smoke test (syntax check)
run: |
docker run --rm --entrypoint python nvidia-stock-bot:ci -c "
import ast, glob
for f in glob.glob('/app/*.py'):
ast.parse(open(f).read(), filename=f)
print('syntax OK')
"
- name: Build test image
run: docker build --target test -t nvidia-stock-bot:test .
- name: Run unit tests
run: docker run --rm nvidia-stock-bot:test pytest -v
- name: Check deprecation warnings
run: |
docker run --rm --entrypoint python \
-e DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/123456789012345678/abcdef" \
-e PRODUCT_NAMES="RTX 5090 Founders Edition" \
-e TEST_MODE=True \
nvidia-stock-bot:ci -W error::DeprecationWarning -c "import main"
- name: Trivy scan (critical, blocking)
run: |
docker run --rm \
-e DOCKER_HOST=tcp://dockerhost:2375 --add-host=dockerhost:host-gateway \
aquasec/trivy image --severity CRITICAL --exit-code 1 --ignore-unfixed nvidia-stock-bot:ci
- name: Trivy scan (high, informative)
run: |
docker run --rm \
-e DOCKER_HOST=tcp://dockerhost:2375 --add-host=dockerhost:host-gateway \
aquasec/trivy image --severity HIGH --exit-code 0 --ignore-unfixed nvidia-stock-bot:ci
- name: Version bump, publish image and create release
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
env:
CI_PUSH_TOKEN: ${{ secrets.CI_PUSH_TOKEN }}
REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }}
run: |
IMAGE="git.djeex.fr/djeex/nvidia-stock-bot"
REPO_API="https://git.djeex.fr/api/v1/repos/Djeex/nvidia-stock-bot"
REPO_URL="https://git.djeex.fr/Djeex/nvidia-stock-bot"
# actions/checkout with persist-credentials:false shouldn't leave a
# credential header behind, but unset it defensively anyway so it
# can never silently shadow the token used for the push below.
git config --unset-all "http.${REPO_URL}.git/.extraheader" || true
git config --unset-all "http.${REPO_URL}/.extraheader" || true
BASE_SHA="${{ github.event.before }}"
if [ -z "$BASE_SHA" ] || ! git cat-file -e "$BASE_SHA" 2>/dev/null; then
BASE_SHA="HEAD~1"
fi
CHANGED_FILES=$(git diff --name-only "$BASE_SHA" "${{ github.sha }}" 2>/dev/null || git diff --name-only HEAD~1 HEAD 2>/dev/null || true)
echo "Changed files:"
echo "$CHANGED_FILES"
RELEVANT=false
if echo "$CHANGED_FILES" | grep -qE '^(Dockerfile|VERSION)$|^app/'; then
RELEVANT=true
fi
VERSION_CHANGED=false
if echo "$CHANGED_FILES" | grep -qx "VERSION"; then
VERSION_CHANGED=true
fi
CURRENT_VERSION=$(cat VERSION)
if [ "$RELEVANT" = "true" ] && [ "$VERSION_CHANGED" = "false" ]; then
IFS='.' read -r MAJOR MINOR PATCH <<< "$CURRENT_VERSION"
PATCH=$((PATCH + 1))
NEW_VERSION="${MAJOR}.${MINOR}.${PATCH}"
echo "$NEW_VERSION" > VERSION
git config user.name "gitea-ci-bot"
git config user.email "[email protected]"
git add VERSION
git commit -m "chore: bump version to ${NEW_VERSION} [skip ci]"
git push "https://djeex:${CI_PUSH_TOKEN}@git.djeex.fr/Djeex/nvidia-stock-bot.git" "HEAD:main"
else
NEW_VERSION="$CURRENT_VERSION"
fi
IFS='.' read -r MAJOR MINOR PATCH <<< "$NEW_VERSION"
docker tag nvidia-stock-bot:ci "${IMAGE}:latest"
docker tag nvidia-stock-bot:ci "${IMAGE}:${MAJOR}.${MINOR}"
docker tag nvidia-stock-bot:ci "${IMAGE}:${NEW_VERSION}"
echo "$REGISTRY_TOKEN" | docker login git.djeex.fr -u djeex --password-stdin
docker push "${IMAGE}:latest"
docker push "${IMAGE}:${MAJOR}.${MINOR}"
docker push "${IMAGE}:${NEW_VERSION}"
PR_NUMBER=$(git log -1 --format=%B | grep -oE '#[0-9]+' | head -1 | tr -d '#' || true)
PR_TITLE=""
LABELS=""
if [ -n "$PR_NUMBER" ]; then
PR_JSON=$(curl -s -H "Authorization: token ${CI_PUSH_TOKEN}" "${REPO_API}/pulls/${PR_NUMBER}")
PR_TITLE=$(echo "$PR_JSON" | jq -r '.title // empty')
LABELS=$(echo "$PR_JSON" | jq -r '[.labels[]?.name] | join(",")')
fi
CATEGORY="🔧 Maintenance"
if echo "$LABELS" | grep -qw "bug"; then
CATEGORY="⚠️ Hotfix"
elif echo "$LABELS" | grep -qw "major"; then
CATEGORY="💥 Breaking change"
elif echo "$LABELS" | grep -qw "minor"; then
CATEGORY="✨ Update"
fi
BODY="### ${CATEGORY}"$'\n\n'
if [ -n "$PR_NUMBER" ]; then
BODY="${BODY}**PR:** [#${PR_NUMBER}](${REPO_URL}/pulls/${PR_NUMBER}) — ${PR_TITLE}"$'\n'
fi
BODY="${BODY}**Commit:** [${GITHUB_SHA:0:7}](${REPO_URL}/commit/${GITHUB_SHA})"
curl -s -X POST \
-H "Authorization: token ${CI_PUSH_TOKEN}" \
-H "Content-Type: application/json" \
-d "$(jq -n --arg tag "$NEW_VERSION" --arg name "$NEW_VERSION" --arg body "$BODY" '{tag_name: $tag, name: $name, body: $body, draft: false, prerelease: false}')" \
"${REPO_API}/releases"
+3 -1
View File
@@ -1,2 +1,4 @@
.venv
__pycache__/
__pycache__/
.coverage
.pytest_cache/
+13 -2
View File
@@ -1,4 +1,4 @@
FROM python:3.13-alpine
FROM python:3.13.15-alpine AS base
RUN apk add --no-cache ca-certificates
@@ -9,4 +9,15 @@ COPY /app/ /app/
RUN pip install --no-cache-dir -r requirements.txt
CMD ["python", "main.py"]
FROM base AS test
RUN pip install --no-cache-dir pytest==9.1.1
COPY pytest.ini /app/pytest.ini
COPY /tests/ /app/tests/
CMD ["pytest", "-v"]
FROM base
CMD ["python", "main.py"]
+1 -1
View File
@@ -1 +1 @@
4.0.3
4.0.4
+13 -14
View File
@@ -79,23 +79,22 @@ if match:
else:
wh_masked_url = "[Invalid webhook URL]"
# HTTP headers - Firefox working set
# HTTP headers
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Cache-Control": "no-cache",
"Pragma": "no-cache",
"Sec-GPC": "1",
"Referer": "https://partners.nvidia.com/",
"Origin": "https://partners.nvidia.com",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Priority": "u=0, i",
"TE": "trailers"
"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\"",
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
"Expires": "0"
}
# Load country setting and localization config
+9 -56
View File
@@ -1,16 +1,14 @@
import requests
import logging
import time
import random
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
from requests.adapters import HTTPAdapter, Retry
# HTTP session with stealth configuration
# HTTP session
session = requests.Session()
retries = Retry(total=2, backoff_factor=3, status_forcelist=[500, 502, 503, 504, 429])
adapter = HTTPAdapter(max_retries=retries, pool_connections=1, pool_maxsize=1)
session.mount('https://', adapter)
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))
session.headers.update(HEADERS)
# Keeping memory of last run
@@ -22,53 +20,15 @@ first_run_dict = {name: True for name in PRODUCT_NAMES}
def check_rtx_50_founders():
global last_sku_dict, global_stock_status_dict, first_run_dict
# First get Akamai cookie by visiting main site
try:
logging.info("Getting Akamai protection cookie...")
session.get("https://marketplace.nvidia.com/fr-fr/consumer/graphics-cards/", timeout=10)
time.sleep(1) # Let the session establish
except Exception as e:
logging.warning(f"Failed to get initial cookie: {e}")
# Fetching nvidia API data
try:
sku_url = API_URL_SKU
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}")
if response.status_code == 429:
logging.warning("Rate limited, waiting longer...")
time.sleep(random.uniform(10, 20))
return
response.raise_for_status()
# Debug response content
logging.info(f"Content-Type: {response.headers.get('Content-Type')}")
logging.info(f"Content-Length: {response.headers.get('Content-Length')}")
logging.info(f"Response text length: {len(response.text)}")
logging.info(f"Response content (first 300 chars): {response.text[:300]}")
# Check if content looks like JSON
if not response.text.strip().startswith('{'):
logging.error("Response doesn't start with '{' - not JSON!")
logging.error(f"Full response: {response.text}")
return
try:
data = response.json()
except Exception as e:
logging.error(f"JSON decode error: {e}")
logging.error(f"Full response text: {response.text}")
return
except requests.exceptions.ReadTimeout:
logging.error("Read timeout - IP may be rate limited/blocked. Try changing IP or wait several hours.")
return
except requests.exceptions.ConnectionError as e:
if "Failed to resolve" in str(e):
logging.error("DNS resolution failed - IP may be DNS blacklisted. Try VPN or different DNS servers.")
else:
logging.error(f"Connection error: {e}")
return
data = response.json()
except requests.exceptions.RequestException as e:
logging.error(f"SKU API error: {e}")
return
@@ -79,17 +39,12 @@ def check_rtx_50_founders():
for product_name in PRODUCT_NAMES:
product_details = None
for p in all_products:
gpu_name = p.get("gpu", "").strip()
# Flexible matching: exact match or partial match
if gpu_name == product_name or product_name in gpu_name:
if p.get("gpu", "").strip() == product_name:
product_details = p
break
if not product_details:
logging.warning(f"⚠️ No product with GPU '{product_name}' found.")
# Debug: show available GPU names for troubleshooting
available_gpus = set(p.get("gpu", "") for p in all_products if p.get("gpu"))
logging.info(f"Available GPUs: {sorted(list(available_gpus))[:10]}") # Show first 10
continue
product_sku = product_details['productSKU']
@@ -107,15 +62,13 @@ def check_rtx_50_founders():
first_run_dict[product_name] = False
# Check product availability in API_URL_STOCK for each SKU
api_stock_url = f"{API_URL_STOCK}{product_sku}"
cache_buster = int(time.time() * 1000)
api_stock_url = f"{API_URL_STOCK}{product_sku}&_t={cache_buster}"
logging.info(f"[{product_name}] Checking stock: {api_stock_url}")
try:
response = session.get(api_stock_url, timeout=10)
logging.info(f"[{product_name}] Stock API response: {response.status_code}")
if response.status_code == 429:
logging.warning(f"[{product_name}] Rate limited, skipping...")
continue
response.raise_for_status()
stock_data = response.json()
except requests.exceptions.RequestException as e:
+1 -1
View File
@@ -1 +1 @@
requests
requests==2.34.2
+1
View File
@@ -0,0 +1 @@
[pytest]
+29
View File
@@ -0,0 +1,29 @@
{
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
"extends": ["config:recommended"],
"timezone": "Europe/Paris",
"labels": ["bot"],
"vulnerabilityAlerts": {
"addLabels": ["bug"]
},
"packageRules": [
{
"matchManagers": ["pip_requirements"],
"matchUpdateTypes": ["patch", "minor"],
"automerge": true
},
{
"matchManagers": ["dockerfile"],
"matchUpdateTypes": ["patch"],
"automerge": true
},
{
"matchUpdateTypes": ["major"],
"addLabels": ["major"]
},
{
"matchUpdateTypes": ["minor"],
"addLabels": ["minor"]
}
]
}
+17
View File
@@ -0,0 +1,17 @@
import os
import sys
_TESTS_DIR = os.path.dirname(os.path.abspath(__file__))
_REPO_ROOT = os.path.dirname(_TESTS_DIR)
# app/*.py open their JSON config files with plain relative paths, so the
# process cwd must be the directory the app modules live in — matches how
# the Dockerfile runs them (WORKDIR /app).
_APP_DIR = os.path.join(_REPO_ROOT, "app")
if not os.path.isfile(os.path.join(_APP_DIR, "gpu_checker.py")):
_APP_DIR = _REPO_ROOT # container test stage: code copied flat next to tests/
if _APP_DIR not in sys.path:
sys.path.insert(0, _APP_DIR)
os.chdir(_APP_DIR)
+99
View File
@@ -0,0 +1,99 @@
import importlib
import sys
import pytest
MODULE_NAME = "env_config"
VALID_WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/abcdef"
ENV_KEYS = [
"DISCORD_WEBHOOK_URL",
"DISCORD_SERVER_NAME",
"DISCORD_ROLES",
"COUNTRY",
"REFRESH_TIME",
"TEST_MODE",
"PRODUCT_NAMES",
]
def _reload_env_config(monkeypatch, env):
for key in ENV_KEYS:
monkeypatch.delenv(key, raising=False)
for key, value in env.items():
monkeypatch.setenv(key, value)
sys.modules.pop(MODULE_NAME, None)
return importlib.import_module(MODULE_NAME)
def test_missing_webhook_exits(monkeypatch):
with pytest.raises(SystemExit):
_reload_env_config(monkeypatch, {"PRODUCT_NAMES": "RTX 5090"})
def test_missing_product_names_exits(monkeypatch):
with pytest.raises(SystemExit):
_reload_env_config(monkeypatch, {"DISCORD_WEBHOOK_URL": VALID_WEBHOOK})
def test_default_role_map_is_everyone(monkeypatch):
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>",
})
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",
})
def test_valid_role_format_accepted(monkeypatch):
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",
})
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",
})
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",
})
+146
View File
@@ -0,0 +1,146 @@
import importlib
import sys
VALID_WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/abcdef"
PRODUCT_NAME = "RTX 5090 Founders Edition"
SKU_PAYLOAD = {
"searchedProducts": {
"productDetails": [
{"gpu": PRODUCT_NAME, "productSKU": "SKU-1", "productUPC": "ABC123"}
]
}
}
class FakeResponse:
def __init__(self, payload, status_code=200):
self._payload = payload
self.status_code = status_code
def raise_for_status(self):
if self.status_code >= 400:
raise Exception(f"HTTP {self.status_code}")
def json(self):
return self._payload
def _stock_payload(in_stock, price="1999"):
return {
"listMap": [
{"fe_sku": "ABC123", "is_active": "true" if in_stock else "false", "price": price}
]
}
def _import_gpu_checker(monkeypatch):
monkeypatch.setenv("DISCORD_WEBHOOK_URL", VALID_WEBHOOK)
monkeypatch.setenv("PRODUCT_NAMES", PRODUCT_NAME)
monkeypatch.setenv("TEST_MODE", "True")
monkeypatch.delenv("DISCORD_ROLES", raising=False)
for mod in ("env_config", "notifier", "gpu_checker"):
sys.modules.pop(mod, None)
return importlib.import_module("gpu_checker")
def _queue_responses(monkeypatch, checker, *payloads):
responses = [FakeResponse(p) for p in payloads]
monkeypatch.setattr(checker.session, "get", lambda *a, **k: responses.pop(0))
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)))
_queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True))
checker.check_rtx_50_founders()
assert len(calls) == 1
assert calls[0][0] == "in_stock"
assert calls[0][1][0] == PRODUCT_NAME
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)))
_queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True))
checker.check_rtx_50_founders()
_queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(False))
checker.check_rtx_50_founders()
assert calls[-1][0] == "out_of_stock"
assert calls[-1][1][0] == PRODUCT_NAME
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)))
_queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True))
checker.check_rtx_50_founders()
_queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(True))
checker.check_rtx_50_founders()
assert len(calls) == 1
def test_sku_change_triggers_notification_after_first_run(monkeypatch):
checker = _import_gpu_checker(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))
_queue_responses(monkeypatch, checker, SKU_PAYLOAD, _stock_payload(False))
checker.check_rtx_50_founders()
assert sku_change_calls == [] # first run must never fire a "change" notification
changed_payload = {
"searchedProducts": {
"productDetails": [
{"gpu": PRODUCT_NAME, "productSKU": "SKU-2", "productUPC": "ABC123"}
]
}
}
_queue_responses(monkeypatch, checker, changed_payload, _stock_payload(False))
checker.check_rtx_50_founders()
assert len(sku_change_calls) == 1
assert sku_change_calls[0][1] == "SKU-1"
assert sku_change_calls[0][2] == "SKU-2"
def test_missing_product_in_api_is_skipped_gracefully(monkeypatch):
checker = _import_gpu_checker(monkeypatch)
calls = []
monkeypatch.setattr(checker, "send_discord_notification", lambda *a: calls.append(a))
empty_payload = {"searchedProducts": {"productDetails": []}}
monkeypatch.setattr(checker.session, "get", lambda *a, **k: FakeResponse(empty_payload))
checker.check_rtx_50_founders() # must not raise, just log a warning and skip
assert calls == []
def test_sku_api_error_is_handled_without_raising(monkeypatch):
checker = _import_gpu_checker(monkeypatch)
def raise_error(*a, **k):
raise checker.requests.exceptions.ConnectionError("boom")
monkeypatch.setattr(checker.session, "get", raise_error)
checker.check_rtx_50_founders() # must not propagate the network error
+163
View File
@@ -0,0 +1,163 @@
import importlib
import sys
VALID_WEBHOOK = "https://discord.com/api/webhooks/123456789012345678/abcdef"
def _import_notifier(monkeypatch, test_mode="False", discord_roles=None):
monkeypatch.setenv("DISCORD_WEBHOOK_URL", VALID_WEBHOOK)
monkeypatch.setenv("PRODUCT_NAMES", "RTX 5090 Founders Edition")
monkeypatch.setenv("TEST_MODE", test_mode)
if discord_roles is not None:
monkeypatch.setenv("DISCORD_ROLES", discord_roles)
else:
monkeypatch.delenv("DISCORD_ROLES", raising=False)
for mod in ("env_config", "notifier"):
sys.modules.pop(mod, None)
return importlib.import_module("notifier")
class FakeResponse:
def __init__(self, status_code=204, text=""):
self.status_code = status_code
self.text = text
def test_test_mode_skips_network_call(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="True")
calls = []
monkeypatch.setattr(notifier.requests, "post", lambda *a, **k: calls.append((a, k)))
notifier.send_discord_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
assert calls == []
def test_in_stock_notification_posts_expected_payload(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="False")
captured = {}
def fake_post(url, json=None, **kwargs):
captured["url"] = url
captured["json"] = json
return FakeResponse()
monkeypatch.setattr(notifier.requests, "post", fake_post)
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"
embed = captured["json"]["embeds"][0]
assert "RTX 5090 Founders Edition" in embed["title"]
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"))
notifier.send_discord_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
def test_discord_notification_survives_connection_error(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="False")
def raise_error(*a, **k):
raise notifier.requests.exceptions.ConnectionError("boom")
monkeypatch.setattr(notifier.requests, "post", raise_error)
# Should not raise even though the request itself blew up (network down, DNS, etc.)
notifier.send_discord_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
def test_out_of_stock_test_mode_skips_network_call(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="True")
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")
assert calls == []
def test_out_of_stock_notification_posts_on_success(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="False")
captured = {}
def fake_post(url, json=None, **kwargs):
captured["json"] = json
return FakeResponse(status_code=204)
monkeypatch.setattr(notifier.requests, "post", fake_post)
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"))
# Should not raise even though the webhook call "fails"
notifier.send_out_of_stock_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
def test_out_of_stock_notification_survives_connection_error(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="False")
def raise_error(*a, **k):
raise notifier.requests.exceptions.ConnectionError("boom")
monkeypatch.setattr(notifier.requests, "post", raise_error)
notifier.send_out_of_stock_notification("RTX 5090 Founders Edition", "https://example.com", "1999")
def test_sku_change_test_mode_skips_network_call(monkeypatch):
notifier = _import_notifier(monkeypatch, test_mode="True")
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")
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"))
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):
notifier = _import_notifier(monkeypatch, test_mode="False")
def raise_error(*a, **k):
raise notifier.requests.exceptions.ConnectionError("boom")
monkeypatch.setattr(notifier.requests, "post", raise_error)
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>")
captured = {}
def fake_post(url, json=None, **kwargs):
captured["json"] = json
return FakeResponse()
monkeypatch.setattr(notifier.requests, "post", fake_post)
notifier.send_sku_change_notification(
"RTX 5090 Founders Edition", "old-sku-123", "new-sku-456", "https://example.com"
)
assert "<@&123456789012345678>" in captured["json"]["content"]
description = captured["json"]["embeds"][0]["description"]
assert "old-sku-123" in description
assert "new-sku-456" in description