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.
This commit is contained in:
@@ -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)
|
||||
@@ -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",
|
||||
})
|
||||
@@ -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
|
||||
@@ -0,0 +1,80 @@
|
||||
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_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_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