DevOps/www
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Levels

Browse Source
This commit is contained in:
eric
2026-10-27 15:56:06 -04:00
parent d01f4a8fb7
commit aa18f6b895
15 changed files with 243 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
membership/members/tests/test_membership.py
+17 -8
View File
@@ -22,8 +22,8 @@ BASE = "ou=people,dc=example,dc=org"
DN = "uid=alice," + BASE
SENTINEL = "cn=sentinel,dc=example,dc=org"
TIERS = {
"supporter": {"label": "Supporter", "price_id": "price_supporter", "price_label": "$5 / month", "group_dn": GROUP},
"patron": {"label": "Patron", "price_id": "price_patron", "price_label": "$15 / month", "group_dn": OTHER_GROUP},
"supporter": {"label": "Supporter", "level": 1, "price_id": "price_supporter", "price_label": "$5 / month", "group_dn": GROUP},
"patron": {"label": "Patron", "level": 2, "price_id": "price_patron", "price_label": "$15 / month", "group_dn": OTHER_GROUP},
}
@@ -66,6 +66,7 @@ class RegistrationTests(TestCase):
self.assertContains(response, "Account created")
provision.assert_called_once_with("alice", "alice@example.org", password)
self.assertEqual(Account.objects.count(), 1)
self.assertEqual(Account.objects.get().membership_level, 0)
self.assertFalse(Subscription.objects.exists())
self.assertFalse(Registration.objects.exists())
self.assertNotIn("account_id", self.client.session)
@@ -131,18 +132,20 @@ class RegistrationTests(TestCase):
@override_settings(MEMBERSHIP_TIERS=TIERS)
class IdentityTests(TestCase):
def claims(self, **changes):
return {"iss": settings.DEX_ISSUER, "sub": "dex-alice", "preferred_username": "alice", "email_verified": True, "email": "alice@example.org", "federated_claims": {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "uuid-alice"}, **changes}
return {"iss": settings.DEX_ISSUER, "sub": "dex-alice", "preferred_username": "alice", "email_verified": True, "email": "alice@example.org", "federated_claims": {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "alice"}, **changes}
@patch("members.views.directory.identity_for_login")
def test_existing_directory_account_links_without_password(self, identity):
identity.return_value = {"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"}
first = views.link_identity(self.claims())
second = views.link_identity(self.claims())
second = views.link_identity(self.claims(sub="ldap-alice"))
self.assertEqual(first.pk, second.pk)
self.assertEqual(first.oidc_sub, "dex-alice")
first.refresh_from_db()
self.assertEqual(first.oidc_sub, "ldap-alice")
self.assertFalse(Subscription.objects.exists())
with self.assertRaises(ValueError):
views.link_identity(self.claims(sub="different-sub"))
views.link_identity(self.claims(sub="next-verified-subject"))
first.refresh_from_db()
self.assertEqual(first.oidc_sub, "next-verified-subject")
@patch("members.views.directory.identity_for_login")
def test_no_email_only_or_unverified_link(self, identity):
@@ -158,7 +161,7 @@ class IdentityTests(TestCase):
@patch("members.views.directory.identity_for_login")
def test_other_connector_or_recreated_ldap_account_rejected(self, identity):
identity.return_value = {"uid": "alice", "email": "alice@example.org", "ldap_dn": DN, "ldap_uuid": "uuid-alice"}
for federated in ({"connector_id": "github", "user_id": "uuid-alice"}, {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "different-uuid"}, {}):
for federated in ({"connector_id": "github", "user_id": "alice"}, {"connector_id": settings.DEX_CONNECTOR_ID, "user_id": "different-uid"}, {}):
with self.assertRaises(ValueError):
views.link_identity(self.claims(federated_claims=federated))
self.assertFalse(Account.objects.exists())
@@ -291,6 +294,8 @@ class ReconciliationTests(TestCase):
self.assertEqual(int(sub.paid_until.timestamp()), self.end)
self.assertEqual(sub.tier, "supporter")
self.assertEqual(ldap.call_args.args[1], {GROUP})
self.member.refresh_from_db()
self.assertEqual(self.member.membership_level, 1)
self.assertEqual(ManagedGroup.objects.count(), 2)
@patch("members.billing.reconcile_groups", return_value=DN)
@@ -345,6 +350,8 @@ class ReconciliationTests(TestCase):
factory.return_value = self.api()
billing.sync_account(self.member.pk)
self.assertEqual(ldap.call_args.args[1], set())
self.member.refresh_from_db()
self.assertEqual(self.member.membership_level, 0)
@patch("members.billing.reconcile_groups", return_value=DN)
@patch("members.billing.client", side_effect=OSError("offline"))
@@ -426,6 +433,8 @@ class ReconciliationTests(TestCase):
factory.return_value = api
billing.sync_account(self.member.pk)
self.assertEqual(ldap.call_args.args[1], {GROUP, OTHER_GROUP})
self.member.refresh_from_db()
self.assertEqual(self.member.membership_level, 2)
@override_settings(LDAP_SEARCH_BASE_DN=BASE, LDAP_GROUP_PLACEHOLDER_DN=SENTINEL)
membership/members/tests/test_postgres.py
+2 -2
View File
@@ -46,7 +46,7 @@ class PostgreSQLLockTests(TransactionTestCase):
future.result(timeout=10)
self.assertTrue(acquired.is_set())
@override_settings(MEMBERSHIP_TIERS={"test": {"group_dn": "cn=paid,dc=example,dc=org"}})
@override_settings(MEMBERSHIP_TIERS={"test": {"level": 1, "group_dn": "cn=paid,dc=example,dc=org"}})
def test_intent_commits_before_external_grant_failure(self):
with patch("members.billing.reconcile_groups", side_effect=OSError("directory unavailable")):
with self.assertRaises(OSError):
@@ -55,7 +55,7 @@ class PostgreSQLLockTests(TransactionTestCase):
# not the previously committed record of the managed group.
self.assertEqual(ManagedGroup.objects.get(account=self.account).dn, "cn=paid,dc=example,dc=org")
@override_settings(SHKEEPER_ENABLED=True, MEMBERSHIP_TIERS={"test": {"group_dn": "cn=paid,dc=example,dc=org"}})
@override_settings(SHKEEPER_ENABLED=True, MEMBERSHIP_TIERS={"test": {"level": 1, "group_dn": "cn=paid,dc=example,dc=org"}})
def test_concurrent_crypto_verification_credits_one_month(self):
invoice = CryptoInvoice.objects.create(account=self.account, tier="test", usd_amount="5.00", crypto="BTC", provider_id="61", wallet="test-wallet", crypto_amount="0.0001", status="UNPAID")
data = {"status": "success", "invoices": [{"external_id": f"www-membership-{invoice.pk}", "fiat": "USD", "amount_fiat": "5.00", "balance_fiat": "5.00", "status": "PAID", "txs": [{"addr": "test-wallet", "crypto": "BTC", "status": "CONFIRMED", "txid": "tx-confirmed", "amount": "0.0001"}]}]}
membership/members/tests/test_shkeeper.py
+4
View File
@@ -173,12 +173,16 @@ class SHKeeperTests(TestCase):
request.return_value = self.remote(invoice)
self.assertTrue(billing.sync_account(self.member.pk))
self.assertEqual(ldap.call_args.args[1], {GROUP})
self.member.refresh_from_db()
self.assertEqual(self.member.membership_level, 1)
request.side_effect = OSError("SHKeeper offline")
self.assertFalse(billing.sync_account(self.member.pk))
self.assertEqual(ldap.call_args.args[1], {GROUP})
CryptoInvoice.objects.update(access_end=timezone.now() - timedelta(seconds=1))
billing.sync_account(self.member.pk, refresh_stripe=False)
self.assertEqual(ldap.call_args.args[1], set())
self.member.refresh_from_db()
self.assertEqual(self.member.membership_level, 0)
@patch("members.billing.reconcile_groups", return_value=DN)
@patch("members.shkeeper.request")