diff --git a/.gitignore b/.gitignore index eed9c57..5ee525b 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,5 @@ webapp/node_modules/ webapp/dist/ *.log .vscode/ +.pytest_cache/ + diff --git a/docs/TESTING.md b/docs/TESTING.md new file mode 100644 index 0000000..c5257cf --- /dev/null +++ b/docs/TESTING.md @@ -0,0 +1,74 @@ +# Automated tests + +A repo-committed `pytest` suite lives under `tests/`. It replaces the +ad-hoc "write a one-off script to a scratch temp dir, run it by hand" +verification method used throughout this project's development — every +test here follows the same pattern those scripts used (a fresh temp +SQLite DB, `create_app()`, `db.create_all()` + `migrate()` + `seed()`, +monkeypatching a module's `send_message` / `fiscalize_in_background` / +`get_invoice_status` attribute directly instead of pulling in a mocking +framework), just organized into real fixtures and committed so it runs in +CI / before every deploy instead of living only in someone's memory. + +## Running it + +```bash +pip install -r requirements-dev.txt +pytest +``` + +`pytest.ini` at the repo root points `pytest` at `tests/` and needs no +extra flags — just run it from the repo root (or the worktree root). + +Each test gets its own fresh temp SQLite database file (created and torn +down per test in `tests/conftest.py`'s `app` fixture) — slower than a +single shared DB, but immune to one test's leftover state (an order, a +lockout, a supplier) leaking into another. Module-level shared state used +intentionally in production for cross-request coordination (login +lockouts, the error-alert cooldown, the Checkbox token cache) is also +explicitly reset between tests — see `_reset_module_level_state()` in +`conftest.py`. + +## What's covered + +| Module | Test file | What's tested | +| --- | --- | --- | +| `bober_bbq/payments/monobank.py` | `tests/test_payments.py` | `claim_order_paid()` atomicity (first caller wins, second loses); the Monobank webhook, checkout poll, and reconciliation job never double-fire paid-order side effects for the same invoice | +| `bober_bbq/utils/checkbox_prro.py` | `tests/test_fiscalization_alerts.py` | `_mark_error()` alerts every owner, records `fiscal_status`/`fiscal_error`, and respects `error_alerts`' per-(exception, source) cooldown | +| `bober_bbq/utils/login_throttle.py` + `/admin/login` | `tests/test_login_throttle.py` | Lockout after `MAX_ATTEMPTS` wrong passwords from one IP (even the correct password is then rejected), an unaffected sibling IP, lockout expiry, and clearing the failure counter on success | +| `run_web._warn_on_insecure_defaults` | `tests/test_startup_checks.py` | Alerts the owner when `SECRET_KEY`/`FLASK_ADMIN_PASSWORD` are still the hardcoded defaults; stays silent once overridden | +| `bober_bbq/utils/supplier_import.py` | `tests/test_supplier_import_security.py` | `_resolve_upload_path()` rejects absolute paths (both slash styles), `../`/`..\` traversal, a wrong-shape-but-real-extension token, and an empty string — all with `FileNotFoundError`, none touching a file outside the upload temp dir; a legitimate token still round-trips | +| `bober_bbq/admin/suppliers.py` | `tests/test_inventory_suppliers.py` | `_apply_stock_topup()` credits stock, auto-assigns `Ingredient.supplier_id` only when unset (never overwrites an existing link), and leaves `StockMovement.note` exactly as typed (regression test for a real bug where it was getting overwritten with the supplier's name) | +| `bober_bbq/api/checkout.py` | `tests/test_checkout.py` | Delivery and pickup order creation via the checkout API produces an `Order` row with sane fields (type, totals, address/pickup fields, linked items) and clears the cart; an empty cart is rejected | + +## What's explicitly NOT covered yet + +- **The React webapp** (`webapp/`) has no automated tests at all (no unit + tests, no component tests, no E2E). It's tested manually in the browser. +- **The Telegram bot's conversation flows** (`bot/`) — aiogram handlers, + FSM states, inline keyboards — aren't covered beyond what the checkout + HTTP API itself exercises (`bot/services/order_service.py` and + `bot/services/cart_service.py`, indirectly, via `tests/test_checkout.py`). + A real bot conversation test would need an aiogram test harness/fake + `Bot` transport, which doesn't exist yet. +- **Checkbox (ПРРО) fiscalization's happy path** — `create_receipt()` + actually calling out to Checkbox's API (signin, shift open/close, + receipt creation, PDF polling) isn't exercised; only the error-handling + side (`_mark_error` / alerting) is. It's all `requests` calls to a real + third-party API with no local fake/sandbox available. +- **Monobank invoice creation** (`create_invoice`, `verify_token`, + webhook-signature verification against a real Monobank public key) — + only the *webhook handler's* payment-status logic is tested, with + signature verification monkeypatched out. Testing the actual HTTP calls + would need a fake Monobank server. +- **Google Drive backups** (`bober_bbq/utils/gdrive_backup.py`) — needs a + real (or mocked) Google service account and Drive API, neither of which + is set up for tests. +- **The scheduled jobs' timing/APScheduler wiring itself** in `run_web.py` + (only the job *functions* they call are tested directly, e.g. + `reconcile_pending_payments()`). +- **Admin panel pages beyond `/admin/login`** — the rest of the Jinja + admin UI (order management, menu editing, reports, settings screens) has + no route-level tests here; `tests/test_inventory_suppliers.py` calls + `_apply_stock_topup()` directly rather than through its HTTP route for + simplicity. diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..da636be --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +testpaths = tests +python_files = test_*.py +addopts = -ra diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..4d0c8ce --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,9 @@ +# Dev/test-only dependencies, on top of requirements.txt. +# +# Deliberately just pytest: the existing ad-hoc verification scripts used +# throughout this project's development monkeypatched dependencies +# directly (module.function = fake) with a plain check(label, cond) +# helper, no mocking framework — the pytest test suite under tests/ +# continues that style with stdlib unittest.mock (via pytest's own +# `monkeypatch` fixture), so nothing beyond pytest itself is needed. +pytest>=8.0 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ce07ede --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,116 @@ +"""Shared pytest fixtures for the Bober BBQ test suite. + +Converts the ad-hoc "scratch smoke script" pattern used throughout this +project's development (a temp SQLite file per run, `create_app()`, +`db.create_all()` + `migrate()` + `seed()`, monkeypatching modules' +`send_message` / `fiscalize_in_background` / `get_invoice_status` +attributes directly instead of a mocking framework) into a proper, +repo-committed pytest fixture. + +Each test gets its OWN fresh temp SQLite database (safer against state +leakage between tests than a session-scoped DB, at the cost of a little +extra setup time — `db.create_all()` + `seed()` against SQLite is fast +enough that this doesn't matter in practice). +""" + +import os +import sys +import tempfile +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +# Must be set before the FIRST import of bober_bbq.config anywhere in the +# process, since Config's class attributes are evaluated once at import +# time. conftest.py is imported by pytest before any test module gets +# collected, so this is early enough. +os.environ.setdefault("SECRET_KEY", "test-secret") +os.environ.setdefault("BOT_TOKEN", "") +os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") +# NOT "admin" on purpose: run_web.py's module-level `_warn_on_insecure_ +# defaults()` call (see below) must find nothing wrong the moment run_web +# is first imported, or it consumes error_alerts' one-per-cooldown-window +# alert slot before test_startup_checks.py ever gets to exercise it. +os.environ.setdefault("FLASK_ADMIN_PASSWORD", "a-real-non-default-test-password") + +from bober_bbq.app import create_app # noqa: E402 +from bober_bbq.config import config # noqa: E402 +from bober_bbq.extensions import db # noqa: E402 +from bober_bbq.migrate import migrate # noqa: E402 +from bober_bbq.seed import seed # noqa: E402 +from bober_bbq.utils import checkbox_prro, error_alerts, login_throttle, reconciliation # noqa: E402 + +# Imported here, once, with the safe SECRET_KEY/FLASK_ADMIN_PASSWORD env +# defaults already in place above — run_web.py runs `_warn_on_insecure_ +# defaults()` unconditionally at module import time (not just under +# `if __name__ == "__main__"`), so importing it for the first time with +# insecure values already in place would fire (and consume the cooldown +# for) the very alert test_startup_checks.py exists to test. Tests call +# `run_web._warn_on_insecure_defaults()` directly afterward against +# whatever config values they set up. +import run_web # noqa: E402 + + +def _reset_module_level_state(): + """These modules intentionally use module-level dicts as shared state + across request threads in production (see their own docstrings/ + comments) — e.g. login_throttle's lockout tracking, error_alerts' + per-(exception, source) cooldown. That state must NOT leak between + tests, or an early test's alert cooldown / lockout would silently + suppress a later, unrelated test's assertions.""" + login_throttle._failures.clear() + login_throttle._locked_until.clear() + error_alerts._last_sent.clear() + checkbox_prro._token_cache.update({"token": None, "at": 0.0}) + reconciliation._last_token_ok = None + + +@pytest.fixture +def app(): + """A fresh Flask app wired to its own temp SQLite DB, seeded with the + normal default data (categories/products/settings/admin user).""" + tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + tmp_db.close() + + # config is a module-level singleton (bober_bbq.config.config); override + # the instance attribute directly rather than the env var, since the + # Config class attribute was already evaluated at import time. + config.SQLALCHEMY_DATABASE_URI = f"sqlite:///{tmp_db.name}" + config.OWNER_IDS = [] + + flask_app = create_app() + flask_app.config["WTF_CSRF_ENABLED"] = False + flask_app.testing = True + # Enables the ?dev_user_id=... Telegram-auth bypass in + # api/auth.require_telegram_auth (see its own comment) — there is no + # real Telegram WebView in tests to sign init data with. + flask_app.debug = True + + with flask_app.app_context(): + db.create_all() + migrate() + seed() + + _reset_module_level_state() + + yield flask_app + + with flask_app.app_context(): + db.session.remove() + db.engine.dispose() + try: + os.unlink(tmp_db.name) + except PermissionError: + # Windows sometimes still holds the file handle briefly after + # dispose() — matches the original scratch-script behavior of + # tolerating this rather than failing the whole run over cleanup. + pass + + +@pytest.fixture +def client(app): + return app.test_client() diff --git a/tests/test_checkout.py b/tests/test_checkout.py new file mode 100644 index 0000000..1b8c2a5 --- /dev/null +++ b/tests/test_checkout.py @@ -0,0 +1,135 @@ +"""Basic checkout smoke coverage: creating a delivery or pickup order via +/api/checkout/* succeeds and produces an Order row with sane fields. + +Uses the ?dev_user_id=... bypass in api.auth.require_telegram_auth (only +active when app.debug is True, set by the `app` fixture) instead of +signing real Telegram WebApp init data — there's no real Telegram WebView +in a test process to sign it with. +""" + +from bober_bbq.extensions import db +from bober_bbq.models import CartItem, Order, Product, Setting, TelegramUser + + +def _make_customer_with_cart_item(app, telegram_id=901): + with app.app_context(): + user = TelegramUser(telegram_id=telegram_id, first_name="Тест") + db.session.add(user) + product = Product.query.filter_by(is_active=True).first() + assert product is not None, "seed() should have created at least one active product" + db.session.add(CartItem(user_id=telegram_id, product_id=product.id, quantity=2)) + # Orders are gated on working hours — force them allowed regardless + # of what time of day the test suite happens to run. + Setting.set("allow_orders_outside_hours", "1") + db.session.commit() + return telegram_id, product.id, product.price + + +def test_delivery_checkout_creates_a_sane_order(app, monkeypatch): + from bober_bbq.utils import notify_manager, telegram_notify + + monkeypatch.setattr(telegram_notify, "send_message", lambda *a, **kw: None) + monkeypatch.setattr(notify_manager, "send_message", lambda *a, **kw: None) + import bober_bbq.api.checkout as checkout_mod + + monkeypatch.setattr(checkout_mod, "send_message", lambda *a, **kw: None) + + telegram_id, product_id, unit_price = _make_customer_with_cart_item(app, telegram_id=901) + client = app.test_client() + + resp = client.post( + "/api/checkout/delivery", + json={ + "name": "Іван Іваненко", + "phone": "+380501234567", + "address_city": "Київ", + "address_street": "Хрещатик", + "address_house": "1", + "payment_method": "cash", + }, + query_string={"dev_user_id": telegram_id}, + ) + + assert resp.status_code == 201, resp.get_data(as_text=True) + body = resp.get_json() + assert body["order_id"] + assert body["payment_method"] == "cash" + + with app.app_context(): + order = db.session.get(Order, body["order_id"]) + assert order is not None + assert order.order_type == "delivery" + assert order.user_id == telegram_id + assert order.status == "new" + assert order.payment_status == "unpaid" + assert order.customer_name == "Іван Іваненко" + assert order.address_city == "Київ" + assert order.items_total == unit_price * 2 + # delivery_fee is derived from items_total by pricing rules we don't + # need to duplicate here — just sanity-check the total is internally + # consistent, matches what the API returned, and isn't free. + assert order.total == order.items_total + order.delivery_fee + assert order.total > 0 + assert body["total"] == order.total + assert len(order.items) == 1 + assert order.items[0].product_id == product_id + + # The cart must be cleared once the order is placed. + assert CartItem.query.filter_by(user_id=telegram_id).count() == 0 + + +def test_pickup_checkout_creates_a_sane_order(app, monkeypatch): + from bober_bbq.utils import notify_manager, telegram_notify + + monkeypatch.setattr(telegram_notify, "send_message", lambda *a, **kw: None) + monkeypatch.setattr(notify_manager, "send_message", lambda *a, **kw: None) + import bober_bbq.api.checkout as checkout_mod + + monkeypatch.setattr(checkout_mod, "send_message", lambda *a, **kw: None) + + telegram_id, product_id, unit_price = _make_customer_with_cart_item(app, telegram_id=902) + client = app.test_client() + + resp = client.post( + "/api/checkout/pickup", + json={ + "name": "Марія Петренко", + "phone": "+380671112233", + "pickup_time": "18:30", + "payment_method": "cash", + }, + query_string={"dev_user_id": telegram_id}, + ) + + assert resp.status_code == 201, resp.get_data(as_text=True) + body = resp.get_json() + assert body["order_id"] + assert body["payment_method"] == "cash" + + with app.app_context(): + order = db.session.get(Order, body["order_id"]) + assert order is not None + assert order.order_type == "pickup" + assert order.user_id == telegram_id + assert order.pickup_time == "18:30" + assert order.delivery_fee == 0 + assert order.total == order.items_total - order.discount_amount + assert len(order.items) == 1 + assert order.items[0].product_id == product_id + + +def test_checkout_rejects_an_empty_cart(app): + with app.app_context(): + user = TelegramUser(telegram_id=903, first_name="Порожній") + db.session.add(user) + Setting.set("allow_orders_outside_hours", "1") + db.session.commit() + + client = app.test_client() + resp = client.post( + "/api/checkout/pickup", + json={"name": "Хтось", "phone": "+380990000000", "pickup_time": "12:00", "payment_method": "cash"}, + query_string={"dev_user_id": 903}, + ) + assert resp.status_code == 400 + assert resp.get_json()["error"] == "cart_empty" diff --git a/tests/test_fiscalization_alerts.py b/tests/test_fiscalization_alerts.py new file mode 100644 index 0000000..686f246 --- /dev/null +++ b/tests/test_fiscalization_alerts.py @@ -0,0 +1,70 @@ +"""checkbox_prro._mark_error() must alert the owner(s) via error_alerts, tag +the order with the fiscal error, and respect error_alerts' cooldown so a +repeated failure doesn't flood the owner's chat. + +Mirrors the ad-hoc smoke34.py script written during development. +""" + +from bober_bbq.extensions import db +from bober_bbq.models import Order, TelegramUser +from bober_bbq.utils import checkbox_prro + + +def _make_order(app, telegram_id=888): + with app.app_context(): + tg_user = TelegramUser(telegram_id=telegram_id, first_name="Test") + db.session.add(tg_user) + order = Order( + user_id=telegram_id, + order_type="delivery", + status="new", + payment_method="cash", + payment_status="unpaid", + total=100, + customer_name="Т", + customer_phone="+380000000000", + ) + db.session.add(order) + db.session.commit() + return order.id + + +def test_mark_error_alerts_owners_and_records_fiscal_error(app, monkeypatch): + from bober_bbq.config import config + from bober_bbq.utils import error_alerts + + config.OWNER_IDS = [111, 222] + sent = [] + monkeypatch.setattr(error_alerts, "send_message", lambda chat_id, text, **kw: sent.append((chat_id, text))) + + order_id = _make_order(app) + with app.app_context(): + order = db.session.get(Order, order_id) + checkbox_prro._mark_error(order, "Checkbox API повернув 500") + + assert sorted(c for c, _ in sent) == [111, 222], "both owners must be alerted" + assert all(order.display_number() in t for _, t in sent), "the alert must mention the order" + assert order.fiscal_status == "error" + assert order.fiscal_error == "Checkbox API повернув 500" + + +def test_repeat_error_within_cooldown_is_not_re_alerted(app, monkeypatch): + from bober_bbq.config import config + from bober_bbq.utils import error_alerts + + config.OWNER_IDS = [111, 222] + sent = [] + monkeypatch.setattr(error_alerts, "send_message", lambda chat_id, text, **kw: sent.append((chat_id, text))) + + order_id = _make_order(app) + with app.app_context(): + order = db.session.get(Order, order_id) + checkbox_prro._mark_error(order, "Checkbox API повернув 500") + assert len(sent) == 2 # sanity check on the first alert + + sent.clear() + # Same (exception type, source) again immediately -> cooldown suppresses it. + checkbox_prro._mark_error(order, "Checkbox API повернув 500 знову") + assert sent == [], "a repeat error within the cooldown window must not re-alert" + # ... but the order's own fiscal_error field still reflects the latest failure. + assert order.fiscal_error == "Checkbox API повернув 500 знову" diff --git a/tests/test_inventory_suppliers.py b/tests/test_inventory_suppliers.py new file mode 100644 index 0000000..81aeca8 --- /dev/null +++ b/tests/test_inventory_suppliers.py @@ -0,0 +1,124 @@ +"""bober_bbq.admin.suppliers._apply_stock_topup(): restocking via a +supplier delivery must credit Ingredient.current_stock, auto-assign +Ingredient.supplier_id only when it was previously unset (never overwrite +an existing link), and leave StockMovement.note exactly as the admin +typed it — NOT overwritten with the supplier's name, which was a real bug +fixed earlier this session (the note field duplicated information already +available via the ingredient's own supplier link, surfaced separately in +inventory reports). +""" + +from bober_bbq.admin.suppliers import _apply_stock_topup +from bober_bbq.extensions import db +from bober_bbq.models import Ingredient, StockMovement, Supplier, SupplierDelivery + + +def _make_supplier(name): + supplier = Supplier(name=name) + db.session.add(supplier) + db.session.commit() + return supplier + + +def test_topup_credits_stock_and_assigns_unset_supplier(app): + with app.app_context(): + supplier = _make_supplier("Постачальник Ковбаси") + ingredient = Ingredient(name="Ковбаса", unit="кг", current_stock=5, supplier_id=None) + db.session.add(ingredient) + db.session.commit() + + delivery = SupplierDelivery( + supplier_id=supplier.id, + ingredient_id=ingredient.id, + product_name="Ковбаса", + quantity=10, + delivery_sum=200, + note="Партія trohy volog", + ) + db.session.add(delivery) + db.session.commit() + + _apply_stock_topup(delivery) + + assert ingredient.current_stock == 15.0 + assert ingredient.supplier_id == supplier.id, "an unset supplier link must be auto-assigned from the delivery" + + +def test_topup_never_overwrites_an_existing_supplier_link(app): + with app.app_context(): + original_supplier = _make_supplier("Оригінальний постачальник") + other_supplier = _make_supplier("Інший постачальник") + ingredient = Ingredient(name="Сіль", unit="кг", current_stock=1, supplier_id=original_supplier.id) + db.session.add(ingredient) + db.session.commit() + + delivery = SupplierDelivery( + supplier_id=other_supplier.id, + ingredient_id=ingredient.id, + product_name="Сіль", + quantity=5, + delivery_sum=50, + ) + db.session.add(delivery) + db.session.commit() + + _apply_stock_topup(delivery) + + assert ingredient.current_stock == 6.0 + assert ingredient.supplier_id == original_supplier.id, "an already-set supplier link must survive a topup from a different supplier" + + +def test_topup_stock_movement_note_is_exactly_what_the_admin_typed(app): + """Regression test for a real production bug: the StockMovement.note + created by a restock was being overwritten with the supplier's name + instead of preserving the admin's own free-text note.""" + with app.app_context(): + supplier = _make_supplier("Постачальник Пряме Поповнення") + ingredient = Ingredient(name="Кетчуп", unit="кг", current_stock=5, supplier_id=supplier.id) + db.session.add(ingredient) + db.session.commit() + + admin_note = "накладна №НК-9, часткова оплата готівкою" + delivery = SupplierDelivery( + supplier_id=supplier.id, + ingredient_id=ingredient.id, + product_name="Кетчуп", + quantity=10, + delivery_sum=200, + reference="НК-9", + note=admin_note, + ) + db.session.add(delivery) + db.session.commit() + + _apply_stock_topup(delivery) + + movement = StockMovement.query.filter_by(ingredient_id=ingredient.id, reason="restock").first() + assert movement is not None + assert movement.note == admin_note, "the movement note must be exactly what the admin typed, not the supplier's name" + assert supplier.name not in (movement.note or ""), "the supplier name must not be injected into the note" + + +def test_topup_with_no_note_leaves_movement_note_blank(app): + with app.app_context(): + supplier = _make_supplier("Постачальник Без Нотатки") + ingredient = Ingredient(name="Цукор", unit="кг", current_stock=0, supplier_id=None) + db.session.add(ingredient) + db.session.commit() + + delivery = SupplierDelivery( + supplier_id=supplier.id, + ingredient_id=ingredient.id, + product_name="Цукор", + quantity=20, + delivery_sum=400, + note=None, + ) + db.session.add(delivery) + db.session.commit() + + _apply_stock_topup(delivery) + + movement = StockMovement.query.filter_by(ingredient_id=ingredient.id, reason="restock").first() + assert movement is not None + assert movement.note is None diff --git a/tests/test_login_throttle.py b/tests/test_login_throttle.py new file mode 100644 index 0000000..1f6ff26 --- /dev/null +++ b/tests/test_login_throttle.py @@ -0,0 +1,101 @@ +"""Brute-force lockout for /admin/login (bober_bbq.utils.login_throttle + +the login route in bober_bbq.admin.routes): N wrong passwords from one IP +locks that IP out (even the correct password is then rejected), a +different IP is unaffected, and a successful login clears the failure +counter for that IP. + +Mirrors the ad-hoc smoke33.py script written during development. +""" + +from bober_bbq.config import config +from bober_bbq.utils import login_throttle + + +def _admin_credentials(app): + with app.app_context(): + return config.FLASK_ADMIN_USERNAME, config.FLASK_ADMIN_PASSWORD + + +def test_repeated_wrong_passwords_lock_out_the_ip(app, monkeypatch): + monkeypatch.setattr(config, "OWNER_IDS", [999]) + from bober_bbq.utils import error_alerts + + alerts = [] + monkeypatch.setattr(error_alerts, "send_message", lambda chat_id, text, **kw: alerts.append((chat_id, text))) + + username, _ = _admin_credentials(app) + client = app.test_client() + client.environ_base["REMOTE_ADDR"] = "203.0.113.7" + + for i in range(login_throttle.MAX_ATTEMPTS): + resp = client.post("/admin/login", data={"username": username, "password": "wrong"}, follow_redirects=True) + if i < login_throttle.MAX_ATTEMPTS - 1: + assert "Невірний логін або пароль" in resp.get_data(as_text=True) + + assert login_throttle.seconds_locked("203.0.113.7") is not None, "IP must be locked after MAX_ATTEMPTS failures" + assert len(alerts) == 1 and alerts[0][0] == 999, "the owner must be alerted about the lockout" + + +def test_correct_password_still_rejected_while_locked_out(app): + username, password = _admin_credentials(app) + client = app.test_client() + client.environ_base["REMOTE_ADDR"] = "203.0.113.8" + + for _ in range(login_throttle.MAX_ATTEMPTS): + client.post("/admin/login", data={"username": username, "password": "wrong"}, follow_redirects=True) + + resp = client.post("/admin/login", data={"username": username, "password": password}, follow_redirects=True) + assert "Забагато невдалих спроб" in resp.get_data(as_text=True) + + with client.session_transaction() as sess: + assert "_user_id" not in sess, "a locked-out IP must not be logged in even with the correct password" + + +def test_a_different_ip_is_unaffected_by_another_ips_lockout(app): + username, password = _admin_credentials(app) + + locked_client = app.test_client() + locked_client.environ_base["REMOTE_ADDR"] = "203.0.113.9" + for _ in range(login_throttle.MAX_ATTEMPTS): + locked_client.post("/admin/login", data={"username": username, "password": "wrong"}, follow_redirects=True) + assert login_throttle.seconds_locked("203.0.113.9") is not None + + other_client = app.test_client() + other_client.environ_base["REMOTE_ADDR"] = "198.51.100.1" + resp = other_client.post("/admin/login", data={"username": username, "password": password}, follow_redirects=True) + assert resp.request.path == "/admin/", "a different IP must be able to log in normally" + + +def test_successful_login_clears_the_failure_counter_for_that_ip(app): + username, password = _admin_credentials(app) + client = app.test_client() + client.environ_base["REMOTE_ADDR"] = "203.0.113.10" + + # A couple of wrong attempts (fewer than MAX_ATTEMPTS, so no lockout yet)... + for _ in range(login_throttle.MAX_ATTEMPTS - 1): + client.post("/admin/login", data={"username": username, "password": "wrong"}, follow_redirects=True) + assert "203.0.113.10" in login_throttle._failures + + # ...then a successful login must wipe that IP's failure history, not + # just let the window expire naturally. + resp = client.post("/admin/login", data={"username": username, "password": password}, follow_redirects=True) + assert resp.request.path == "/admin/" + assert "203.0.113.10" not in login_throttle._failures + assert login_throttle.seconds_locked("203.0.113.10") is None + + +def test_lockout_expires_after_the_window_passes(app): + username, password = _admin_credentials(app) + client = app.test_client() + client.environ_base["REMOTE_ADDR"] = "203.0.113.11" + + for _ in range(login_throttle.MAX_ATTEMPTS): + client.post("/admin/login", data={"username": username, "password": "wrong"}, follow_redirects=True) + assert login_throttle.seconds_locked("203.0.113.11") is not None + + # Force-expire rather than sleeping the real 15 minutes out. + login_throttle._locked_until["203.0.113.11"] = 0 + assert login_throttle.seconds_locked("203.0.113.11") is None + + resp = client.post("/admin/login", data={"username": username, "password": password}, follow_redirects=True) + assert resp.request.path == "/admin/", "correct password must succeed once the lockout window has passed" diff --git a/tests/test_payments.py b/tests/test_payments.py new file mode 100644 index 0000000..8025638 --- /dev/null +++ b/tests/test_payments.py @@ -0,0 +1,157 @@ +"""Payment-race coverage: claim_order_paid() atomicity, and the cross-path +dedup between the Monobank webhook, the checkout poll, and the periodic +reconciliation job — all three can observe the same "paid" invoice, but the +paid-order side effects (staff notification, fiscalization) must fire +exactly once total, not once per path. + +Mirrors the ad-hoc smoke33.py script written during development, converted +into proper pytest fixtures/tests. +""" + +import json + +import bober_bbq.api.checkout as checkout_mod +import bober_bbq.api.payments as payments_mod +import bober_bbq.utils.reconciliation as reconciliation_mod +from bober_bbq.extensions import db +from bober_bbq.models import Order, TelegramUser +from bober_bbq.payments.monobank import claim_order_paid + + +def _make_paid_candidate_order(app, invoice_id="inv-race", telegram_id=777): + with app.app_context(): + tg_user = TelegramUser(telegram_id=telegram_id, first_name="Test") + db.session.add(tg_user) + order = Order( + user_id=telegram_id, + order_type="delivery", + status="new", + payment_method="card", + payment_status="unpaid", + monobank_invoice_id=invoice_id, + total=100, + customer_name="Т", + customer_phone="+380000000000", + ) + db.session.add(order) + db.session.commit() + return order.id + + +def test_claim_order_paid_is_atomic_first_caller_wins(app): + order_id = _make_paid_candidate_order(app) + + with app.app_context(): + order_view_1 = db.session.get(Order, order_id) + order_view_2 = db.session.get(Order, order_id) + + first = claim_order_paid(order_view_1) + second = claim_order_paid(order_view_2) + + assert first is True, "the first caller to claim the transition must win" + assert second is False, "a second claim on an already-paid order must lose" + + fresh = db.session.get(Order, order_id) + assert fresh.payment_status == "paid" + assert fresh.paid_at is not None + + +def _patch_side_effect_hooks(monkeypatch, calls): + """Monkeypatches the module-level names each call site imported directly + (`from ... import send_message`, etc.) — the same style used by every + ad-hoc smoke script this session, kept deliberately rather than + introducing a mocking framework, since each call site binds its own + reference at import time and patching the source module wouldn't affect + already-bound names in api/payments.py, api/checkout.py, or + utils/reconciliation.py.""" + + def fake_fiscalize_in_background(order_id): + calls["fiscalize"].append(order_id) + + def fake_send_message(chat_id, text, **kw): + if str(chat_id).startswith("-100"): + calls["staff_msgs"].append((chat_id, text)) + else: + calls["user_msgs"].append((chat_id, text)) + + def fake_staff_chat_id(order_type): + return "-100999" + + for mod in (payments_mod, checkout_mod, reconciliation_mod): + monkeypatch.setattr(mod.checkbox_prro, "fiscalize_in_background", fake_fiscalize_in_background) + monkeypatch.setattr(mod, "send_message", fake_send_message) + monkeypatch.setattr(mod, "staff_chat_id", fake_staff_chat_id) + + monkeypatch.setattr(checkout_mod, "PAID_STATUSES", {"success"}) + + +def test_duplicate_monobank_webhook_delivery_fires_side_effects_once(app, monkeypatch): + """A duplicate webhook delivery is a real-world occurrence (Monobank + retries on anything but a prompt 200) — the second delivery must not + double-notify or double-fiscalize.""" + order_id = _make_paid_candidate_order(app, invoice_id="inv-webhook-dup") + calls = {"fiscalize": [], "staff_msgs": [], "user_msgs": []} + _patch_side_effect_hooks(monkeypatch, calls) + monkeypatch.setattr(payments_mod, "verify_webhook_signature", lambda raw, sig: True) + + client = app.test_client() + payload = json.dumps({"invoiceId": "inv-webhook-dup", "status": "success"}).encode() + + resp1 = client.post( + "/api/payments/monobank/webhook", data=payload, content_type="application/json", headers={"X-Sign": "x"} + ) + resp2 = client.post( + "/api/payments/monobank/webhook", data=payload, content_type="application/json", headers={"X-Sign": "x"} + ) + + assert resp1.status_code == 200 + assert resp2.status_code == 200 + assert calls["fiscalize"] == [order_id], "fiscalize must fire exactly once despite the duplicate webhook" + assert len(calls["staff_msgs"]) == 1, "staff must be notified exactly once" + assert len(calls["user_msgs"]) == 1, "the customer must be notified exactly once" + + with app.app_context(): + fresh = db.session.get(Order, order_id) + assert fresh.payment_status == "paid" + + +def test_webhook_checkout_poll_and_reconciliation_dedup_across_paths(app, monkeypatch): + """The three independent observers of "this invoice is now paid" — the + Monobank webhook, the Mini App's own payment-status poll, and the + periodic reconciliation job — must combine to fire the paid-order side + effects exactly once total, no matter which order they run in or how + many of them see the paid status.""" + order_id = _make_paid_candidate_order(app, invoice_id="inv-cross-path", telegram_id=778) + calls = {"fiscalize": [], "staff_msgs": [], "user_msgs": []} + _patch_side_effect_hooks(monkeypatch, calls) + monkeypatch.setattr(payments_mod, "verify_webhook_signature", lambda raw, sig: True) + monkeypatch.setattr(checkout_mod, "get_invoice_status", lambda invoice_id: {"status": "success"}) + monkeypatch.setattr(reconciliation_mod, "get_invoice_status", lambda invoice_id: {"status": "success"}) + + client = app.test_client() + + # 1) Webhook observes it first and wins the race. + payload = json.dumps({"invoiceId": "inv-cross-path", "status": "success"}).encode() + resp = client.post( + "/api/payments/monobank/webhook", data=payload, content_type="application/json", headers={"X-Sign": "x"} + ) + assert resp.status_code == 200 + assert calls["fiscalize"] == [order_id] + assert len(calls["staff_msgs"]) == 1 + + # 2) The Mini App poll observes the same "success" status a moment + # later (e.g. the customer tapped "Перевірити оплату" right as the + # webhook was landing) — claim_order_paid() must reject the second + # transition, so no additional side effects fire. + resp2 = client.get(f"/api/checkout/payment-status/{order_id}?dev_user_id=778") + assert resp2.get_json() == {"paid": True} + assert calls["fiscalize"] == [order_id], "checkout poll must not re-fiscalize after the webhook already won" + assert len(calls["staff_msgs"]) == 1, "checkout poll must not re-notify staff" + + # 3) The periodic reconciliation job later also observes the same + # invoice as paid (e.g. its own polling cycle) — must be a no-op too. + with app.app_context(): + reconciliation_mod.reconcile_pending_payments() + + assert calls["fiscalize"] == [order_id], "reconciliation must not re-fiscalize" + assert len(calls["staff_msgs"]) == 1, "reconciliation must not re-notify staff" diff --git a/tests/test_startup_checks.py b/tests/test_startup_checks.py new file mode 100644 index 0000000..9de8d00 --- /dev/null +++ b/tests/test_startup_checks.py @@ -0,0 +1,44 @@ +"""run_web._warn_on_insecure_defaults(): fires an owner alert when +config.SECRET_KEY / config.FLASK_ADMIN_PASSWORD are still the hardcoded +defaults from bober_bbq/config.py, and stays silent once they're +overridden with real values. + +run_web is imported once, up front, in conftest.py — see the comment +there for why (its module-level code runs this same check unconditionally +at import time, and must find nothing wrong then so it doesn't consume +the alert cooldown this test relies on). + +Mirrors the ad-hoc smoke33.py script written during development. +""" + +import run_web + +from bober_bbq.config import config +from bober_bbq.utils import error_alerts + + +def test_warns_when_defaults_are_still_in_place(app, monkeypatch): + alerts = [] + monkeypatch.setattr(error_alerts, "send_message", lambda chat_id, text, **kw: alerts.append((chat_id, text))) + monkeypatch.setattr(config, "OWNER_IDS", [999]) + monkeypatch.setattr(config, "SECRET_KEY", "dev-secret-change-me") + monkeypatch.setattr(config, "FLASK_ADMIN_PASSWORD", "admin") + + run_web._warn_on_insecure_defaults() + + assert len(alerts) == 1 + assert alerts[0][0] == 999 + assert "SECRET_KEY" in alerts[0][1] + assert "admin" in alerts[0][1] + + +def test_stays_silent_once_defaults_are_overridden(app, monkeypatch): + alerts = [] + monkeypatch.setattr(error_alerts, "send_message", lambda chat_id, text, **kw: alerts.append((chat_id, text))) + monkeypatch.setattr(config, "OWNER_IDS", [999]) + monkeypatch.setattr(config, "SECRET_KEY", "a-real-random-secret") + monkeypatch.setattr(config, "FLASK_ADMIN_PASSWORD", "a-real-password") + + run_web._warn_on_insecure_defaults() + + assert alerts == [] diff --git a/tests/test_supplier_import_security.py b/tests/test_supplier_import_security.py new file mode 100644 index 0000000..d0679ae --- /dev/null +++ b/tests/test_supplier_import_security.py @@ -0,0 +1,92 @@ +"""Path-traversal protection for the supplier-import upload token +(bober_bbq.utils.supplier_import._resolve_upload_path and its callers). + +`token` round-trips through a client-visible hidden form field across the +preview/confirm steps of the Excel/CSV import wizard, so every function +that resolves it against TMP_DIR must reject anything that isn't exactly +"<32 hex chars>" — otherwise a crafted token (an +absolute path, or one with ../ segments) could read or delete an arbitrary +file on the server. + +Mirrors the ad-hoc smoke31.py script written during development. +""" + +import tempfile +from pathlib import Path + +import pytest + +from bober_bbq.utils import supplier_import as si + +MALICIOUS_TOKENS = [ + pytest.param("__VICTIM_ABS_BACKSLASH__", id="absolute-path-backslash"), + pytest.param("__VICTIM_ABS_SLASH__", id="absolute-path-forward-slash"), + pytest.param("../../etc/passwd", id="relative-traversal-unix-style"), + pytest.param("..\\..\\windows\\win.ini", id="relative-traversal-windows-style"), + pytest.param("not-even-close-to-a-uuid.xlsx", id="wrong-shape-real-extension"), + pytest.param("", id="empty-string"), +] + + +@pytest.fixture +def victim_file(): + """A file OUTSIDE supplier_import.TMP_DIR that a malicious token must + never be able to touch — simulates a sensitive file elsewhere on the + server (an .env, another tenant's DB file, etc.).""" + victim = Path(tempfile.gettempdir()) / "test_supplier_import_victim.txt" + victim.write_text("do not delete me") + yield victim + if victim.exists(): + victim.unlink() + + +def _resolve_token(raw_token, victim_file): + # The absolute-path variants need the actual victim path at test-run + # time (tempdir differs per machine), so they're built here rather + # than baked into the parametrize table above. + if raw_token == "__VICTIM_ABS_BACKSLASH__": + return str(victim_file) + if raw_token == "__VICTIM_ABS_SLASH__": + return str(victim_file).replace("\\", "/") + return raw_token + + +@pytest.mark.parametrize("raw_token", MALICIOUS_TOKENS) +def test_discard_upload_rejects_malicious_token_without_touching_victim(app, victim_file, raw_token): + token = _resolve_token(raw_token, victim_file) + with app.app_context(): + si.discard_upload(token) # must be a silent no-op for a bad token + assert victim_file.exists(), f"discard_upload({token!r}) must not delete a file outside TMP_DIR" + assert victim_file.read_text() == "do not delete me" + + +@pytest.mark.parametrize("raw_token", MALICIOUS_TOKENS) +def test_load_preview_rejects_malicious_token_with_file_not_found(app, victim_file, raw_token): + token = _resolve_token(raw_token, victim_file) + with app.app_context(), pytest.raises(FileNotFoundError): + si.load_preview(token) + + +@pytest.mark.parametrize("raw_token", MALICIOUS_TOKENS) +def test_parse_rows_rejects_malicious_token_with_file_not_found(app, victim_file, raw_token): + token = _resolve_token(raw_token, victim_file) + with app.app_context(), pytest.raises(FileNotFoundError): + si.parse_rows(token, "Sheet1", 0, {}) + + +def test_legitimate_token_round_trips_normally(app): + class FakeFileStorage: + filename = "test.csv" + + def save(self, path): + Path(path).write_bytes("Товар;Кількість;Ціна\nЦукор;5;20\n".encode("utf-8-sig")) + + with app.app_context(): + token = si.save_upload(FakeFileStorage()) + assert si._TOKEN_RE.match(token) is not None + + preview = si.load_preview(token) + assert preview["row_count"] >= 1 + + si.discard_upload(token) + assert not (si.TMP_DIR / token).exists()