Compare commits
5
Commits
4.1.1
..
fixed-cache
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
df1a60034c | ||
|
|
7094801549 | ||
|
|
389d10615e | ||
|
|
62b16f1dbf | ||
|
|
09662e656e |
@@ -1,289 +0,0 @@
|
|||||||
name: CI
|
|
||||||
|
|
||||||
on:
|
|
||||||
push:
|
|
||||||
branches: [main]
|
|
||||||
pull_request:
|
|
||||||
branches: [main]
|
|
||||||
schedule:
|
|
||||||
- cron: "0 6 * * 1"
|
|
||||||
|
|
||||||
jobs:
|
|
||||||
build-and-scan:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
steps:
|
|
||||||
- name: Checkout
|
|
||||||
uses: actions/checkout@v7
|
|
||||||
with:
|
|
||||||
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
|
|
||||||
if grep -q "Building wheel for" build.log; then
|
|
||||||
echo "::warning::A dependency was built from source — check Python/Alpine compatibility"
|
|
||||||
fi
|
|
||||||
|
|
||||||
- name: Smoke test (syntax check)
|
|
||||||
run: |
|
|
||||||
docker run --rm --entrypoint python nvidia-stock-bot:ci -c "
|
|
||||||
import ast, glob
|
|
||||||
ok = True
|
|
||||||
for f in glob.glob('/app/*.py'):
|
|
||||||
with open(f) as fh:
|
|
||||||
source = fh.read()
|
|
||||||
try:
|
|
||||||
ast.parse(source, filename=f)
|
|
||||||
except SyntaxError as e:
|
|
||||||
print(f'::error::Syntax error in {f}: {e}')
|
|
||||||
ok = False
|
|
||||||
if not ok:
|
|
||||||
exit(1)
|
|
||||||
print('OK: syntax is valid')
|
|
||||||
"
|
|
||||||
|
|
||||||
- name: Run unit tests
|
|
||||||
run: |
|
|
||||||
docker build --target test -t nvidia-stock-bot:test .
|
|
||||||
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: |
|
|
||||||
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" 2>&1 | tee deprecation.log || true
|
|
||||||
if grep -qi "deprecat" deprecation.log; then
|
|
||||||
echo "::warning::Deprecation warning detected, check logs"
|
|
||||||
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 "[email protected]"
|
|
||||||
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 \
|
|
||||||
-e DOCKER_HOST=tcp://dockerhost:2375 \
|
|
||||||
--add-host=dockerhost:host-gateway \
|
|
||||||
aquasec/trivy:0.74.0 image --exit-code 0 --severity HIGH nvidia-stock-bot:ci
|
|
||||||
|
|
||||||
- name: Publish tagged image
|
|
||||||
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
|
|
||||||
run: |
|
|
||||||
BEFORE="${{ github.event.before }}"
|
|
||||||
if [ -n "$BEFORE" ] && [ "$BEFORE" != "0000000000000000000000000000000000000000" ] && git cat-file -e "$BEFORE" 2>/dev/null; then
|
|
||||||
BASE_REF="$BEFORE"
|
|
||||||
else
|
|
||||||
BASE_REF="HEAD~1"
|
|
||||||
fi
|
|
||||||
CHANGED=$(git diff --name-only "$BASE_REF" "${{ github.sha }}")
|
|
||||||
echo "Changed files:"
|
|
||||||
echo "$CHANGED"
|
|
||||||
|
|
||||||
if ! echo "$CHANGED" | grep -qE '^(Dockerfile|VERSION)$|^app/'; then
|
|
||||||
echo "No container-relevant file changed, skipping publish."
|
|
||||||
exit 0
|
|
||||||
fi
|
|
||||||
|
|
||||||
if echo "$CHANGED" | grep -qE '^VERSION$'; then
|
|
||||||
echo "VERSION was manually edited in this push, using it as-is."
|
|
||||||
else
|
|
||||||
echo "VERSION untouched but container files changed, auto-bumping the build number (Z)."
|
|
||||||
OLD_VERSION=$(tr -d '[:space:]' < VERSION)
|
|
||||||
IFS='.' read -r MAJOR MINOR PATCH <<< "$OLD_VERSION"
|
|
||||||
NEW_VERSION="${MAJOR}.${MINOR}.$((PATCH + 1))"
|
|
||||||
echo "$NEW_VERSION" > VERSION
|
|
||||||
|
|
||||||
git config user.name "nvidia-stock-bot-ci"
|
|
||||||
git config user.email "[email protected]"
|
|
||||||
git add VERSION
|
|
||||||
git commit -m "Bump build version to $NEW_VERSION [skip ci]"
|
|
||||||
|
|
||||||
# Belt and suspenders: actions/checkout can leave its own ephemeral
|
|
||||||
# credential injected as an extraheader, which would silently override
|
|
||||||
# the URL-embedded token below. persist-credentials:false on checkout
|
|
||||||
# should already prevent this, but strip it here too just in case.
|
|
||||||
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:main
|
|
||||||
fi
|
|
||||||
|
|
||||||
VERSION=$(tr -d '[:space:]' < VERSION)
|
|
||||||
IFS='.' read -r MAJOR MINOR PATCH <<< "$VERSION"
|
|
||||||
MINOR_TAG="${MAJOR}.${MINOR}"
|
|
||||||
|
|
||||||
IMAGE=git.djeex.fr/djeex/nvidia-stock-bot
|
|
||||||
echo "${{ secrets.REGISTRY_TOKEN }}" | docker login git.djeex.fr -u Djeex --password-stdin
|
|
||||||
|
|
||||||
# Retag the already-built, already-scanned image — never rebuild for publish,
|
|
||||||
# so what ships is byte-for-byte what Trivy just scanned.
|
|
||||||
docker tag nvidia-stock-bot:ci "$IMAGE:latest"
|
|
||||||
docker tag nvidia-stock-bot:ci "$IMAGE:$MINOR_TAG"
|
|
||||||
docker tag nvidia-stock-bot:ci "$IMAGE:$VERSION"
|
|
||||||
docker push "$IMAGE:latest"
|
|
||||||
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)
|
|
||||||
|
|
||||||
CATEGORY="🔧 Maintenance"
|
|
||||||
CHANGE_TITLE="$TRIGGER_MSG"
|
|
||||||
|
|
||||||
if [ -n "$PR_NUM" ]; then
|
|
||||||
PR_JSON=$(curl -s -H "Authorization: token ${{ secrets.CI_PUSH_TOKEN }}" \
|
|
||||||
"https://git.djeex.fr/api/v1/repos/Djeex/nvidia-stock-bot/pulls/$PR_NUM")
|
|
||||||
PR_TITLE=$(echo "$PR_JSON" | jq -r '.title // empty' 2>/dev/null || true)
|
|
||||||
LABELS=$(echo "$PR_JSON" | jq -r '.labels[]?.name' 2>/dev/null || true)
|
|
||||||
|
|
||||||
if [ -n "$PR_TITLE" ]; then
|
|
||||||
CHANGE_TITLE="$PR_TITLE"
|
|
||||||
fi
|
|
||||||
|
|
||||||
if echo "$LABELS" | grep -qx 'bug'; then
|
|
||||||
CATEGORY="⚠️ Hotfix"
|
|
||||||
elif echo "$LABELS" | grep -qx 'major'; then
|
|
||||||
CATEGORY="💥 Breaking change"
|
|
||||||
elif echo "$LABELS" | grep -qx 'minor'; then
|
|
||||||
CATEGORY="✨ Update"
|
|
||||||
fi
|
|
||||||
fi
|
|
||||||
|
|
||||||
REPO_URL="https://git.djeex.fr/Djeex/nvidia-stock-bot"
|
|
||||||
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 }})"
|
|
||||||
if [ -n "$PR_NUM" ]; then
|
|
||||||
SOURCE_LINE="[#${PR_NUM}](${REPO_URL}/pulls/${PR_NUM}) · ${SOURCE_LINE}"
|
|
||||||
fi
|
|
||||||
|
|
||||||
BODY=$(cat <<EOF
|
|
||||||
## Changelog
|
|
||||||
|
|
||||||
### ${CATEGORY}
|
|
||||||
${CHANGE_TITLE}
|
|
||||||
|
|
||||||
**Source:** ${SOURCE_LINE}
|
|
||||||
**Image:** \`${IMAGE}:${VERSION}\`
|
|
||||||
|
|
||||||
**Commits:**
|
|
||||||
${COMMIT_LIST}
|
|
||||||
EOF
|
|
||||||
)
|
|
||||||
|
|
||||||
JSON_PAYLOAD=$(jq -n \
|
|
||||||
--arg tag "$VERSION" \
|
|
||||||
--arg name "$VERSION" \
|
|
||||||
--arg body "$BODY" \
|
|
||||||
'{tag_name: $tag, name: $name, target_commitish: "main", body: $body}')
|
|
||||||
|
|
||||||
curl -s -o /dev/null -w "Release API response: %{http_code}\n" -X POST \
|
|
||||||
-H "Authorization: token ${{ secrets.CI_PUSH_TOKEN }}" \
|
|
||||||
-H "Content-Type: application/json" \
|
|
||||||
-d "$JSON_PAYLOAD" \
|
|
||||||
"https://git.djeex.fr/api/v1/repos/Djeex/nvidia-stock-bot/releases"
|
|
||||||
|
|
||||||
MIRROR_NOTICE="_This github repo is a mirror of https://git.djeex.fr/Djeex/nvidia-stock-bot. You'll find full package, PR, history and release note there._"
|
|
||||||
GH_BODY=$(printf '%s\n\n%s' "$MIRROR_NOTICE" "$BODY")
|
|
||||||
|
|
||||||
GH_JSON_PAYLOAD=$(jq -n \
|
|
||||||
--arg tag "$VERSION" \
|
|
||||||
--arg name "$VERSION" \
|
|
||||||
--arg body "$GH_BODY" \
|
|
||||||
--arg sha "${{ github.sha }}" \
|
|
||||||
'{tag_name: $tag, name: $name, target_commitish: $sha, body: $body}')
|
|
||||||
|
|
||||||
GH_STATUS=0
|
|
||||||
for i in 1 2 3 4 5; do
|
|
||||||
GH_STATUS=$(curl -s -o /tmp/gh_release.json -w "%{http_code}" -X POST \
|
|
||||||
-H "Authorization: Bearer ${{ secrets.GH_TOKEN }}" \
|
|
||||||
-H "Accept: application/vnd.github+json" \
|
|
||||||
-H "Content-Type: application/json" \
|
|
||||||
-d "$GH_JSON_PAYLOAD" \
|
|
||||||
"https://api.github.com/repos/Djeex/nvidia-stock-bot/releases")
|
|
||||||
if [ "$GH_STATUS" = "201" ]; then
|
|
||||||
break
|
|
||||||
fi
|
|
||||||
echo "GitHub Release attempt $i failed (HTTP $GH_STATUS) — mirror may not have synced this commit yet, retrying in 15s..."
|
|
||||||
sleep 15
|
|
||||||
done
|
|
||||||
echo "GitHub Release API response: $GH_STATUS"
|
|
||||||
if [ "$GH_STATUS" != "201" ]; then
|
|
||||||
cat /tmp/gh_release.json 2>/dev/null || true
|
|
||||||
fi
|
|
||||||
@@ -1,4 +1,2 @@
|
|||||||
.venv
|
.venv
|
||||||
__pycache__/
|
__pycache__/
|
||||||
.coverage
|
|
||||||
.pytest_cache/
|
|
||||||
+1
-23
@@ -1,4 +1,4 @@
|
|||||||
FROM python:3.14.7-alpine AS base
|
FROM python:3.13-alpine
|
||||||
|
|
||||||
RUN apk add --no-cache ca-certificates
|
RUN apk add --no-cache ca-certificates
|
||||||
|
|
||||||
@@ -9,26 +9,4 @@ COPY /app/ /app/
|
|||||||
|
|
||||||
RUN pip install --no-cache-dir -r requirements.txt
|
RUN pip install --no-cache-dir -r requirements.txt
|
||||||
|
|
||||||
FROM base AS test
|
|
||||||
|
|
||||||
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
|
|
||||||
|
|
||||||
USER nvbot
|
|
||||||
|
|
||||||
CMD ["python", "main.py"]
|
CMD ["python", "main.py"]
|
||||||
+42
-52
@@ -1,13 +1,11 @@
|
|||||||
import json
|
|
||||||
import logging
|
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
# Read version from VERSION file
|
# Read version from VERSION file
|
||||||
with open(
|
with open(os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION"), "r", encoding="utf-8") as f:
|
||||||
os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION"), encoding="utf-8"
|
|
||||||
) as f:
|
|
||||||
VERSION = f.read().strip()
|
VERSION = f.read().strip()
|
||||||
|
|
||||||
# Logger setup
|
# Logger setup
|
||||||
@@ -25,13 +23,13 @@ logging.info("=" * 60)
|
|||||||
|
|
||||||
# Env variables
|
# Env variables
|
||||||
try:
|
try:
|
||||||
DISCORD_WEBHOOK_URL = os.environ["DISCORD_WEBHOOK_URL"]
|
DISCORD_WEBHOOK_URL = os.environ['DISCORD_WEBHOOK_URL']
|
||||||
DISCORD_SERVER_NAME = os.environ.get("DISCORD_SERVER_NAME", "Shared for free")
|
DISCORD_SERVER_NAME = os.environ.get('DISCORD_SERVER_NAME', 'Shared for free')
|
||||||
DISCORD_ROLES = os.environ.get("DISCORD_ROLES")
|
DISCORD_ROLES = os.environ.get('DISCORD_ROLES')
|
||||||
COUNTRY = os.environ.get("COUNTRY") or "US"
|
COUNTRY = os.environ.get('COUNTRY') or 'US'
|
||||||
REFRESH_TIME = int(os.environ.get("REFRESH_TIME") or 30)
|
REFRESH_TIME = int(os.environ.get('REFRESH_TIME') or 30)
|
||||||
TEST_MODE = os.environ.get("TEST_MODE", "False").lower() == "true"
|
TEST_MODE = os.environ.get('TEST_MODE', 'False').lower() == 'true'
|
||||||
PRODUCT_NAMES = os.environ["PRODUCT_NAMES"]
|
PRODUCT_NAMES = os.environ['PRODUCT_NAMES']
|
||||||
|
|
||||||
# Errors and warning
|
# Errors and warning
|
||||||
except KeyError as e:
|
except KeyError as e:
|
||||||
@@ -51,74 +49,71 @@ if not DISCORD_WEBHOOK_URL:
|
|||||||
logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.")
|
logging.error("❌ DISCORD_WEBHOOK_URL is required but not defined.")
|
||||||
sys.exit(1)
|
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
|
# Role mapping
|
||||||
DISCORD_ROLE_MAP = {}
|
DISCORD_ROLE_MAP = {}
|
||||||
if not DISCORD_ROLES or not DISCORD_ROLES.strip():
|
if not DISCORD_ROLES or not DISCORD_ROLES.strip():
|
||||||
logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.")
|
logging.warning("⚠️ DISCORD_ROLES not defined or empty. Defaulting all roles to @everyone.")
|
||||||
for name in PRODUCT_NAMES:
|
for name in PRODUCT_NAMES:
|
||||||
DISCORD_ROLE_MAP[name] = "@everyone"
|
DISCORD_ROLE_MAP[name] = '@everyone'
|
||||||
else:
|
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):
|
if len(roles) != len(PRODUCT_NAMES):
|
||||||
logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.")
|
logging.error("❌ The number of DISCORD_ROLES must match PRODUCT_NAMES.")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
for name, role in zip(PRODUCT_NAMES, roles):
|
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}")
|
logging.error(f"❌ Invalid DISCORD_ROLE format for {name}: {role}")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
DISCORD_ROLE_MAP[name] = role
|
DISCORD_ROLE_MAP[name] = role
|
||||||
|
|
||||||
# Masked webhook in terminal
|
# Masked webhook in terminal
|
||||||
match = re.search(r"/(\d+)/(.*)", DISCORD_WEBHOOK_URL)
|
match = re.search(r'/(\d+)/(.*)', DISCORD_WEBHOOK_URL)
|
||||||
if match:
|
if match:
|
||||||
webhook_id = match.group(1)
|
webhook_id = match.group(1)
|
||||||
webhook_token = match.group(2)
|
webhook_token = match.group(2)
|
||||||
masked_webhook_id = webhook_id[: len(webhook_id) - 10] + "*" * 10
|
masked_webhook_id = webhook_id[:len(webhook_id) - 10] + '*' * 10
|
||||||
masked_webhook_token = webhook_token[: len(webhook_token) - 120] + "*" * 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}"
|
wh_masked_url = f"https://discord.com/api/webhooks/{masked_webhook_id}/{masked_webhook_token}"
|
||||||
else:
|
else:
|
||||||
wh_masked_url = "[Invalid webhook URL]"
|
wh_masked_url = "[Invalid webhook URL]"
|
||||||
|
|
||||||
# HTTP headers
|
# HTTP headers - Firefox working set
|
||||||
HEADERS = {
|
HEADERS = {
|
||||||
"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",
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:145.0) Gecko/20100101 Firefox/145.0",
|
||||||
"Accept": "application/json, text/plain, */*",
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||||
"Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
|
"Accept-Language": "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3",
|
||||||
"Accept-Encoding": "gzip, deflate, br, zstd",
|
"Accept-Encoding": "gzip, deflate, br, zstd",
|
||||||
"Referer": "https://partners.nvidia.com/",
|
"Cache-Control": "no-cache",
|
||||||
"Origin": "https://partners.nvidia.com",
|
|
||||||
"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"',
|
|
||||||
"Cache-Control": "no-cache, no-store, must-revalidate",
|
|
||||||
"Pragma": "no-cache",
|
"Pragma": "no-cache",
|
||||||
"Expires": "0",
|
"Sec-GPC": "1",
|
||||||
|
"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"
|
||||||
}
|
}
|
||||||
|
|
||||||
# Load country setting and localization config
|
# Load country setting and localization config
|
||||||
country_code = os.environ.get("COUNTRY", "US").upper()
|
country_code = os.environ.get("COUNTRY", "US").upper()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with open("localization.json", encoding="utf-8") as f:
|
with open("localization.json", "r", encoding="utf-8") as f:
|
||||||
localization_config = json.load(f)
|
localization_config = json.load(f)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
logging.error("❌ localization.json file not found.")
|
logging.error("❌ localization.json file not found.")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
# Find country entry
|
# Find country entry
|
||||||
country_entry = next(
|
country_entry = next((entry for entry in localization_config if entry["country_code"].upper() == country_code), None)
|
||||||
(entry for entry in localization_config if entry["country_code"].upper() == country_code), None
|
|
||||||
)
|
|
||||||
|
|
||||||
if not country_entry:
|
if not country_entry:
|
||||||
logging.warning(f"⚠️ Country '{country_code}' not found in localization.json. Defaulting to US.")
|
logging.warning(f"⚠️ Country '{country_code}' not found in localization.json. Defaulting to US.")
|
||||||
country_entry = next(
|
country_entry = next((entry for entry in localization_config if entry["country_code"].upper() == "US"), None)
|
||||||
(entry for entry in localization_config if entry["country_code"].upper() == "US"), None
|
|
||||||
)
|
|
||||||
if not country_entry:
|
if not country_entry:
|
||||||
logging.error("❌ US fallback not found in localization.json.")
|
logging.error("❌ US fallback not found in localization.json.")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
@@ -130,7 +125,7 @@ currency = country_entry["currency"]
|
|||||||
|
|
||||||
# Load language file
|
# Load language file
|
||||||
try:
|
try:
|
||||||
with open("languages.json", encoding="utf-8") as f:
|
with open("languages.json", "r", encoding="utf-8") as f:
|
||||||
loc_lang = json.load(f)
|
loc_lang = json.load(f)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
logging.error("❌ languages.json file not found.")
|
logging.error("❌ languages.json file not found.")
|
||||||
@@ -147,15 +142,9 @@ if not loc:
|
|||||||
|
|
||||||
# Ensure all required keys are present
|
# Ensure all required keys are present
|
||||||
required_keys = [
|
required_keys = [
|
||||||
"in_stock_title",
|
"in_stock_title", "out_of_stock_title", "sku_change_title",
|
||||||
"out_of_stock_title",
|
"buy_now", "price", "time", "footer",
|
||||||
"sku_change_title",
|
"sku_description", "imminent_drop"
|
||||||
"buy_now",
|
|
||||||
"price",
|
|
||||||
"time",
|
|
||||||
"footer",
|
|
||||||
"sku_description",
|
|
||||||
"imminent_drop",
|
|
||||||
]
|
]
|
||||||
missing_keys = [key for key in required_keys if key not in loc]
|
missing_keys = [key for key in required_keys if key not in loc]
|
||||||
fallback = loc_lang.get("en", {})
|
fallback = loc_lang.get("en", {})
|
||||||
@@ -171,16 +160,17 @@ for key in missing_keys:
|
|||||||
locale = full_lang_code.lower()
|
locale = full_lang_code.lower()
|
||||||
API_URL_SKU = os.getenv(
|
API_URL_SKU = os.getenv(
|
||||||
"API_URL_SKU",
|
"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 = 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 = os.getenv(
|
||||||
"PRODUCT_URL",
|
"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
|
# Public constants
|
||||||
|
|||||||
+64
-24
@@ -1,20 +1,16 @@
|
|||||||
|
import requests
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
|
import random
|
||||||
import requests
|
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
|
from requests.adapters import HTTPAdapter, Retry
|
||||||
|
|
||||||
from env_config import API_URL_SKU, API_URL_STOCK, HEADERS, PRODUCT_NAMES, PRODUCT_URL
|
# HTTP session with stealth configuration
|
||||||
from notifier import (
|
|
||||||
send_discord_notification,
|
|
||||||
send_out_of_stock_notification,
|
|
||||||
send_sku_change_notification,
|
|
||||||
)
|
|
||||||
|
|
||||||
# HTTP session
|
|
||||||
session = requests.Session()
|
session = requests.Session()
|
||||||
retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
|
retries = Retry(total=2, backoff_factor=3, status_forcelist=[500, 502, 503, 504, 429])
|
||||||
session.mount("https://", HTTPAdapter(max_retries=retries))
|
adapter = HTTPAdapter(max_retries=retries, pool_connections=1, pool_maxsize=1)
|
||||||
|
session.mount('https://', adapter)
|
||||||
session.headers.update(HEADERS)
|
session.headers.update(HEADERS)
|
||||||
|
|
||||||
# Keeping memory of last run
|
# Keeping memory of last run
|
||||||
@@ -22,40 +18,82 @@ last_sku_dict = {}
|
|||||||
global_stock_status_dict = {}
|
global_stock_status_dict = {}
|
||||||
first_run_dict = {name: True for name in PRODUCT_NAMES}
|
first_run_dict = {name: True for name in PRODUCT_NAMES}
|
||||||
|
|
||||||
|
|
||||||
# Stock check function
|
# Stock check function
|
||||||
def check_rtx_50_founders():
|
def check_rtx_50_founders():
|
||||||
global last_sku_dict, global_stock_status_dict, first_run_dict
|
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
|
# Fetching nvidia API data
|
||||||
try:
|
try:
|
||||||
cache_buster = int(time.time() * 1000)
|
sku_url = API_URL_SKU
|
||||||
sku_url = f"{API_URL_SKU}&_t={cache_buster}"
|
|
||||||
|
|
||||||
response = session.get(sku_url, timeout=10)
|
response = session.get(sku_url, timeout=10)
|
||||||
logging.info(f"SKU API response: {response.status_code}")
|
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()
|
response.raise_for_status()
|
||||||
data = response.json()
|
|
||||||
|
# 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
|
||||||
except requests.exceptions.RequestException as e:
|
except requests.exceptions.RequestException as e:
|
||||||
logging.error(f"SKU API error: {e}")
|
logging.error(f"SKU API error: {e}")
|
||||||
return
|
return
|
||||||
|
|
||||||
# Checking productSKU and productUPC for all GPU set in PRODUCT_NAME
|
# 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:
|
for product_name in PRODUCT_NAMES:
|
||||||
product_details = None
|
product_details = None
|
||||||
for p in all_products:
|
for p in all_products:
|
||||||
if p.get("gpu", "").strip() == product_name:
|
gpu_name = p.get("gpu", "").strip()
|
||||||
|
# Flexible matching: exact match or partial match
|
||||||
|
if gpu_name == product_name or product_name in gpu_name:
|
||||||
product_details = p
|
product_details = p
|
||||||
break
|
break
|
||||||
|
|
||||||
if not product_details:
|
if not product_details:
|
||||||
logging.warning(f"⚠️ No product with GPU '{product_name}' found.")
|
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
|
continue
|
||||||
|
|
||||||
product_sku = product_details["productSKU"]
|
product_sku = product_details['productSKU']
|
||||||
product_upc = product_details.get("productUPC", "")
|
product_upc = product_details.get('productUPC', "")
|
||||||
if not isinstance(product_upc, list):
|
if not isinstance(product_upc, list):
|
||||||
product_upc = [product_upc]
|
product_upc = [product_upc]
|
||||||
|
|
||||||
@@ -69,13 +107,15 @@ def check_rtx_50_founders():
|
|||||||
first_run_dict[product_name] = False
|
first_run_dict[product_name] = False
|
||||||
|
|
||||||
# Check product availability in API_URL_STOCK for each SKU
|
# 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}"
|
||||||
api_stock_url = f"{API_URL_STOCK}{product_sku}&_t={cache_buster}"
|
|
||||||
logging.info(f"[{product_name}] Checking stock: {api_stock_url}")
|
logging.info(f"[{product_name}] Checking stock: {api_stock_url}")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
response = session.get(api_stock_url, timeout=10)
|
response = session.get(api_stock_url, timeout=10)
|
||||||
logging.info(f"[{product_name}] Stock API response: {response.status_code}")
|
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()
|
response.raise_for_status()
|
||||||
stock_data = response.json()
|
stock_data = response.json()
|
||||||
except requests.exceptions.RequestException as e:
|
except requests.exceptions.RequestException as e:
|
||||||
@@ -87,8 +127,8 @@ def check_rtx_50_founders():
|
|||||||
products_price = "Price not available"
|
products_price = "Price not available"
|
||||||
if isinstance(products, list) and len(products) > 0:
|
if isinstance(products, list) and len(products) > 0:
|
||||||
for p in products:
|
for p in products:
|
||||||
price = p.get("price", "Price not available")
|
price = p.get("price", 'Price not available')
|
||||||
if price != "Price not available":
|
if price != 'Price not available':
|
||||||
products_price = price
|
products_price = price
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
|
|||||||
+4
-6
@@ -1,20 +1,17 @@
|
|||||||
|
import time
|
||||||
import logging
|
import logging
|
||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
import time
|
|
||||||
|
|
||||||
from env_config import REFRESH_TIME
|
|
||||||
from gpu_checker import check_rtx_50_founders
|
from gpu_checker import check_rtx_50_founders
|
||||||
|
from env_config import REFRESH_TIME
|
||||||
|
|
||||||
# Signal handler function
|
# Signal handler function
|
||||||
def handle_exit(signum, frame):
|
def handle_exit(signum, frame):
|
||||||
logging.info(f"🛑 Received signal {signum}. Exiting gracefully...")
|
logging.info(f"🛑 Received signal {signum}. Exiting gracefully...")
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|
||||||
|
|
||||||
# Register signal handlers
|
# 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
|
signal.signal(signal.SIGTERM, handle_exit) # docker stop / kill -15
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
@@ -27,3 +24,4 @@ if __name__ == "__main__":
|
|||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
logging.info("🛑 Script interrupted by user (KeyboardInterrupt). Exiting gracefully.")
|
logging.info("🛑 Script interrupted by user (KeyboardInterrupt). Exiting gracefully.")
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
|
|
||||||
+19
-43
@@ -1,29 +1,15 @@
|
|||||||
import logging
|
|
||||||
import time
|
import time
|
||||||
|
import logging
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
from env_config import (
|
from env_config import (
|
||||||
DISCORD_ROLE_MAP,
|
DISCORD_WEBHOOK_URL, DISCORD_SERVER_NAME, DISCORD_ROLE_MAP, TEST_MODE, currency,
|
||||||
DISCORD_SERVER_NAME,
|
in_stock_title, out_of_stock_title, sku_change_title,
|
||||||
DISCORD_WEBHOOK_URL,
|
buy_now, price_label, time_label, footer, sku_description, imminent_drop
|
||||||
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"
|
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"
|
THUMBNAIL = "https://git.djeex.fr/Djeex/nvidia-stock-bot/raw/branch/main/assets/img/RTX5000.jpg"
|
||||||
|
|
||||||
|
|
||||||
# In stock
|
# In stock
|
||||||
def send_discord_notification(gpu_name, product_link, products_price):
|
def send_discord_notification(gpu_name, product_link, products_price):
|
||||||
timestamp = int(time.time())
|
timestamp = int(time.time())
|
||||||
@@ -38,20 +24,17 @@ def send_discord_notification(gpu_name, product_link, products_price):
|
|||||||
"author": {"name": "Nvidia Founder Editions"},
|
"author": {"name": "Nvidia Founder Editions"},
|
||||||
"fields": [
|
"fields": [
|
||||||
{"name": price_label, "value": f"`{currency}{products_price}`", "inline": True},
|
{"name": price_label, "value": f"`{currency}{products_price}`", "inline": True},
|
||||||
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True},
|
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}
|
||||||
],
|
],
|
||||||
"description": buy_now.format(product_link=product_link),
|
"description": buy_now.format(product_link=product_link),
|
||||||
"footer": {
|
"footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR}
|
||||||
"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME),
|
|
||||||
"icon_url": AVATAR,
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
payload = {
|
payload = {
|
||||||
"content": DISCORD_ROLE_MAP.get(gpu_name, "@everyone"),
|
"content": DISCORD_ROLE_MAP.get(gpu_name, "@everyone"),
|
||||||
"username": "NviBot",
|
"username": "NviBot",
|
||||||
"avatar_url": AVATAR,
|
"avatar_url": AVATAR,
|
||||||
"embeds": [embed],
|
"embeds": [embed]
|
||||||
}
|
}
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -63,7 +46,6 @@ def send_discord_notification(gpu_name, product_link, products_price):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(f"🚨 Error sending webhook: {e}")
|
logging.error(f"🚨 Error sending webhook: {e}")
|
||||||
|
|
||||||
|
|
||||||
# Out of stock
|
# Out of stock
|
||||||
def send_out_of_stock_notification(gpu_name, product_link, products_price):
|
def send_out_of_stock_notification(gpu_name, product_link, products_price):
|
||||||
timestamp = int(time.time())
|
timestamp = int(time.time())
|
||||||
@@ -77,16 +59,15 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price):
|
|||||||
"thumbnail": {"url": THUMBNAIL},
|
"thumbnail": {"url": THUMBNAIL},
|
||||||
"url": product_link,
|
"url": product_link,
|
||||||
"author": {"name": "Nvidia Founder Editions"},
|
"author": {"name": "Nvidia Founder Editions"},
|
||||||
"footer": {
|
"footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR},
|
||||||
"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME),
|
"fields": [{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}]
|
||||||
"icon_url": AVATAR,
|
|
||||||
},
|
|
||||||
"fields": [
|
|
||||||
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}
|
|
||||||
],
|
|
||||||
}
|
}
|
||||||
|
|
||||||
payload = {"username": "NviBot", "avatar_url": AVATAR, "embeds": [embed]}
|
payload = {
|
||||||
|
"username": "NviBot",
|
||||||
|
"avatar_url": AVATAR,
|
||||||
|
"embeds": [embed]
|
||||||
|
}
|
||||||
|
|
||||||
try:
|
try:
|
||||||
response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
|
response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
|
||||||
@@ -97,7 +78,6 @@ def send_out_of_stock_notification(gpu_name, product_link, products_price):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(f"🚨 Error sending webhook: {e}")
|
logging.error(f"🚨 Error sending webhook: {e}")
|
||||||
|
|
||||||
|
|
||||||
# SKU change
|
# SKU change
|
||||||
def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
|
def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
|
||||||
timestamp = int(time.time())
|
timestamp = int(time.time())
|
||||||
@@ -110,20 +90,15 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
|
|||||||
"url": product_link,
|
"url": product_link,
|
||||||
"description": sku_description.format(old_sku=old_sku, new_sku=new_sku),
|
"description": sku_description.format(old_sku=old_sku, new_sku=new_sku),
|
||||||
"color": 16776960,
|
"color": 16776960,
|
||||||
"footer": {
|
"footer": {"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME), "icon_url": AVATAR},
|
||||||
"text": footer.format(DISCORD_SERVER_NAME=DISCORD_SERVER_NAME),
|
"fields": [{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}]
|
||||||
"icon_url": AVATAR,
|
|
||||||
},
|
|
||||||
"fields": [
|
|
||||||
{"name": time_label, "value": f"<t:{timestamp}:d> <t:{timestamp}:T>", "inline": True}
|
|
||||||
],
|
|
||||||
}
|
}
|
||||||
|
|
||||||
payload = {
|
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",
|
"username": "NviBot",
|
||||||
"avatar_url": AVATAR,
|
"avatar_url": AVATAR,
|
||||||
"embeds": [embed],
|
"embeds": [embed]
|
||||||
}
|
}
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -134,3 +109,4 @@ def send_sku_change_notification(gpu_name, old_sku, new_sku, product_link):
|
|||||||
logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
|
logging.error(f"❌ Webhook error: {response.status_code} - {response.text}")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logging.error(f"🚨 Error sending webhook: {e}")
|
logging.error(f"🚨 Error sending webhook: {e}")
|
||||||
|
|
||||||
@@ -1 +1 @@
|
|||||||
requests==2.34.2
|
requests
|
||||||
@@ -1 +0,0 @@
|
|||||||
[pytest]
|
|
||||||
@@ -1,29 +0,0 @@
|
|||||||
{
|
|
||||||
"$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"]
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
@@ -1,13 +0,0 @@
|
|||||||
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"]
|
|
||||||
@@ -1,17 +0,0 @@
|
|||||||
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)
|
|
||||||
@@ -1,120 +0,0 @@
|
|||||||
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",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
@@ -1,162 +0,0 @@
|
|||||||
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
|
|
||||||
@@ -1,187 +0,0 @@
|
|||||||
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
|
|
||||||
Reference in New Issue
Block a user