# -*- coding: utf-8 -*- #!/usr/bin/env python """Core Unit Test Module""" import os os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'configs.systest') import sys import pytest import inspect import simplejson as json from Cookie import SimpleCookie from importlib import import_module import requests from mongoengine import connect from django.conf import settings from django.contrib.auth import get_user, logout, login from django.http import HttpRequest from django.test.runner import DiscoverRunner from django.test import TestCase as _TestCase from django.test.client import Client from jsonpath_rw import parse from apps.web.utils import get_backend_by_role from apps.web.user.models import MyUser from apps.web.user.auth import end_user_login, end_user_login_session_key def pluck(response, field): return parse(field).find(json.loads(response.content))[0].value def pluck_description(response): return pluck(response, 'description') def pluck_dataList(response): return pluck(response, 'payload.dataList') def pluck_result(response): return pluck(response, 'result') def result_is_ok(response): return pluck_result(response) == 1 def result_is_not_ok(response): return pluck_result(response) == 0 def skip_if_no_django(): """Raises a skip exception when no Django settings are available""" if not django_settings_is_configured(): pytest.skip('no Django settings') def django_settings_is_configured(): # Avoid importing Django if it has not yet been imported if not os.environ.get('DJANGO_SETTINGS_MODULE') \ and 'django.conf' not in sys.modules: return False # If DJANGO_SETTINGS_MODULE is defined at this point, Django is assumed to # always be loaded. return True def get_django_version(): return __import__('django').VERSION #: 由于项目运用MongoDB，需要hack Django 默认的TestCase/runner #: set TEST_RUNNER = 'washpayer.tests.MongoBasedTestRunner' to configs/base.py TEST_DB = 'testsuite' WECHAT_USER_AGENT = 'Mozilla/5.0 (Linux; Android 6.0; 1503-M02 Build/MRA58K) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/37.0.0.0 Mobile MQQBrowser/6.2 TBS/036558 Safari/537.36 MicroMessenger/6.3.25.861 NetType/WIFI Language/zh_CN' ALIPAY_USER_AGENT = 'Mozilla/5.0 (Linux; U; Android 7.1.1; zh-CN; OD103 Build/NMF26F) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/40.0.2214.89 UCBrowser/11.6.4.950 UWS/2.11.0.20 Mobile Safari/537.36 UCBS/2.11.0.20 Nebula AlipayDefined(nt:WIFI,ws:360|0|3.0) AliApp(AP/10.1.2.091816) AlipayClient/10.1.2.091816 Language/en useStatusBar/true' PC_USER_AGENT = 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.78 Safari/537.36' _running_test = False Url = str class RequestTestClient(Client): _current_user = None def get(self, path, data=None, follow=False, secure=False, **extra): if callable(path): path = path() assert isinstance(path, Url), 'path has to be a string' return super(RequestTestClient, self).get(path, data=None, follow=False, secure=False, **extra) def post(self, path, data=None, content_type='application/json', follow=False, secure=False, **extra): if callable(path): path = path() assert isinstance(path, Url), 'path has to be a string' return super(RequestTestClient, self).post(path, data=data, content_type=content_type, follow=follow, secure=secure, **extra) def _login(self, login_fn, user, user_agent=WECHAT_USER_AGENT): # Create a fake request to store login details. request = HttpRequest() request.META['HTTP_USER_AGENT'] = user_agent engine = import_module(settings.SESSION_ENGINE) if self.session: request.session = self.session request.user = get_user(request) else: request.session = engine.SessionStore() login_fn(request, user) # Save the session values. request.session.save() # Set the cookie to represent the session. session_cookie = settings.SESSION_COOKIE_NAME self.cookies[session_cookie] = request.session.session_key cookie_data = { 'max-age': None, 'path': '/', 'domain': settings.COOKIE_DOMAIN, 'secure': False, 'expires': None, } self.cookies[session_cookie].update(cookie_data) self.current_user = user return True def current_user(self, role=None): return self._current_user or self.login_as(role) def login_as(self, role, **credentials): backend = get_backend_by_role(role)() credentials['password'] = credentials['password'] user = backend.authenticate(**credentials) assert user is not None, u'could not find user, cannot login, credentials=%s' % (json.dumps(credentials),) return self._login(login_fn=login, user=user) def login_as_dealer(self, **credentials): return self.login_as('dealer', **credentials) def login_as_agent(self, **credentials): return self.login_as('agent', **credentials) def login_as_manager(self, **credentials): return self.login_as('manager', **credentials) def login_as_super_manager(self, **credentials): return self.login_as('supermanager', **credentials) def login_as_endUser(self, user_agent=WECHAT_USER_AGENT, **credentials): user = MyUser.objects(**credentials).get() request = HttpRequest() request.META['HTTP_USER_AGENT'] = user_agent self._token = token = end_user_login(request, user) cookie_data = { 'value': token, 'max-age': None, 'path': '/', 'domain': settings.COOKIE_DOMAIN, 'secure': False, 'expires': None, } self.cookies[end_user_login_session_key] = cookie_data self.current_user = user return True def post_json(self, path, data=None, **kwargs): data = {} if data is None else data return self.post(path, json.dumps(data), content_type='application/json', **kwargs) class EndUserTestClient(RequestTestClient): def _get_headers(self): return { settings.JWT_AUTH_DOMAIN_DJANGO: settings.SERVICE_DOMAIN.USER, settings.JWT_TOKEN_DJANGO: getattr(self, '_token', None) } def get(self, path, data=None, follow=False, secure=False, **extra): headers = self._get_headers() extra.update(headers) return super(EndUserTestClient, self).get(path, data=None, follow=False, secure=False, **extra) def post(self, path, data=None, content_type='application/json', follow=False, secure=False, **extra): headers = self._get_headers() extra.update(headers) return super(EndUserTestClient, self).post(path, data=data, content_type=content_type, follow=follow, secure=secure, **extra) class WechatRequestTestClient(EndUserTestClient): def __init__(self, **defaults): super(WechatRequestTestClient, self).__init__(HTTP_USER_AGENT=WECHAT_USER_AGENT, **defaults) class AlipayRequestTestClient(EndUserTestClient): def __init__(self, **defaults): super(AlipayRequestTestClient, self).__init__(HTTP_USER_AGENT=ALIPAY_USER_AGENT, **defaults) class TestCase(_TestCase): client_class = RequestTestClient def _fixture_setup(self): pass def _fixture_teardown(self): pass class MongoBasedTestRunner(DiscoverRunner): def setup_databases(self, **kwargs): global _running_test _running_test = True db_name = TEST_DB connect(db_name) return db_name def teardown_databases(self, db_name, **kwargs): from pymongo import MongoClient client = MongoClient() client.drop_database(db_name) class APITestCase(TestCase): def setUp(self): self.client = requests class CookieBasedTestCase(TestCase): def setUp(self): self.cookie = SimpleCookie({}) def get_views(functions): return filter(lambda f: 'request' in inspect.getargspec(f).args, functions)