update
This commit is contained in:
386
AzA march 2026/tests/test_wc_period_phase1f.py
Normal file
386
AzA march 2026/tests/test_wc_period_phase1f.py
Normal file
@@ -0,0 +1,386 @@
|
||||
# Tests Phase 1f: Woo-Bridge Perioden ohne /wc/v3/subscriptions
|
||||
from __future__ import annotations
|
||||
|
||||
import gc
|
||||
import os
|
||||
import sqlite3
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import stripe_routes
|
||||
import wc_routes
|
||||
from wc_period_payload import coerce_next_payment_unix, resolve_wc_period_fields
|
||||
|
||||
|
||||
def _safe_unlink(path: str) -> None:
|
||||
gc.collect()
|
||||
try:
|
||||
if os.path.isfile(path):
|
||||
os.unlink(path)
|
||||
except PermissionError:
|
||||
pass
|
||||
|
||||
|
||||
class TestResolveWcPeriodFields(unittest.TestCase):
|
||||
def test_explicit_pair(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(current_period_start=1700000000, current_period_end=1702592000)
|
||||
self.assertEqual(ps, 1700000000)
|
||||
self.assertEqual(pe, 1702592000)
|
||||
|
||||
def test_reject_start_ge_end(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(current_period_start=1702592000, current_period_end=1700000000)
|
||||
self.assertIsNone(ps)
|
||||
self.assertIsNone(pe)
|
||||
|
||||
def test_next_payment_flexible_billing_m(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(
|
||||
next_payment_date="2030-06-15T00:00:00+00:00",
|
||||
billing_period="M",
|
||||
billing_interval=1,
|
||||
)
|
||||
self.assertIsNotNone(ps)
|
||||
self.assertIsNotNone(pe)
|
||||
assert ps is not None and pe is not None
|
||||
self.assertLess(ps, pe)
|
||||
|
||||
def test_next_payment_monthly_word(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(
|
||||
next_payment_date="2030-06-15T00:00:00+00:00",
|
||||
billing_period="month",
|
||||
billing_interval=1,
|
||||
)
|
||||
self.assertIsNotNone(ps)
|
||||
self.assertIsNotNone(pe)
|
||||
assert ps is not None and pe is not None
|
||||
self.assertLess(ps, pe)
|
||||
|
||||
def test_next_payment_unix_int(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(
|
||||
next_payment_date=1893456000,
|
||||
billing_period="month",
|
||||
billing_interval=1,
|
||||
)
|
||||
self.assertIsNotNone(ps)
|
||||
self.assertIsNotNone(pe)
|
||||
|
||||
def test_coerce_ms(self) -> None:
|
||||
u = coerce_next_payment_unix(1893456000000)
|
||||
self.assertEqual(u, 1893456000)
|
||||
|
||||
def test_only_end_with_billing(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(
|
||||
current_period_end=2000000000,
|
||||
billing_period="month",
|
||||
billing_interval=1,
|
||||
)
|
||||
self.assertIsNotNone(ps)
|
||||
self.assertIsNotNone(pe)
|
||||
assert ps is not None and pe is not None
|
||||
def test_only_end_with_billing_m(self) -> None:
|
||||
ps, pe = resolve_wc_period_fields(
|
||||
current_period_end=2000000000,
|
||||
billing_period="M",
|
||||
billing_interval=1,
|
||||
)
|
||||
self.assertIsNotNone(ps)
|
||||
self.assertIsNotNone(pe)
|
||||
assert ps is not None and pe is not None
|
||||
self.assertLess(ps, pe)
|
||||
|
||||
|
||||
def _make_client(tmp_db: str, events_log: str) -> TestClient:
|
||||
stripe_routes.DB_PATH = __import__("pathlib").Path(tmp_db)
|
||||
stripe_routes.EVENTS_LOG = __import__("pathlib").Path(events_log)
|
||||
wc_routes.DB_PATH = tmp_db
|
||||
stripe_routes._ensure_storage()
|
||||
app = FastAPI()
|
||||
app.include_router(wc_routes.router, prefix="/wc")
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
class TestWcProvisionPeriods(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._old_secret = os.environ.get("WC_PROVISION_SECRET")
|
||||
os.environ["WC_PROVISION_SECRET"] = "unit_test_wc_secret_1f"
|
||||
|
||||
def tearDown(self) -> None:
|
||||
if self._old_secret is None:
|
||||
os.environ.pop("WC_PROVISION_SECRET", None)
|
||||
else:
|
||||
os.environ["WC_PROVISION_SECRET"] = self._old_secret
|
||||
wc_routes.DB_PATH = None
|
||||
|
||||
def test_provision_without_period_backward_compat(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
c = _make_client(dbp, elog)
|
||||
body = {
|
||||
"customer_email": "buyer@example.com",
|
||||
"wc_order_id": 9001,
|
||||
"wc_subscription_id": 8001,
|
||||
"lookup_key": "aza_basic_monthly",
|
||||
}
|
||||
with patch("stripe_routes.send_license_email", return_value=True):
|
||||
r = c.post("/wc/provision", json=body, headers={"X-WC-Secret": "unit_test_wc_secret_1f"})
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
data = r.json()
|
||||
self.assertEqual(data["status"], "provisioned")
|
||||
self.assertNotIn("period_synced", data)
|
||||
con = sqlite3.connect(dbp)
|
||||
row = con.execute(
|
||||
"SELECT current_period_start, current_period_end FROM licenses WHERE subscription_id=?",
|
||||
("wc_sub_8001",),
|
||||
).fetchone()
|
||||
con.close()
|
||||
self.assertIsNotNone(row)
|
||||
self.assertIsNone(row[0])
|
||||
self.assertIsNone(row[1])
|
||||
del c
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
def test_provision_with_period_fields(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
c = _make_client(dbp, elog)
|
||||
body = {
|
||||
"customer_email": "buyer2@example.com",
|
||||
"wc_order_id": 9002,
|
||||
"wc_subscription_id": 8002,
|
||||
"current_period_start": 1700000000,
|
||||
"current_period_end": 1702592000,
|
||||
}
|
||||
with patch("stripe_routes.send_license_email", return_value=True):
|
||||
r = c.post("/wc/provision", json=body, headers={"X-WC-Secret": "unit_test_wc_secret_1f"})
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertTrue(r.json().get("period_synced"))
|
||||
con = sqlite3.connect(dbp)
|
||||
row = con.execute(
|
||||
"SELECT current_period_start, current_period_end, customer_email FROM licenses WHERE subscription_id=?",
|
||||
("wc_sub_8002",),
|
||||
).fetchone()
|
||||
con.close()
|
||||
self.assertEqual(row[0], 1700000000)
|
||||
self.assertEqual(row[1], 1702592000)
|
||||
del c
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
def test_idempotent_call_updates_period(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
c = _make_client(dbp, elog)
|
||||
base = {
|
||||
"customer_email": "seq@example.com",
|
||||
"wc_order_id": 9100,
|
||||
"wc_subscription_id": 8100,
|
||||
"lookup_key": "aza_basic_monthly",
|
||||
}
|
||||
with patch("stripe_routes.send_license_email", return_value=True):
|
||||
r1 = c.post("/wc/provision", json=dict(base), headers={"X-WC-Secret": "unit_test_wc_secret_1f"})
|
||||
self.assertEqual(r1.status_code, 200)
|
||||
self.assertEqual(r1.json()["status"], "provisioned")
|
||||
body2 = dict(base)
|
||||
body2["current_period_start"] = 1700000100
|
||||
body2["current_period_end"] = 1702592100
|
||||
r2 = c.post("/wc/provision", json=body2, headers={"X-WC-Secret": "unit_test_wc_secret_1f"})
|
||||
self.assertEqual(r2.status_code, 200)
|
||||
self.assertEqual(r2.json()["status"], "already_provisioned")
|
||||
self.assertTrue(r2.json().get("period_synced"))
|
||||
con = sqlite3.connect(dbp)
|
||||
row = con.execute(
|
||||
"SELECT current_period_start, current_period_end FROM licenses WHERE subscription_id=?",
|
||||
("wc_sub_8100",),
|
||||
).fetchone()
|
||||
con.close()
|
||||
self.assertEqual(row, (1700000100, 1702592100))
|
||||
del c
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
|
||||
class TestSubscriptionPeriodEndpoint(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._old_secret = os.environ.get("WC_PROVISION_SECRET")
|
||||
os.environ["WC_PROVISION_SECRET"] = "unit_test_wc_secret_1f"
|
||||
|
||||
def tearDown(self) -> None:
|
||||
if self._old_secret is None:
|
||||
os.environ.pop("WC_PROVISION_SECRET", None)
|
||||
else:
|
||||
os.environ["WC_PROVISION_SECRET"] = self._old_secret
|
||||
wc_routes.DB_PATH = None
|
||||
|
||||
def _seed_license(self, dbp: str) -> None:
|
||||
stripe_routes.DB_PATH = __import__("pathlib").Path(dbp)
|
||||
stripe_routes._ensure_storage()
|
||||
with sqlite3.connect(dbp) as con:
|
||||
con.execute(
|
||||
"""
|
||||
INSERT INTO licenses (
|
||||
subscription_id, customer_id, status, lookup_key,
|
||||
allowed_users, devices_per_user, customer_email, client_reference_id,
|
||||
current_period_start, current_period_end, updated_at, license_key
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
""",
|
||||
(
|
||||
"wc_sub_7001",
|
||||
"wc_order_6001",
|
||||
"active",
|
||||
"aza_basic_monthly",
|
||||
1,
|
||||
2,
|
||||
"u@example.com",
|
||||
"wc_order_6001",
|
||||
None,
|
||||
None,
|
||||
1,
|
||||
"LIC_TEST",
|
||||
),
|
||||
)
|
||||
con.commit()
|
||||
|
||||
def test_no_secret_401(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
self._seed_license(dbp)
|
||||
wc_routes.DB_PATH = dbp
|
||||
app = FastAPI()
|
||||
app.include_router(wc_routes.router, prefix="/wc")
|
||||
cl = TestClient(app)
|
||||
r = cl.post("/wc/subscription_period", json={"wc_subscription_id": 7001, "current_period_start": 100, "current_period_end": 200})
|
||||
self.assertEqual(r.status_code, 401)
|
||||
del cl
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
def test_wrong_secret_401(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
self._seed_license(dbp)
|
||||
wc_routes.DB_PATH = dbp
|
||||
app = FastAPI()
|
||||
app.include_router(wc_routes.router, prefix="/wc")
|
||||
cl = TestClient(app)
|
||||
r = cl.post(
|
||||
"/wc/subscription_period",
|
||||
json={"wc_subscription_id": 7001, "current_period_start": 1000, "current_period_end": 2000},
|
||||
headers={"X-WC-Secret": "wrong"},
|
||||
)
|
||||
self.assertEqual(r.status_code, 401)
|
||||
del cl
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
def test_updates_only_periods(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
self._seed_license(dbp)
|
||||
c = _make_client(dbp, elog)
|
||||
r = c.post(
|
||||
"/wc/subscription_period",
|
||||
json={
|
||||
"wc_subscription_id": 7001,
|
||||
"wc_order_id": 6001,
|
||||
"current_period_start": 1800000000,
|
||||
"current_period_end": 1802000000,
|
||||
},
|
||||
headers={"X-WC-Secret": "unit_test_wc_secret_1f"},
|
||||
)
|
||||
self.assertEqual(r.status_code, 200, r.text)
|
||||
self.assertEqual(r.json(), {"ok": True, "rows_updated": 1})
|
||||
con = sqlite3.connect(dbp)
|
||||
row = con.execute(
|
||||
"SELECT current_period_start, current_period_end, customer_email, status FROM licenses WHERE subscription_id=?",
|
||||
("wc_sub_7001",),
|
||||
).fetchone()
|
||||
con.close()
|
||||
self.assertEqual(row[0], 1800000000)
|
||||
self.assertEqual(row[1], 1802000000)
|
||||
self.assertEqual(row[2], "u@example.com")
|
||||
self.assertEqual(row[3], "active")
|
||||
del c
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
def test_response_has_no_secret(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
self._seed_license(dbp)
|
||||
c = _make_client(dbp, elog)
|
||||
r = c.post(
|
||||
"/wc/subscription_period",
|
||||
json={"wc_subscription_id": 7001, "current_period_start": 1800000000, "current_period_end": 1802000000},
|
||||
headers={"X-WC-Secret": "unit_test_wc_secret_1f"},
|
||||
)
|
||||
body = r.json()
|
||||
self.assertNotIn("secret", str(body).lower())
|
||||
del c
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
def test_unknown_subscription_404(self) -> None:
|
||||
import tempfile
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
|
||||
dbp = f.name
|
||||
elog = dbp + ".jsonl"
|
||||
try:
|
||||
self._seed_license(dbp)
|
||||
c = _make_client(dbp, elog)
|
||||
r = c.post(
|
||||
"/wc/subscription_period",
|
||||
json={
|
||||
"wc_subscription_id": 99999,
|
||||
"current_period_start": 1800000000,
|
||||
"current_period_end": 1802000000,
|
||||
},
|
||||
headers={"X-WC-Secret": "unit_test_wc_secret_1f"},
|
||||
)
|
||||
self.assertEqual(r.status_code, 404)
|
||||
del c
|
||||
finally:
|
||||
_safe_unlink(dbp)
|
||||
_safe_unlink(elog)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user