manyi/testcase/unit/conftest.py

# -*- coding: utf-8 -*-
# !/usr/bin/env python

import atexit
import os
import shutil
import tempfile

import pytest
from django.http import HttpRequest
from mockredis import MockRedis
from typing import List, Any

from apilib.utils_json import JsonResponse
from apps.web.ad.models import Advertisement
from apps.web.agent.models import Agent
from apps.web.common.models import WithdrawRecord, Banks
from apps.web.common.transaction import WithdrawStatus
from apps.web.core import ROLE
from apps.web.core.models import WechatAuthApp, WechatPayApp, AliApp, WechatManagerApp,  \
    WechatUserManagerApp
from apps.web.core.payment.wechat import WechatPaymentGateway
from apps.web.dealer.define import DEALER_INCOME_TYPE
from apps.web.dealer.models import Dealer, OnSale, OnSaleRecord
from apps.web.device.models import Device, Group, DeviceType
from apps.web.management.models import Manager
from apps.web.superadmin.models import SuperManager
from apps.web.user.models import MyUser
from testcase.providers import generate_dict
from .base import skip_if_no_django, RequestTestClient, WechatRequestTestClient, AlipayRequestTestClient
from .common import *

HOME_URL = 'http://127.0.0.1:8080/'

URLS = {
    "dealer": {
        "login": "app/service/dealerLogin"
    }
}


### Exceptions
class FixtureCreationError(Exception):
    pass


### Resources
@atexit.register
def delete_all_data():
    pass


@pytest.fixture
def temp_dir():
    dir_path = tempfile.mkdtemp()
    yield dir_path
    shutil.rmtree(dir_path)


@pytest.fixture(scope = 'session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'redis://'
    }


@pytest.fixture(scope = 'function')
def credentials(request):
    import simplejson as json
    path = os.path.join(request.node.fspath.dirpath().join('..').strpath, 'data', 'credentials.json')
    with open(path, 'r') as f:
        return json.load(f)


@pytest.fixture(scope = 'session')
def mocked_redis():
    """
    mocked redis-py client
    :return:
    """
    return MockRedis()


@pytest.fixture
def http_request():
    """
    mocked django HttpRequest
    :return:
    """
    return HttpRequest()


@pytest.fixture(scope = 'session')
def http_json_response(dict_):
    """
    mocked django JsonResponse
    :param dict_:
    :return:
    """
    return JsonResponse(dict_)


## wechat related
@pytest.fixture(scope = 'function')
def wechat_auth_app():
    yield WechatAuthApp(appid = WECHAT_CONFIG.appid, secret = WECHAT_CONFIG.secret)


@pytest.fixture(scope = 'function')
def wechat_payment_app():
    WechatPayApp.objects(appid = WECHAT_CONFIG.appid, mchid = WECHAT_CONFIG.mchid).delete()
    app = WechatPayApp(
        appid = WECHAT_CONFIG.appid,
        secret = WECHAT_CONFIG.secret,
        mchid = WECHAT_CONFIG.mchid,
        apikey = WECHAT_CONFIG.apikey,
        sslCert = WECHAT_CONFIG.sslCert,
        sslKey = WECHAT_CONFIG.sslKey,
        manual_withdraw = WECHAT_CONFIG.manual_withdraw)
    app.save()
    yield app


@pytest.fixture(scope = 'function')
def wechat_managerial_app():
    yield WechatManagerApp.create(appid = WECHAT_CONFIG.appid, secret = WECHAT_CONFIG.secret)


@pytest.fixture(scope = 'function')
def wechat_user_managerial_app():
    yield WechatUserManagerApp.create(appid = WECHAT_CONFIG.appid, secret = WECHAT_CONFIG.secret)


## alipay related
@pytest.fixture(scope = 'function')
def alipay_payment_app():
    AliApp.objects(appid = ALIPAY_CONFIG.appid).delete()
    app = AliApp(appid = ALIPAY_CONFIG.appid,
                 public_key_path = ALIPAY_CONFIG.alipayPublicKey,
                 app_private_key_path = ALIPAY_CONFIG.appPrivateKey)
    app.save()
    yield app


@pytest.fixture(scope = 'session', autouse = True)
def default_wechat_auth_app():
    yield WechatAuthApp(appid = DEFAULT_WECHAT_CONFIG.appid, secret = DEFAULT_WECHAT_CONFIG.secret)


@pytest.fixture(scope = 'session', autouse = True)
def default_wechat_payment_app():
    WechatPayApp.objects(appid = DEFAULT_WECHAT_CONFIG.appid, mchid = DEFAULT_WECHAT_CONFIG.mchid).delete()

    app = WechatPayApp(
        appid = DEFAULT_WECHAT_CONFIG.appid,
        secret = DEFAULT_WECHAT_CONFIG.secret,
        mchid = DEFAULT_WECHAT_CONFIG.mchid,
        apikey = DEFAULT_WECHAT_CONFIG.apikey,
        sslCert = DEFAULT_WECHAT_CONFIG.sslCert,
        sslKey = DEFAULT_WECHAT_CONFIG.sslKey,
        manual_withdraw = DEFAULT_WECHAT_CONFIG.manual_withdraw)
    app.save()

    yield app


@pytest.fixture(scope = 'session', autouse = True)
def default_wechat_managerial_app():
    yield WechatManagerApp.create(appid = DEFAULT_WECHAT_CONFIG.appid, secret = DEFAULT_WECHAT_CONFIG.secret)


@pytest.fixture(scope = 'session', autouse = True)
def default_wechat_user_managerial_app():
    yield WechatUserManagerApp.create(appid = DEFAULT_WECHAT_CONFIG.appid, secret = DEFAULT_WECHAT_CONFIG.secret)


## alipay related
@pytest.fixture(scope = 'session', autouse = True)
def default_alipay_payment_app():
    AliApp.objects(appid = DEFAULT_ALIPAY_CONFIG.appid).delete()
    app = AliApp(appid = DEFAULT_ALIPAY_CONFIG.appid,
                 public_key_path = DEFAULT_ALIPAY_CONFIG.public_key_path,
                 app_private_key_path = DEFAULT_ALIPAY_CONFIG.app_private_key_path)
    app.save()
    yield app


## Sensible Defaults
@pytest.fixture(scope = 'session', autouse = True)
def default_agent(default_wechat_auth_app, default_wechat_managerial_app, default_wechat_user_managerial_app,
                  default_wechat_payment_app, default_alipay_payment_app, default_wechat_mini_app):
    """
    需要设置一个默认的支持所有功能的代理商
    :return:
    """
    Agent.objects(id = DEFAULT_AGENT_ID).delete()

    agent = Agent(**DEFAULT_AGENT_FIXTURE)  # type: Agent
    agent.payAppWechat = default_wechat_payment_app
    agent.wechatLoginAuthApp = default_wechat_auth_app
    agent.wechatUserManagerialApp = default_wechat_user_managerial_app
    agent.wechatDealerManagerialApp = default_wechat_managerial_app
    agent.payAppAli = default_alipay_payment_app
    agent.save()

    agent.payAppWechat.occupant = agent
    agent.payAppAli.occupant = agent
    agent.wechatLoginAuthApp.occupant = agent
    agent.wechatUserManagerialApp.occupant = agent
    agent.wechatDealerManagerialApp.occupant = agent

    yield agent


@pytest.fixture(scope = 'session', autouse = True)
def default_manager():
    # type: ()->Manager
    """
    :return:
    """
    _ = Manager.objects(id = DEFAULT_MANAGER_ID).update(upsert = True, **DEFAULT_MANAGER_FIXTURE)
    if _:
        doc = Manager.objects(id = DEFAULT_MANAGER_ID).get()  # type: Manager
        doc.set_password(DEFAULT_MANAGER_FIXTURE['password'])
        yield doc

    else:
        raise FixtureCreationError('cannot create default Manager fixture, id=%s' % (str(DEFAULT_MANAGER_ID)))


### Data models

@pytest.fixture(scope = 'function')
def device(dealer):
    # type: (Any)->Device
    """
    a device dict
    :return:
    """
    Device.objects(id = DEVICE_ID).delete()
    Device.invalid_device_cache(DEVICE_FIXTURE['devNo'])

    _ = Device(**DEVICE_FIXTURE).save()  # .cached
    yield _


@pytest.fixture()
def multiple_devices():
    # type: ()->List[Device]

    yield []


@pytest.fixture(scope = 'session', autouse = True)
def device_type():
    # type: ()->DeviceType
    """

    :return:
    """
    DeviceType.objects(id = DEVICE_TYPE_ID).delete()

    _ = DeviceType(**DEVICE_TYPE_FIXTURE).save()
    yield _

@pytest.fixture(scope = 'function')
def source_key():
    # type: ()->str

    yield 'ledger-wechat-{}-1510834731'.format(AGENT_ID)


@pytest.fixture(scope = 'function')
def dealer(agent_no_customized):
    # type: ()->Dealer
    """
    a Dealer object
    :return:
    """
    Dealer.objects(id = DEALER_ID).delete()
    Dealer.invalid_cache(DEALER_ID)
    _ = Dealer(**DEALER_FIXTURE).save()
    _.set_password(DEALER_FIXTURE['password'])
    _._raw_password = DEALER_FIXTURE['raw_password']
    yield _


@pytest.fixture(scope = 'session', autouse = True)
def partner():
    # type: ()->Dealer
    """

    :return:
    """
    Dealer.objects(id = PARTNER_ID).delete()
    Dealer.invalid_cache(PARTNER_ID)
    _ = Dealer(**PARTNER_FIXTURE).save()
    _.set_password(PARTNER_FIXTURE['password'])
    yield _


@pytest.fixture(scope = 'session', autouse = True)
def sole_group():
    # type: ()->Group
    _ = Group.objects(id = GROUP_ID).update(upsert = True, **SOLE_GROUP_FIXTURE)
    if _:
        doc = Group.objects(id = GROUP_ID).get()
        yield doc

    else:
        raise FixtureCreationError('cannot create GROUP_ID fixture, id=%s' % (str(GROUP_ID)))


@pytest.fixture(scope = 'session', autouse = True)
def partner_group():
    # type: ()->Group
    _ = Group.objects(id = PARTNER_GROUP_ID).update(upsert = True, **PARTNER_GROUP_FIXTURE)
    if _:
        doc = Group.objects(id = PARTNER_GROUP_ID).get()
        yield doc

    else:
        raise FixtureCreationError('cannot create PARTNER_GROUP_ID fixture, id=%s' % (str(PARTNER_GROUP_ID)))


@pytest.fixture(scope = 'function')
def agent(wechat_auth_app, wechat_payment_app, wechat_managerial_app, wechat_user_managerial_app, alipay_payment_app, manager):
    # type: (WechatAuthApp, WechatPayApp, WechatManagerApp, WechatUserManagerApp, AliApp, Manager)->Agent
    """
    :return:
    """
    Agent.objects(id = AGENT_ID).delete()

    agent = Agent(**AGENT_FIXTURE)  # type: Agent
    agent.payAppWechat = wechat_payment_app
    agent.wechatLoginAuthApp = wechat_auth_app
    agent.wechatUserManagerialApp = wechat_user_managerial_app
    agent.wechatDealerManagerialApp = wechat_managerial_app
    agent.payAppAli = alipay_payment_app
    agent.save()

    agent.payAppWechat.occupant = agent
    agent.payAppAli.occupant = agent
    agent.wechatLoginAuthApp.occupant = agent
    agent.wechatUserManagerialApp.occupant = agent
    agent.wechatDealerManagerialApp.occupant = agent

    yield agent


@pytest.fixture(scope = 'function')
def agent_no_customized(bankcard):
    # type: (*Any)->Agent
    """
    :return:
    """
    Agent.objects(id = AGENT_ID).delete()

    agent = Agent(**AGENT_NO_CUSTOMIZED_FIXTURE)  # type: Agent
    agent.save()

    yield agent


@pytest.fixture(scope = 'function')
def manager():
    # type: ()->Manager
    """
    :return:
    """
    _ = Manager.objects(id = MANAGER_ID).update(upsert = True, **MANAGER_FIXTURE)
    if _:
        doc = Manager.objects(id = MANAGER_ID).get()  # type: Manager
        doc.set_password(MANAGER_FIXTURE['password'])
        yield doc

    else:
        raise FixtureCreationError('cannot create Manager fixture, id=%s' % (str(MANAGER_ID)))


@pytest.fixture(scope = 'function')
def super_manager():
    # type: ()->SuperManager
    """

    :return:
    """

    SuperManager.objects(id = SUPER_MANAGER_ID).delete()

    doc = SuperManager(id = SUPER_MANAGER_ID, username = SUPER_MANAGER_FIXTURE['username']).save()

    doc.set_password(SUPER_MANAGER_FIXTURE['password'])

    yield doc


@pytest.fixture(scope = 'function')
def user():
    # type: ()->MyUser
    """
    :return:
    """
    _ = MyUser.objects(id = MY_USER_ID).update(upsert = True, **MY_USER_FIXTURE)
    if _:
        return MyUser.objects(id = MY_USER_ID).get()
    else:
        raise FixtureCreationError('cannot create MyUser fixture, id=%s' % (str(MY_USER_ID)))


### Clients

@pytest.fixture(scope = 'function')
def client():
    """A Django test client instance.
    :return:
    """
    skip_if_no_django()

    client = RequestTestClient()
    yield client
    client.logout()


@pytest.fixture(scope = 'function')
def wechat_client():
    client = WechatRequestTestClient()
    yield client


@pytest.fixture(scope = 'function')
def alipay_client():
    client = AlipayRequestTestClient()
    yield client


@pytest.fixture(scope = 'function')
def manager_client():
    """
    a client with manager role
    :return:
    """
    client = RequestTestClient()
    client.login_as_manager(username = MANAGER_FIXTURE['username'], password = MANAGER_FIXTURE['password'])
    yield client
    client.logout()


@pytest.fixture()
def super_manager_client():
    """

    :return:
    """
    client = RequestTestClient()
    client.login_as_super_manager(username = SUPER_MANAGER_FIXTURE['username'],
                                  password = SUPER_MANAGER_FIXTURE['password'])
    yield client
    client.logout()


@pytest.fixture(scope = 'function')
def dealer_client(dealer):
    """
    a client with dealer role
    :return:
    """
    client = RequestTestClient()
    client._re_login = lambda: client.login_as_dealer(username = DEALER_FIXTURE['username'],
                                                      password = DEALER_FIXTURE['password'],
                                                      agentId = DEALER_FIXTURE['agentId'])
    client._re_login()
    yield client
    client.logout()


@pytest.fixture(scope = 'function')
def agent_client():
    """
    a client with agent role
    :return:
    """
    client = RequestTestClient()
    client.login_as_agent(username = AGENT_FIXTURE['username'], password = AGENT_FIXTURE['password'])
    yield client
    client.logout()


@pytest.fixture(scope = 'session')
def default_agent_client(default_agent):
    """
    a client with agent role
    :return:
    """
    client = RequestTestClient()
    client.login_as_agent(username = DEFAULT_AGENT_FIXTURE['username'], password = DEFAULT_AGENT_FIXTURE['password'])
    yield client
    client.logout()


@pytest.fixture(scope = 'function')
def agent_no_customized_client(agent_no_customized):
    """
    a client with agent role
    :return:
    """
    client = RequestTestClient()
    client.login_as_agent(username = AGENT_NO_CUSTOMIZED_FIXTURE['username'],
                          password = AGENT_NO_CUSTOMIZED_FIXTURE['password'])
    yield client
    client.logout()


@pytest.fixture
def wechat_user_client(user, wechat_managerial_app, wechat_payment_app, wechat_auth_app):
    """
    a client with end user model
    :return:
    """
    client = WechatRequestTestClient()
    client.login_as_endUser(openId = user.openId, groupId = user.groupId)
    yield client
    client.logout()


@pytest.fixture
def alipay_user_client(user):
    """

    :return:
    """
    client = AlipayRequestTestClient()
    client.login_as_endUser(openId = user.openId, groupId = user.groupId)
    yield client
    client.logout()


# Others

@pytest.fixture()
def pay_after_ad():
    yield Advertisement(**PAY_AFTER_AD_FIXTURE)


@pytest.fixture()
def banner_ad():
    yield Advertisement(**BANNER_AD_FIXTURE)


@pytest.fixture()
def onsale():
    OnSaleRecord.objects().delete()
    _ = OnSale(**ON_SALE_FIXTURE).save()
    yield _
    _.delete()


@pytest.fixture()
def dealerWithdrawRecord():
    payload = generate_dict(WithdrawRecord)
    payload['phone'] = DEALER_FIXTURE['username']
    payload['name'] = DEALER_FIXTURE['nickname']
    payload['ownerId'] = DEALER_FIXTURE['id']
    payload['role'] = ROLE.dealer
    payload['incomeType'] = DEALER_INCOME_TYPE.DEVICE_INCOME
    payload['serviceFee'] = RMB('0.08')
    payload['amount'] = RMB('10')
    payload['status'] = WithdrawStatus.SUCCEEDED
    payload['order'] = WithdrawRecord.make_no(ROLE.dealer, DEALER_FIXTURE['id'])
    payload['withdrawFeeRatio'] = 8

    _ = WithdrawRecord(**payload).save()
    yield _
    _.delete()


@pytest.fixture()
def gateway_key(default_agent):
    return WechatPaymentGateway(default_agent.payAppWechat).gateway_key


@pytest.fixture()
def gateway(default_agent):
    return WechatPaymentGateway(default_agent.payAppWechat)