# 添加發(fā)布會(huì)接口實(shí)現(xiàn)
def add_event(request):
  eid = request.POST.get('eid','')         # 發(fā)布會(huì)id
  name = request.POST.get('name','')        # 發(fā)布會(huì)標(biāo)題
  limit = request.POST.get('limit','')       # 限制人數(shù)
  status = request.POST.get('status','')      # 狀態(tài)
  address = request.POST.get('address','')     # 地址
  start_time = request.POST.get('start_time','')  # 發(fā)布會(huì)時(shí)間

  if eid =='' or name == '' or limit == '' or address == '' or start_time == '':
    return JsonResponse({'status':10021,'message':'parameter error'})

  result = Event.objects.filter(id=eid)
  if result:
    return JsonResponse({'status':10022,'message':'event id already exists'})

  result = Event.objects.filter(name=name)
  if result:
    return JsonResponse({'status':10023,'message':'event name already exists'})

  if status == '':
    status = 1

  try:
    Event.objects.create(id=eid,name=name,limit=limit,address=address,status=int(status),start_time=start_time)
  except ValidationError:
    error = 'start_time format error. It must be in YYYY-MM-DD HH:MM:SS format.'
    return JsonResponse({'status':10024,'message':error})

  return JsonResponse({'status':200,'message':'add event success'})

# 發(fā)布會(huì)查詢接口實(shí)現(xiàn)
def get_event_list(request):

  eid = request.GET.get("eid", "")   # 發(fā)布會(huì)id
  name = request.GET.get("name", "")  # 發(fā)布會(huì)名稱

  if eid == '' and name == '':
    return JsonResponse({'status':10021,'message':'parameter error'})

  if eid != '':
    event = {}
    try:
      result = Event.objects.get(id=eid)
    except ObjectDoesNotExist:
      return JsonResponse({'status':10022, 'message':'query result is empty'})
    else:
      event['eid'] = result.id
      event['name'] = result.name
      event['limit'] = result.limit
      event['status'] = result.status
      event['address'] = result.address
      event['start_time'] = result.start_time
      return JsonResponse({'status':200, 'message':'success', 'data':event})

  if name != '':
    datas = []
    results = Event.objects.filter(name__contains=name)
    if results:
      for r in results:
        event = {}
        event['eid'] = r.id
        event['name'] = r.name
        event['limit'] = r.limit
        event['status'] = r.status
        event['address'] = r.address
        event['start_time'] = r.start_time
        datas.append(event)
      return JsonResponse({'status':200, 'message':'success', 'data':datas})
    else:
      return JsonResponse({'status':10022, 'message':'query result is empty'})

#查詢發(fā)布會(huì)接口測(cè)試代碼
import requests

url = "http://127.0.0.1:8000/api/get_event_list/"
r = requests.get(url, params={'eid':'1'})
result = r.json()
print(result)
assert result['status'] == 200
assert result['message'] == "success"
assert result['data']['name'] == "xx 產(chǎn)品發(fā)布會(huì)"
assert result['data']['address'] == "北京林匹克公園水立方"
assert result['data']['start_time'] == "2016-10-15T18:00:00"

#發(fā)布會(huì)查詢接口測(cè)試代碼 
import unittest
import requests

class GetEventListTest(unittest.TestCase):

  def setUp(self):
    self.base_url = "http://127.0.0.1:8000/api/get_event_list/"

  def test_get_event_list_eid_null(self):
    ''' eid 參數(shù)為空 '''
    r = requests.get(self.base_url, params={'eid':''})
    result = r.json()
    self.assertEqual(result['status'], 10021)
    self.assertEqual(result['message'], 'parameter error')

  def test_get_event_list_eid_error(self):
    ''' eid=901 查詢結(jié)果為空 '''
    r = requests.get(self.base_url, params={'eid':901})
    result = r.json()
    self.assertEqual(result['status'], 10022)
    self.assertEqual(result['message'], 'query result is empty')

  def test_get_event_list_eid_success(self):
    ''' 根據(jù) eid 查詢結(jié)果成功 '''
    r = requests.get(self.base_url, params={'eid':1})
    result = r.json()
    self.assertEqual(result['status'], 200)
    self.assertEqual(result['message'], 'success')
    self.assertEqual(result['data']['name'],u'mx6發(fā)布會(huì)')
    self.assertEqual(result['data']['address'],u'北京國(guó)家會(huì)議中心')

  def test_get_event_list_nam_result_null(self):
    ''' 關(guān)鍵字‘a(chǎn)bc'查詢 '''
    r = requests.get(self.base_url, params={'name':'abc'})
    result = r.json()
    self.assertEqual(result['status'], 10022)
    self.assertEqual(result['message'], 'query result is empty')

  def test_get_event_list_name_find(self):
    ''' 關(guān)鍵字‘發(fā)布會(huì)'模糊查詢 '''
    r = requests.get(self.base_url, params={'name':'發(fā)布會(huì)'})
    result = r.json()
    self.assertEqual(result['status'], 200)
    self.assertEqual(result['message'], 'success')
    self.assertEqual(result['data'][0]['name'],u'mx6發(fā)布會(huì)')
    self.assertEqual(result['data'][0]['address'],u'北京國(guó)家會(huì)議中心')
49if __name__ == '__main__':
   unittest.main()

import pymysql.cursors
import os
import configparser as cparser


# ======== Reading db_config.ini setting ===========
base_dir = str(os.path.dirname(os.path.dirname(__file__)))
base_dir = base_dir.replace('\\', '/')
file_path = base_dir + "/db_config.ini"

cf = cparser.ConfigParser()

cf.read(file_path)
host = cf.get("mysqlconf", "host")
port = cf.get("mysqlconf", "port")
db  = cf.get("mysqlconf", "db_name")
user = cf.get("mysqlconf", "user")
password = cf.get("mysqlconf", "password")


# ======== MySql base operating ===================
class DB:

  def __init__(self):
    try:
      # Connect to the database
      self.connection = pymysql.connect(host=host,
                       port=int(port),
                       user=user,
                       password=password,
                       db=db,
                       charset='utf8mb4',
                       cursorclass=pymysql.cursors.DictCursor)
    except pymysql.err.OperationalError as e:
      print("Mysql Error %d: %s" % (e.args[0], e.args[1]))

  # clear table data
  def clear(self, table_name):
    # real_sql = "truncate table " + table_name + ";"
    real_sql = "delete from " + table_name + ";"
    with self.connection.cursor() as cursor:
      cursor.execute("SET FOREIGN_KEY_CHECKS=0;")
      cursor.execute(real_sql)
    self.connection.commit()

  # insert sql statement
  def insert(self, table_name, table_data):
    for key in table_data:
      table_data[key] = "'"+str(table_data[key])+"'"
    key  = ','.join(table_data.keys())
    value = ','.join(table_data.values())
    real_sql = "INSERT INTO " + table_name + " (" + key + ") VALUES (" + value + ")"
    #print(real_sql)

    with self.connection.cursor() as cursor:
      cursor.execute(real_sql)

    self.connection.commit()

  # close database
  def close(self):
    self.connection.close()

  # init data
  def init_data(self, datas):
    for table, data in datas.items():
      self.clear(table)
      for d in data:
        self.insert(table, d)
    self.close()


if __name__ == '__main__':

  db = DB()
  table_name = "sign_event"
  data = {'id':1,'name':'紅米','`limit`':2000,'status':1,'address':'北京會(huì)展中心','start_time':'2016-08-20 00:25:42'}
  table_name2 = "sign_guest"
  data2 = {'realname':'alen','phone':12312341234,'email':'alen@mail.com','sign':0,'event_id':1}

  db.clear(table_name)
  db.insert(table_name, data)
  db.close()

import sys
sys.path.append('../db_fixture')
try:
  from mysql_db import DB
except ImportError:
  from .mysql_db import DB

# create data
datas = {
  'sign_event':[
    {'id':1,'name':'紅米Pro發(fā)布會(huì)','`limit`':2000,'status':1,'address':'北京會(huì)展中心','start_time':'2017-08-20 14:00:00'},
    {'id':2,'name':'可參加人數(shù)為0','`limit`':0,'status':1,'address':'北京會(huì)展中心','start_time':'2017-08-20 14:00:00'},
    {'id':3,'name':'當(dāng)前狀態(tài)為0關(guān)閉','`limit`':2000,'status':0,'address':'北京會(huì)展中心','start_time':'2017-08-20 14:00:00'},
    {'id':4,'name':'發(fā)布會(huì)已結(jié)束','`limit`':2000,'status':1,'address':'北京會(huì)展中心','start_time':'2001-08-20 14:00:00'},
    {'id':5,'name':'小米5發(fā)布會(huì)','`limit`':2000,'status':1,'address':'北京國(guó)家會(huì)議中心','start_time':'2017-08-20 14:00:00'},
  ],
  'sign_guest':[
    {'id':1,'realname':'alen','phone':13511001100,'email':'alen@mail.com','sign':0,'event_id':1},
    {'id':2,'realname':'has sign','phone':13511001101,'email':'sign@mail.com','sign':1,'event_id':1},
    {'id':3,'realname':'tom','phone':13511001102,'email':'tom@mail.com','sign':0,'event_id':5},
  ],
}

# Inster table datas
def init_data():
  DB().init_data(datas)


if __name__ == '__main__':
  init_data()

import unittest
import requests
import os, sys
parentdir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, parentdir)
from db_fixture import test_data


class AddEventTest(unittest.TestCase):
  ''' 添加發(fā)布會(huì) '''

  def setUp(self):
    self.base_url = "http://127.0.0.1:8000/api/add_event/"

  def tearDown(self):
    print(self.result)

  def test_add_event_all_null(self):
    ''' 所有參數(shù)為空 '''
    payload = {'eid':'','':'','limit':'','address':"",'start_time':''}
    r = requests.post(self.base_url, data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10021)
    self.assertEqual(self.result['message'], 'parameter error')

  def test_add_event_eid_exist(self):
    ''' id已經(jīng)存在 '''
    payload = {'eid':1,'name':'一加4發(fā)布會(huì)','limit':2000,'address':"深圳寶體",'start_time':'2017'}
    r = requests.post(self.base_url, data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10022)
    self.assertEqual(self.result['message'], 'event id already exists')

  def test_add_event_name_exist(self):
    ''' 名稱已經(jīng)存在 '''
    payload = {'eid':11,'name':'紅米Pro發(fā)布會(huì)','limit':2000,'address':"深圳寶體",'start_time':'2017'}
    r = requests.post(self.base_url,data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10023)
    self.assertEqual(self.result['message'], 'event name already exists')

  def test_add_event_data_type_error(self):
    ''' 日期格式錯(cuò)誤 '''
    payload = {'eid':11,'name':'一加4手機(jī)發(fā)布會(huì)','limit':2000,'address':"深圳寶體",'start_time':'2017'}
    r = requests.post(self.base_url,data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 10024)
    self.assertIn('start_time format error.', self.result['message'])

  def test_add_event_success(self):
    ''' 添加成功 '''
    payload = {'eid':11,'name':'一加4手機(jī)發(fā)布會(huì)','limit':2000,'address':"深圳寶體",'start_time':'2017-05-10 12:00:00'}
    r = requests.post(self.base_url,data=payload)
    self.result = r.json()
    self.assertEqual(self.result['status'], 200)
    self.assertEqual(self.result['message'], 'add event success')


if __name__ == '__main__':
  test_data.init_data() # 初始化接口測(cè)試數(shù)據(jù)
  unittest.main()

import time, sys
sys.path.append('./interface')
sys.path.append('./db_fixture')
from HTMLTestRunner import HTMLTestRunner
import unittest
from db_fixture import test_data


# 指定測(cè)試用例為當(dāng)前文件夾下的 interface 目錄
test_dir = './interface'
discover = unittest.defaultTestLoader.discover(test_dir, pattern='*_test.py')


if __name__ == "__main__":
  test_data.init_data() # 初始化接口測(cè)試數(shù)據(jù)

  now = time.strftime("%Y-%m-%d %H_%M_%S")
  filename = './report/' + now + '_result.html'
  fp = open(filename, 'wb')
  runner = HTMLTestRunner(stream=fp,
              title='Guest Manage System Interface Test Report',
              description='Implementation Example with: ')
  runner.run(discover)
  fp.close()