jina-hubble-sdk/tests/integration/payment/test_paying_user.py at 0519bd118b8cdce0cdd707c7d63ab29780627928 · jina-ai/jina-hubble-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import time
from datetime import datetime

import pytest
from dateutil.relativedelta import relativedelta
from mock import patch  # noqa: F401

INTERNAL_APP_ID = 'hubble-sdk'
INTERNAL_PRODUCT_ID = 'hubble-sdk'

PRICE_STRIPE_ID = 'price_1MTl37AkuPxeor9kLZxJ5lfd'

PAYING_USER_ID_1 = '63d754c6234f12b36dbd827f'
PAYING_USER_EMAIL_1 = '[email protected]'

PAYING_USER_ID_2 = '63d754de234f12b36dbd8580'
PAYING_USER_EMAIL_2 = '[email protected]'


@pytest.mark.parametrize('user_token', [PAYING_USER_ID_1], indirect=True)
def test_get_summary(stripe_client, payment_client, user_token):

    # creating stripe customer for user
    customer = stripe_client.get_customer(
        email=PAYING_USER_EMAIL_1, payment_method='pm_card_visa'
    )

    summary = payment_client.get_summary(token=user_token, app_id=INTERNAL_APP_ID)
    expected_result = {'subscriptionItems': [], 'hasPaymentMethod': True}
    assert summary['data'] == expected_result

    # creating subscription
    stripe_client.create_subscription(
        customer_id=customer['customer_id'], items=[PRICE_STRIPE_ID]
    )

    summary = payment_client.get_summary(token=user_token, app_id=INTERNAL_APP_ID)

    expected_result = {
        'subscriptionItems': [
            {
                'internalAppId': INTERNAL_APP_ID,
                'internalProductId': INTERNAL_PRODUCT_ID,
                'usageQuantity': 0,
            }
        ],
        'hasPaymentMethod': True,
    }

    assert summary['data'] == expected_result


@pytest.mark.parametrize('user_token', [PAYING_USER_ID_2], indirect=True)
def test_submit_usage_report(stripe_client, payment_client, user_token):

    # try to submit a usage report
    customer = stripe_client.get_customer(
        email=PAYING_USER_EMAIL_2, payment_method='pm_card_visa'
    )

    stripe_client.create_subscription(
        customer_id=customer['customer_id'], items=[PRICE_STRIPE_ID]
    )

    payment_client.report_usage(
        token=user_token,
        app_id=INTERNAL_APP_ID,
        product_id=INTERNAL_PRODUCT_ID,
        quantity=100,
    )

    # NOTE: sleeping to wait for the usage report to be processed
    time.sleep(75)

    summary = payment_client.get_summary(token=user_token, app_id=INTERNAL_APP_ID)

    expected_result = {
        'subscriptionItems': [
            {
                'internalAppId': INTERNAL_APP_ID,
                'internalProductId': INTERNAL_PRODUCT_ID,
                'usageQuantity': 100,
            }
        ],
        'hasPaymentMethod': True,
    }

    assert summary['data'] == expected_result

    # advancing test clock by one month
    # this will trigger a new subscription period
    now = datetime.now()
    later = now + relativedelta(days=+35)
    stripe_client.advance_clock(test_clock_id=customer['test_clock_id'], date=later)

    summary = payment_client.get_summary(token=user_token, app_id=INTERNAL_APP_ID)

    expected_result = {
        'subscriptionItems': [
            {
                'internalAppId': INTERNAL_APP_ID,
                'internalProductId': INTERNAL_PRODUCT_ID,
                'usageQuantity': 0,
            }
        ],
        'hasPaymentMethod': True,
    }

    assert summary['data'] == expected_result


@pytest.mark.parametrize(
    'user_token', [PAYING_USER_ID_1, PAYING_USER_ID_2], indirect=True
)
def test_get_authorized_jwt(payment_client, user_token):
    jwt = payment_client.get_authorized_jwt(token=user_token)['data']
    is_authorized = payment_client.verify_authorized_jwt(token=jwt)
    assert is_authorized is True