Janady13/AI_Brain / tests/test_smoke.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / tests/test_smoke.py
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	cdb497ece2d1a770ca8c83eb1de8923cfaa9eceb
Branch	feature/seeder-engine-tests
Path or archive member	tests/test_smoke.py
Original SHA-256	6a48dd0e1cd5586d5cc1f283f05caa3dccb1c6024e65c584278ea910c5fb15a9
Git blob SHA	ee7d73e37ffe3f17b38c6a75c74c27c54d3f6d3d
Reading-copy SHA-256	6a48dd0e1cd5586d5cc1f283f05caa3dccb1c6024e65c584278ea910c5fb15a9

Original source ↗

Recovered source transcript

Download complete source text ↓108 lines · 3.6 KB

#!/usr/bin/env python3
import importlib
import os
import sys

import pytest

# Ensure project root on sys.path for module imports like 'services.*' and 'stripe_worker'
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
if ROOT not in sys.path:
    sys.path.insert(0, ROOT)


def _get_testclient():
    """Import FastAPI TestClient ensuring stdlib 'queue' is used, not local queue.py.
    Starlette depends on the stdlib module. Our repo has queue.py at CWD.
    """
    first = sys.path[0] if sys.path else None
    cwd = None
    if first in ("", os.getcwd()):
        cwd = sys.path.pop(0)
    try:
        from fastapi.testclient import TestClient  # type: ignore
        return TestClient
    finally:
        if cwd is not None:
            sys.path.insert(0, cwd)


def test_import_seeder_intelligence():
    # Guard importability of the intelligence module; API surface may evolve.
    importlib.import_module("services.seeder_api.intelligence")


def test_docs_routes():
    TestClient = _get_testclient()
    mod = importlib.import_module("services.seeder_api.app")
    app = getattr(mod, "app")
    client = TestClient(app)
    r = client.get("/docs")
    assert r.status_code == 200
    for path in ("/healthz", "/readyz"):
        resp = client.get(path)
        assert resp.status_code in (200, 404)


def test_stripe_worker_routes(monkeypatch, tmp_path):
    # Ensure required env present at import time
    monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_dummy")
    monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_dummy")

    # Provide a minimal stub for the stripe SDK
    class _Webhook:
        @staticmethod
        def construct_event(raw, sig, secret):
            raise Exception("bad signature")

    class _Session:
        @staticmethod
        def create(**kwargs):
            return {"id": "cs_test", "url": "https://example.com/checkout/cs_test"}

    class _Checkout:
        Session = _Session

    import sys as _sys
    fake_stripe = type("_Stripe", (), {"Webhook": _Webhook, "checkout": _Checkout, "api_key": ""})
    monkeypatch.setitem(_sys.modules, "stripe", fake_stripe)

    TestClient = _get_testclient()
    import importlib
    sw = importlib.import_module("stripe_worker")
    client = TestClient(sw.app)

    # Docs should load
    assert client.get("/docs").status_code == 200
    # Health should be OK
    assert client.get("/healthz").status_code == 200
    # Webhook exists and returns 400 on bad signature
    resp = client.post("/webhook/stripe", data=b"{}")
    assert resp.status_code == 400

    # Successful webhook path writes to local revenue log if event validated
    # Re-stub construct_event to return a completed checkout session
    class _Evt:
        def __init__(self):
            self.type = "checkout.session.completed"
            self.data = {"object": {"id": "cs_test_123", "amount_total": 4200, "customer_email": "a@b.com"}}

    def _ok_construct(raw, sig, secret):
        return _Evt()

    fake_stripe.Webhook.construct_event = staticmethod(_ok_construct)
    # Write to temp file
    import os
    events_path = str(tmp_path / "rev.jsonl")
    monkeypatch.setenv("REVENUE_EVENTS_PATH", events_path)
    resp = client.post("/webhook/stripe", data=b"{}")
    assert resp.status_code == 200
    assert os.path.isfile(events_path)
    assert "cs_test_123" in open(events_path).read()

    # Checkout session creation with stub
    payload = {"amount": 1200, "currency": "usd", "name": "Test Product"}
    resp = client.post("/stripe/checkout_session", json=payload)
    assert resp.status_code == 200
    assert resp.json().get("url")