<commit_before>import datetime
from functools import wraps
from mock import Mock, patch
import operator
import os
import re
import sys
from typing import Any, Callable, Dict, List, Optional, TypeVar, Tuple
import ujson

from django.core import signing
from django.core.management import call_command
from django.http import HttpResponse
from django.utils.timezone import utc as timezone_utc

import stripe

from zerver.lib.actions import do_deactivate_user, do_create_user, \
    do_activate_user, do_reactivate_user, activity_change_requires_seat_update
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.timestamp import timestamp_to_datetime, datetime_to_timestamp
from zerver.models import Realm, UserProfile, get_realm, RealmAuditLog
from corporate.lib.stripe import catch_stripe_errors, \
    do_subscribe_customer_to_plan, attach_discount_to_realm, \
    get_seat_count, extract_current_subscription, sign_string, unsign_string, \
    get_next_billing_log_entry, run_billing_processor_one_step, \
    BillingError, StripeCardError, StripeConnectionError, stripe_get_customer
from corporate.models import Customer, Plan, Coupon, BillingProcessor

CallableT = TypeVar('CallableT', bound=Callable[..., Any])

GENERATE_STRIPE_FIXTURES = False

fixture_data_file = open(os.path.join(os.path.dirname(__file__), 'stripe_fixtures.json'), 'r')
fixture_data = ujson.load(fixture_data_file)

def mock_create_customer(*args: Any, **kwargs: Any) -> stripe.Customer:
    return stripe.util.convert_to_stripe_object(fixture_data["create_customer"])

def mock_create_subscription(*args: Any, **kwargs: Any) -> stripe.Subscription:
    return stripe.util.convert_to_stripe_object(fixture_data["create_subscription"])

def mock_customer_with_subscription(*args: Any, **kwargs: Any) -> stripe.Customer:
    return stripe.util.convert_to_stripe_object(fixture_data["customer_with_subscription"])

def mock_customer_with_canceled_subscription(*args: Any, **kwargs: Any) -> stripe.Customer:
    customer = mock_customer_with_subscription()
    customer.subscriptions.data[0].status = "canceled"
    customer.subscriptions.data[0].canceled_at = 1532602160
    return customer

def mock_customer_with_cancel_at_period_end_subscription(*args: Any, **kwargs: Any) -> stripe.Customer:  # nocoverage
    customer = mock_customer_with_subscription()
    customer.subscriptions.data[0].canceled_at = 1532602243
    customer.subscriptions.data[0].cancel_at_period_end = True
    return customer

def mock_customer_with_account_balance(account_balance: int) -> Callable[[str, List[str]], stripe.Customer]:
    def customer_with_account_balance(stripe_customer_id: str, expand: List[str]) -> stripe.Customer:
        stripe_customer = mock_customer_with_subscription()
        stripe_customer.account_balance = account_balance
        return stripe_customer
    return customer_with_account_balance

def mock_upcoming_invoice(*args: Any, **kwargs: Any) -> stripe.Invoice:
    return stripe.util.convert_to_stripe_object(fixture_data["upcoming_invoice"])

def mock_invoice_preview_for_downgrade(total: int=-1000) -> Callable[[str, str, Dict[str, Any]], stripe.Invoice]:
    def invoice_preview(customer: str, subscription: str,
                        subscription_items: Dict[str, Any]) -> stripe.Invoice:
        # TODO: Get a better fixture; this is not at all what these look like
        stripe_invoice = stripe.util.convert_to_stripe_object(fixture_data["upcoming_invoice"])
        stripe_invoice.total = total
        return stripe_invoice
    return invoice_preview

# TODO: check that this creates a token similar to what is created by our
# actual Stripe Checkout flows
def stripe_create_token(card_number: str="4242424242424242") -> stripe.Token:
    return stripe.Token.create(
        card={
            "number": card_number,
            "exp_month": 3,
            "exp_year": 2033,
            "cvc": "333",
            "name": "<NAME>",
            "address_line1": "Under the sea,",
            "address_city": "Pacific",
            "address_zip": "33333",
            "address_country": "United States",
        })

def stripe_fixture_path(decorated_function_name: str, mocked_function_name: str, call_count: int) -> str:
    # Make the eventual filename a bit shorter, and also we conventionally
    # use test_* for the python test files
    if decorated_function_name[:5] == 'test_':
        decorated_function_name = decorated_function_name[5:]
    return "corporate/tests/stripe_fixtures/{}:{}.{}.json".format(
        decorated_function_name, mocked_function_name[7:], call_count)

def generate_and_save_stripe_fixture(decorated_function_name: str, mocked_function_name: str,
                                     mocked_function: CallableT) -> Callable[[Any, Any], Any]:  # nocoverage
    def _generate_and_save_stripe_fixture(*args: Any, **kwargs: Any) -> Any:
        # Talk to Stripe
        stripe_object = mocked_function(*args, **kwargs)
        # Note that mock is not the same as mocked_function, even though their
        # definitions look the same
        mock = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        fixture_path = stripe_fixture_path(decorated_function_name, mocked_function_name, mock.call_count)
        with open(fixture_path, 'w') as f:
            f.write(str(stripe_object) + "\n")
        return stripe_object
    return _generate_and_save_stripe_fixture

def read_stripe_fixture(decorated_function_name: str,
                        mocked_function_name: str) -> Callable[[Any, Any], Any]:
    def _read_stripe_fixture(*args: Any, **kwargs: Any) -> Any:
        mock = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        fixture_path = stripe_fixture_path(decorated_function_name, mocked_function_name, mock.call_count)
        return stripe.util.convert_to_stripe_object(ujson.load(open(fixture_path, 'r')))
    return _read_stripe_fixture

def mock_stripe(mocked_function_name: str,
                generate_this_fixture: bool=False) -> Callable[[CallableT], CallableT]:
    def _mock_stripe(decorated_function: CallableT) -> CallableT:
        mocked_function = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        if GENERATE_STRIPE_FIXTURES or generate_this_fixture:
            side_effect = generate_and_save_stripe_fixture(
                decorated_function.__name__, mocked_function_name, mocked_function)  # nocoverage
        else:
            side_effect = read_stripe_fixture(decorated_function.__name__, mocked_function_name)

        @patch(mocked_function_name, side_effect=side_effect)
        @wraps(decorated_function)
        def wrapped(*args: Any, **kwargs: Any) -> Any:
            return decorated_function(*args, **kwargs)
        return wrapped
    return _mock_stripe

# A Kandra is a fictional character that can become anything. Used as a
# wildcard when testing for equality.
class Kandra(object):
    def __eq__(self, other: Any) -> bool:
        return True

class StripeTest(ZulipTestCase):
    @mock_stripe("stripe.Coupon.create")
    @mock_stripe("stripe.Plan.create")
    @mock_stripe("stripe.Product.create")
    def setUp(self, mock3: Mock, mock2: Mock, mock1: Mock) -> None:
        call_command("setup_stripe")

        # legacy
        self.token = 'token'
        # The values below should be copied from stripe_fixtures.json
        self.stripe_customer_id = 'cus_D7OT2jf5YAtZQL'
        self.customer_created = 1529990750
        self.stripe_coupon_id = Coupon.objects.get(percent_off=85).stripe_coupon_id
        self.stripe_plan_id = 'plan_Do3xCvbzO89OsR'
        self.subscription_created = 1529990751
        self.quantity = 8

        self.signed_seat_count, self.salt = sign_string(str(self.quantity))

    def get_signed_seat_count_from_response(self, response: HttpResponse) -> Optional[str]:
        match = re.search(r'name=\"signed_seat_count\" value=\"(.+)\"', response.content.decode("utf-8"))
        return match.group(1) if match else None

    def get_salt_from_response(self, response: HttpResponse) -> Optional[str]:
        match = re.search(r'name=\"salt\" value=\"(\w+)\"', response.content.decode("utf-8"))
        return match.group(1) if match else None

    @patch("corporate.lib.stripe.billing_logger.error")
    def test_catch_stripe_errors(self, mock_billing_logger_error: Mock) -> None:
        @catch_stripe_errors
        def raise_invalid_request_error() -> None:
            raise stripe.error.InvalidRequestError(
                "message", "param", "code", json_body={})
        with self.assertRaises(BillingError) as context:
            raise_invalid_request_error()
        self.assertEqual('other stripe error', context.exception.description)
        mock_billing_logger_error.assert_called()

        @catch_stripe_errors
        def raise_card_error() -> None:
            error_message = "The card number is not a valid credit card number."
            json_body = {"error": {"message": error_message}}
            raise stripe.error.CardError(error_message, "number", "invalid_number",
                                         json_body=json_body)
        with self.assertRaises(StripeCardError) as context:
            raise_card_error()
        self.assertIn('not a valid credit card', context.exception.message)
        self.assertEqual('card error', context.exception.description)
        mock_billing_logger_error.assert_called()

    @mock_stripe("stripe.Token.create")
    @mock_stripe("stripe.Customer.create")
    @mock_stripe("stripe.Subscription.create")
    @mock_stripe("stripe.Customer.retrieve")
    def test_initial_upgrade(self, mock4: Mock, mock3: Mock, mock2: Mock, mock1: Mock) -> None:
        user = self.example_user("hamlet")
        self.login(user.email)
        response = self.client_get("/upgrade/")
        self.assert_in_success_response(['We can also bill by invoice'], response)
        self.assertFalse(user.realm.has_seat_based_plan)
        self.assertNotEqual(user.realm.plan_type, Realm.STANDARD)
        self.assertFalse(Customer.objects.filter(realm=user.realm).exists())

        # Click "Make payment" in Stripe Checkout
        self.client_post("/upgrade/", {
            'stripeToken': stripe_create_token().id,
            'signed_seat_count': self.get_signed_seat_count_from_response(response),
            'salt': self.get_salt_from_response(response),
            'plan': Plan.CLOUD_ANNUAL})

        # Check that we correctly created Customer and Subscription objects in Stripe
        stripe_customer = stripe_get_customer(Customer.objects.get(realm=user.realm).stripe_customer_id)
        self.assertEqual(stripe_customer.default_source.id[:5], 'card_')
        self.assertEqual(stripe_customer.description, "zulip (Zulip Dev)")
        self.assertEqual(stripe_customer.discount, None)
        self.assertEqual(stripe_customer.email, user.email)<commit_msg>billing: Add test for billing disabled settings.<commit_after>import datetime
from functools import wraps
from mock import Mock, patch
import operator
import os
import re
import sys
from typing import Any, Callable, Dict, List, Optional, TypeVar, Tuple
import ujson

from django.core import signing
from django.core.management import call_command
from django.http import HttpResponse
from django.utils.timezone import utc as timezone_utc

import stripe

from zerver.lib.actions import do_deactivate_user, do_create_user, \
    do_activate_user, do_reactivate_user, activity_change_requires_seat_update
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.timestamp import timestamp_to_datetime, datetime_to_timestamp
from zerver.models import Realm, UserProfile, get_realm, RealmAuditLog
from corporate.lib.stripe import catch_stripe_errors, \
    do_subscribe_customer_to_plan, attach_discount_to_realm, \
    get_seat_count, extract_current_subscription, sign_string, unsign_string, \
    get_next_billing_log_entry, run_billing_processor_one_step, \
    BillingError, StripeCardError, StripeConnectionError, stripe_get_customer
from corporate.models import Customer, Plan, Coupon, BillingProcessor

CallableT = TypeVar('CallableT', bound=Callable[..., Any])

GENERATE_STRIPE_FIXTURES = False

fixture_data_file = open(os.path.join(os.path.dirname(__file__), 'stripe_fixtures.json'), 'r')
fixture_data = ujson.load(fixture_data_file)

def mock_create_customer(*args: Any, **kwargs: Any) -> stripe.Customer:
    return stripe.util.convert_to_stripe_object(fixture_data["create_customer"])

def mock_create_subscription(*args: Any, **kwargs: Any) -> stripe.Subscription:
    return stripe.util.convert_to_stripe_object(fixture_data["create_subscription"])

def mock_customer_with_subscription(*args: Any, **kwargs: Any) -> stripe.Customer:
    return stripe.util.convert_to_stripe_object(fixture_data["customer_with_subscription"])

def mock_customer_with_canceled_subscription(*args: Any, **kwargs: Any) -> stripe.Customer:
    customer = mock_customer_with_subscription()
    customer.subscriptions.data[0].status = "canceled"
    customer.subscriptions.data[0].canceled_at = 1532602160
    return customer

def mock_customer_with_cancel_at_period_end_subscription(*args: Any, **kwargs: Any) -> stripe.Customer:  # nocoverage
    customer = mock_customer_with_subscription()
    customer.subscriptions.data[0].canceled_at = 1532602243
    customer.subscriptions.data[0].cancel_at_period_end = True
    return customer

def mock_customer_with_account_balance(account_balance: int) -> Callable[[str, List[str]], stripe.Customer]:
    def customer_with_account_balance(stripe_customer_id: str, expand: List[str]) -> stripe.Customer:
        stripe_customer = mock_customer_with_subscription()
        stripe_customer.account_balance = account_balance
        return stripe_customer
    return customer_with_account_balance

def mock_upcoming_invoice(*args: Any, **kwargs: Any) -> stripe.Invoice:
    return stripe.util.convert_to_stripe_object(fixture_data["upcoming_invoice"])

def mock_invoice_preview_for_downgrade(total: int=-1000) -> Callable[[str, str, Dict[str, Any]], stripe.Invoice]:
    def invoice_preview(customer: str, subscription: str,
                        subscription_items: Dict[str, Any]) -> stripe.Invoice:
        # TODO: Get a better fixture; this is not at all what these look like
        stripe_invoice = stripe.util.convert_to_stripe_object(fixture_data["upcoming_invoice"])
        stripe_invoice.total = total
        return stripe_invoice
    return invoice_preview

# TODO: check that this creates a token similar to what is created by our
# actual Stripe Checkout flows
def stripe_create_token(card_number: str="4242424242424242") -> stripe.Token:
    return stripe.Token.create(
        card={
            "number": card_number,
            "exp_month": 3,
            "exp_year": 2033,
            "cvc": "333",
            "name": "<NAME>",
            "address_line1": "Under the sea,",
            "address_city": "Pacific",
            "address_zip": "33333",
            "address_country": "United States",
        })

def stripe_fixture_path(decorated_function_name: str, mocked_function_name: str, call_count: int) -> str:
    # Make the eventual filename a bit shorter, and also we conventionally
    # use test_* for the python test files
    if decorated_function_name[:5] == 'test_':
        decorated_function_name = decorated_function_name[5:]
    return "corporate/tests/stripe_fixtures/{}:{}.{}.json".format(
        decorated_function_name, mocked_function_name[7:], call_count)

def generate_and_save_stripe_fixture(decorated_function_name: str, mocked_function_name: str,
                                     mocked_function: CallableT) -> Callable[[Any, Any], Any]:  # nocoverage
    def _generate_and_save_stripe_fixture(*args: Any, **kwargs: Any) -> Any:
        # Talk to Stripe
        stripe_object = mocked_function(*args, **kwargs)
        # Note that mock is not the same as mocked_function, even though their
        # definitions look the same
        mock = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        fixture_path = stripe_fixture_path(decorated_function_name, mocked_function_name, mock.call_count)
        with open(fixture_path, 'w') as f:
            f.write(str(stripe_object) + "\n")
        return stripe_object
    return _generate_and_save_stripe_fixture

def read_stripe_fixture(decorated_function_name: str,
                        mocked_function_name: str) -> Callable[[Any, Any], Any]:
    def _read_stripe_fixture(*args: Any, **kwargs: Any) -> Any:
        mock = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        fixture_path = stripe_fixture_path(decorated_function_name, mocked_function_name, mock.call_count)
        return stripe.util.convert_to_stripe_object(ujson.load(open(fixture_path, 'r')))
    return _read_stripe_fixture

def mock_stripe(mocked_function_name: str,
                generate_this_fixture: bool=False) -> Callable[[CallableT], CallableT]:
    def _mock_stripe(decorated_function: CallableT) -> CallableT:
        mocked_function = operator.attrgetter(mocked_function_name)(sys.modules[__name__])
        if GENERATE_STRIPE_FIXTURES or generate_this_fixture:
            side_effect = generate_and_save_stripe_fixture(
                decorated_function.__name__, mocked_function_name, mocked_function)  # nocoverage
        else:
            side_effect = read_stripe_fixture(decorated_function.__name__, mocked_function_name)

        @patch(mocked_function_name, side_effect=side_effect)
        @wraps(decorated_function)
        def wrapped(*args: Any, **kwargs: Any) -> Any:
            return decorated_function(*args, **kwargs)
        return wrapped
    return _mock_stripe

# A Kandra is a fictional character that can become anything. Used as a
# wildcard when testing for equality.
class Kandra(object):
    def __eq__(self, other: Any) -> bool:
        return True

class StripeTest(ZulipTestCase):
    @mock_stripe("stripe.Coupon.create")
    @mock_stripe("stripe.Plan.create")
    @mock_stripe("stripe.Product.create")
    def setUp(self, mock3: Mock, mock2: Mock, mock1: Mock) -> None:
        call_command("setup_stripe")

        # legacy
        self.token = 'token'
        # The values below should be copied from stripe_fixtures.json
        self.stripe_customer_id = 'cus_D7OT2jf5YAtZQL'
        self.customer_created = 1529990750
        self.stripe_coupon_id = Coupon.objects.get(percent_off=85).stripe_coupon_id
        self.stripe_plan_id = 'plan_Do3xCvbzO89OsR'
        self.subscription_created = 1529990751
        self.quantity = 8

        self.signed_seat_count, self.salt = sign_string(str(self.quantity))

    def get_signed_seat_count_from_response(self, response: HttpResponse) -> Optional[str]:
        match = re.search(r'name=\"signed_seat_count\" value=\"(.+)\"', response.content.decode("utf-8"))
        return match.group(1) if match else None

    def get_salt_from_response(self, response: HttpResponse) -> Optional[str]:
        match = re.search(r'name=\"salt\" value=\"(\w+)\"', response.content.decode("utf-8"))
        return match.group(1) if match else None

    @patch("corporate.lib.stripe.billing_logger.error")
    def test_catch_stripe_errors(self, mock_billing_logger_error: Mock) -> None:
        @catch_stripe_errors
        def raise_invalid_request_error() -> None:
            raise stripe.error.InvalidRequestError(
                "message", "param", "code", json_body={})
        with self.assertRaises(BillingError) as context:
            raise_invalid_request_error()
        self.assertEqual('other stripe error', context.exception.description)
        mock_billing_logger_error.assert_called()

        @catch_stripe_errors
        def raise_card_error() -> None:
            error_message = "The card number is not a valid credit card number."
            json_body = {"error": {"message": error_message}}
            raise stripe.error.CardError(error_message, "number", "invalid_number",
                                         json_body=json_body)
        with self.assertRaises(StripeCardError) as context:
            raise_card_error()
        self.assertIn('not a valid credit card', context.exception.message)
        self.assertEqual('card error', context.exception.description)
        mock_billing_logger_error.assert_called()

    def test_billing_not_enabled(self) -> None:
        with self.settings(BILLING_ENABLED=False):
            self.login(self.example_email("iago"))
            response = self.client_get("/upgrade/")
            self.assert_in_success_response(["Page not found (404)"], response)

    @mock_stripe("stripe.Token.create")
    @mock_stripe("stripe.Customer.create")
    @mock_stripe("stripe.Subscription.create")
    @mock_stripe("stripe.Customer.retrieve")
    def test_initial_upgrade(self, mock4: Mock, mock3: Mock, mock2: Mock, mock1: Mock) -> None:
        user = self.example_user("hamlet")
        self.login(user.email)
        response = self.client_get("/upgrade/")
        self.assert_in_success_response(['We can also bill by invoice'], response)
        self.assertFalse(user.realm.has_seat_based_plan)
        self.assertNotEqual(user.realm.plan_type, Realm.STANDARD)
        self.assertFalse(Customer.objects.filter(realm=user.realm).exists())

        # Click "Make payment" in Stripe Checkout
        self.client_post("/upgrade/", {
            'stripeToken': stripe_create_token().id,
            'signed_seat_count': self.get_signed_seat_count_from_response(response),
            'salt': self.get_salt_from_response(response),
            'plan': Plan.CLOUD_ANNUAL})

        # Check that we correctly created Customer and Subscription objects in Stripe
        stripe_customer = stripe_get_customer(Customer.objects.get(realm=user.realm).stripe_customer_id)
        self.assertEqual(stripe_customer.default_source.id[:5], 'card_')
        self.assertEqual(stripe_customer.description, "zulip (Zulip Dev)")
        self.assertEqual(stripe_customer.discount, None)
        self.assertEqual(stripe_customer.email, user.email)
{'fraction_non_alphanumeric': 0.09820218628602403, 'fraction_numerical': 0.009350438160628784, 'mean_word_length': 3.8936781609195403, 'pattern_counts': {'":': 22, '<': 5, '<?xml version=': 0, '>': 60, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5806418', 'n_tokens_mistral': 6929, 'n_tokens_neox': 6673, 'n_words': 1354}