​```
import asyncio
import aiohttp
import json
def fetch_page(url):
    try:
        response = yield from aiohttp.request('GET', url)
        string = (yield from response.read()).decode('utf-8')
        if response.status == 200:
            data = json.loads(string)
            print(data)
        else:
            print("data fetch failed for")
            print(response.content, response.status)
    except asyncio.TimeoutError:
        print("访问失败")
if __name__ == '__main__':
    url = 'http://gc.ditu.aliyun.com/geocoding?a=%E8%8B%8F%E5%B7%9E%E5%B8%82'
    # urls = [url] * 4
    loop = asyncio.get_event_loop()
    tasks = []
    for i in range(5):
        tasks.append(asyncio.async(fetch_page(url)))
    # tasks = [asyncio.async(fetch_page(z, i)) for i, z in enumerate(urls)]
    print(tasks)
    loop.run_until_complete(asyncio.wait(tasks, timeout=5))
    loop.close()
​```

封装一次

​```
class fetch():
    def __init__(self, dict_http):
        '''
        http请求的封装，传入dict
        :param dict_http:
        '''
        self.dict_http = dict_http
    def get(self, url, param):
        data = {}
        url = self.dict_http["protocol"] + self.dict_http["host"] + ":" + str(self.dict_http["port"]) + url
        print(url)
        try:
            response = yield from aiohttp.request("GET", url, headers=self.dict_http["header"], params=param)
            string = (yield from response.read()).decode('utf-8')
            if response.status == 200:
                data = json.loads(string)
            else:
                print("data fetch failed for")
                print(response.content, response.status)
            data["status_code"] = response.status
            print(data)
        except asyncio.TimeoutError:
            print("访问失败")
        return data
    def post(self,url, param):
        data = {}
        url = self.dict_http["protocol"] + self.dict_http["host"] + ':' + str(self.dict_http["port"]) + url
        try:
            response = yield from aiohttp.request('POST', url, data=param)
            string = (yield from response.read()).decode('utf-8')
            if response.status == 200:
                data = json.loads(string)
            else:
                print("data fetch failed for")
                print(response.content, response.status)
            data["status_code"] = response.status
            print(data)
        except asyncio.TimeoutError:
            print("访问失败")
        return data
if __name__ == '__main__':
    url = '/iplookup/iplookup.php'
    loop = asyncio.get_event_loop()
    tasks = []
    dict_http = {}
    dict_http["protocol"] = "http://"
    dict_http["host"] = "int.dpool.sina.com.cn"
    dict_http["port"] = 80
    dict_http["header"] = {"Accept":"text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8","User-Agent":"Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36"}
    f = fetch(dict_http)
    for i in range(2):
        tasks.append(asyncio.async(f.get(url, param="format=json&ip=218.4.255.255")))
    loop.run_until_complete(asyncio.wait(tasks, timeout=5))
    loop.close()

​```

l = [2, 4, 3, 5]
str = "bcdea"
def badIterable():
    i = 0
    iterable = ["a", "b", "c"]
    for item in iterable:
        print(i, item)
        i += 1
def goodIterable():
    iterable = ["a", "b", "c"]
    for i, item in enumerate(iterable):
        print(i, item)

#结果：[(0, 'a'), (1, 'b'), (2, 'c')]
def listIterable():
    le = list(enumerate('abc'))
    return le
#结果：[(1, 'a'), (2, 'b'), (3, 'c')]
def listIterable1():
    le = list(enumerate('abc', 1))
# 正序
def sortList(li):
    li.sort()
    return li
# 倒序
def reverseList(li):
    # li.reverse()
    temp = li[::-1]
    print(temp)
    return temp

# 打印
def testSet():
    my_set = {i * 15 for i in range(100)}
    print(my_set)
# 交集
def intersection():
    x = set("span")
    y = set("pta")
    print(x&y)
# 合集
def collection():
    x = set("span")
    y = set("pta")
    print(x|y)
# 差集
def subtract():
    x = set("span")
    y = set("pta")
    print(x-y)

#去掉重复数据
def removeRepeat():
    a = [11,22,33,44,11,22]
    b = set(a)
    print(b)
    c = [i for i in b]
    print(c)

foo = "t1m3a3pp4mt"    
print("".join(list(set(foo)))[::-1])   # 去掉重复字符串，倒序排列
print("".join(list(set(foo)))[::1]) # 去掉重复自父亲，正序排列

import matplotlib.pyplot as pl
   import matplotlib.dates as mdates
   import datetime
   x = [
     datetime.datetime(2011,1,1,1,1,2),
     datetime.datetime(2011,1,1,1,1,3),
     datetime.datetime(2011,1,1,1,1,4),
     datetime.datetime(2011,1,1,1,1,5),
     datetime.datetime(2011,1,1,1,1,6),
     datetime.datetime(2011,1,1,1,1,7),
     ]

   list1 = [20,10,90,10,50,3]
   list2 = [1000,3000,2899,1922,16000,89222]
   a1 = pl.subplot(311) # 曲线图一
   a1.set_title("CPU")
   a1.set_ylabel("占用情况%")
   a1.plot(x,list1)
   # a1.xaxis.set_major_locator(mdates.SecondLocator(interval=2))
   a1.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))

   a2 = pl.subplot(312) #曲线图二
   a3 = pl.subplot(313) #曲线图三

   a2.set_title("内存")
   a2.set_ylabel("使用情况 K")
   a2.plot(x,list2)
   # a1.xaxis.set_major_locator(mdates.SecondLocator(interval=2))
   a2.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))


   a3.set_title("流量")
   a3.set_ylabel("使用情况 K")
   a3.plot(x,list2)
   # a1.xaxis.set_major_locator(mdates.SecondLocator(interval=2))
   a3.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))

   # a1.margins(x=0.2)
   pl.tight_layout()
   pl.show()

import pytest

@pytest.mark.finished
def test_add():
    print("测试函数：test_add")


@pytest.mark.finished
def test_subtract():
    print("测试函数：test_subtract")


@pytest.mark.unfinished
def test_no_finish():
    pass
if __name__ == "__main__":
    pytest.main(["-s", "pt_test1.py"])

pytest.main(["-s", "pt_test1.py"])

# test_db.py

@pytest.fixture()
def db():
    print('Connection successful')

    yield

    print('Connection closed')


def search_user(user_id):
    d = {
        '001': 'xiaoming'
    }
    return d[user_id]


def test_search(db):
    assert search_user('001') == 'xiaoming

============================= test session starts =============================
platform win32 -- Python 3.6.4, pytest-3.6.1, py-1.5.2, pluggy-0.6.0
rootdir: F:\self-repo\learning-pytest, inifile:
collected 1 item

tests\fixture\test_db.py Connection successful
.Connection closed

# conftest.py 

import os
from selenium import webdriver
import pytest

@pytest.fixture(scope='session')
def driver(request):
    PATH = lambda p: os.path.abspath(
        os.path.join(os.path.dirname(__file__), p)
    )
    chromedriver = PATH("exe/chromedriver.exe")
    os.environ["webdriver.chrome.driver"] = chromedriver
    driver = webdriver.Chrome(chromedriver)
    driver.maximize_window()  # 将浏览器最大化
    driver.implicitly_wait(2)
    # openurl = "http://www.baidu.com"
    # driver.get(openurl)

    def end():
        driver.quit()

    request.addfinalizer(end)  # 终结函数
    return driver


# test_1.py
import pytest
def test_001(driver):
    openurl = "http://www.baidu.com"
    driver.get(openurl)
    driver.find_element_by_id("kw").click()

if __name__ == '__main__':
    pytest.main(['-v', 'test_1.py'])

# test_2.py
import pytest
import time

def test_001(driver):
    # openurl = "http://www.baidu.com"
    # driver.get(openurl)
    driver.find_element_by_id("kw").send_keys("baidu")
    time.sleep(5)
if __name__ == '__main__':
    pytest.main(['-v', 'test_2.py'])

pytest -s TestCase\

# coding:utf-8  
# singleton.py
#单例模式函数，用来修饰类
def singleton(cls,*args,**kw):
    instances = {}
    def _singleton():
        if cls not in instances:
            instances[cls] = cls(*args,**kw)
        return instances[cls]
    return _singleton

# -*- coding:utf-8 -*-
from selenium import webdriver
from Utils.singleton import singleton
import os
@singleton
class GetSeleniumDriver(object):
    def __init__(self):
        PATH = lambda p: os.path.abspath(
        os.path.join(os.path.dirname(__file__), p))
        chromedriver = PATH("../exe/chromedriver.exe")
        os.environ["webdriver.chrome.driver"] = chromedriver
        self.driver = webdriver.Chrome(chromedriver)
        self.driver.maximize_window()  # 将浏览器最大化
        self.driver.implicitly_wait(2)

# test_3.py

import pytest
import time
from Utils.GetSeleniumDriver import GetSeleniumDriver
class TestCase003(object):

   # 初始化
    def setup_class(self):
        self.driver=GetSeleniumDriver().driver
        self.driver.get("http://www.baidu.com")

    def test_003(self):
        self.driver.find_element_by_id("kw").send_keys("baidu")
        time.sleep(5)
    # 清理后，影响其他函数
    def teardown_class(self):
        # self.driver.close()
        # self.driver.quit()
        pass


if __name__ == '__main__':
    pytest.main(['-v', 'test_3.py'])

@pytest.mark.finished
   def test_baidu2(self, driver):
       """
       测试百度下
       """
       page = BaiduPage(driver, "test_baidu2")
       page.operate()
       page.check_point()

pytest -m finished webAuto\testcase --html=report.html --self-contained-html --capture=sys

from schematics.models import Model
from schematics.types import StringType, URLType
import json
class Person(Model):
     name = StringType(required=True) #不能为空
     website = URLType()
person = Person({'name':'sss',
                'website': 'http://soundcloud.com/joestrummer'})
person.name=u"哈哈"
js = json.dumps(person.to_primitive())
print(js)
f = person.validate() ##验证字段是否正确，如果不正确会报错，如果正确返回的就是none
print(f)

from schematics.models import Model
from schematics.types import StringType, IntType
from schematics.types.compound import ListType,MultiType
class Person(Model):
    mls = ListType(MultiType())
    shi = StringType()
    code = IntType()
mls_test = Person({'mls': [{
     'en_US': 'Hello, world!',
    'fr_FR': 'Bonjour tout le monde!',
    'es_MX': '¡Hola, mundo!',
     }],'shi':'kun','code':0})
print(mls_test.mls[0]["en_US"])

from schematics.models import Model
from schematics.types import StringType, IntType,MultilingualStringType
class Person(Model):
    mls = MultilingualStringType()
    shi = StringType()
    code = IntType()
mls_test = Person({'mls': {
     'en_US': 'Hello, world!',
    'fr_FR': 'Bonjour tout le monde!',
    'es_MX': '¡Hola, mundo!',
     },'shi':'kun','code':0})
print(mls_test.mls["en_US"])

from jsonschema import validate

schema = {
    "type" : "object",
    "properties" : {
        "price" : {"type" : "number"},
        "name" : {"type" : "string"},
        "list":{"maxItems":2},
        "address":{'regex':'bj'},
    },
}

validate({"name" : "Eggs", "price" : "dfda",'list':[1,5],'address':'bj-jiuxianqiao'}, schema)

import unittest
import os


class MyUnittest(unittest.TestCase):

    def setUp(self) -> None:
        print("setUp_每运行一次用例都会执行一次")
        self.name = "hi"
        print("setUp_name_%s" % self.name)

    def tearDown(self) -> None:
        print("tearDown_每结束一次用例都会执行一次")
        print("tearDown_name_%s" % self.name)

    @classmethod
    def setUpClass(cls) -> None:
        print("setUpClass_整个测试开始后执行，只执行一次")
        cls.func = "setUpClass"
        print("setUpClass_func_%s" % cls.func)

    @classmethod
    def tearDownClass(cls) -> None:
        print("tearDownClass_整个测试完成后执行，只执行一次")
        print("tearDownClass_func_%s" % cls.func)

    def test_add(self):
        print("测试函数：test_add")
        self.name = "test_add"
        print("setUpClass_cls.func_%s" % self.func)

    def test_subtract(self):
        print("测试函数：test_subtract")
        self.name = "test_subtract"
        self.func = "test_subtract"


if __name__ == "__main__":
    # 运行方式一
    unittest.main()

setUpClass_整个测试开始后执行，只执行一次
setUpClass_func_setUpClass
test_add (ut_module.ut1_test.MyUnittest) ... setUp_每运行一次用例都会执行一次
setUp_name_hi
测试函数：test_add
setUpClass_cls.func_setUpClass
tearDown_每结束一次用例都会执行一次
tearDown_name_test_add
ok
test_subtract (ut_module.ut1_test.MyUnittest) ... setUp_每运行一次用例都会执行一次
setUp_name_hi
测试函数：test_subtract
tearDown_每结束一次用例都会执行一次
tearDown_name_test_subtract
ok
tearDownClass_整个测试完成后执行，只执行一次
tearDownClass_func_setUpClass

----------------------------------------------------------------------
Ran 2 tests in 0.003s

OK

setUp_name_hi
测试函数：test_add
setUpClass_cls.func_setUpClass
tearDown_每结束一次用例都会执行一次
tearDown_name_test_add

在def setUpClass赋值
self.func = "setUpClass"
在 def test_subtract中赋值
self.func = "test_subtract"
# 执行结果
setUpClass_整个测试开始后执行，只执行一次
setUpClass_func_setUpClass
.....
tearDownClass_整个测试完成后执行，只执行一次
tearDownClass_func_setUpClass

unittest.main()

# 构造测试集
suite = unittest.TestSuite()
suite.addTest(MyUnittest("test_add"))
suite.addTest(MyUnittest("test_subtract"))
# 执行测试
runner = unittest.TextTestRunner()
runner.run(suite)

# TestLoader 用来加载TestCase到TestSuite中的，其中有几个  loadTestsFrom__()方法，就是从各个地方寻找TestCase，创建它们的实例，然后add到TestSuite中，再返回一个TestSuite实例
# TestSuite 测试套件集合，最终将传递给testRunner进行测试执行
# TextTestRunner 用来执行测试用例，将测试结果保存在TextTestResult中
# 此用法可以同时测试多个类
suite1 = unittest.TestLoader().loadTestsFromTestCase(MyUnittest)
suite2 = unittest.TestLoader().loadTestsFromTestCase(MyUnittest1)
suite = unittest.TestSuite([suite1, suite2 ])
unittest.TextTestRunner(verbosity=2).run(suite)

# 进入到测试目录
os.chdir(".//ut_module")
os.system("python -m unittest -v ut1_test")

# 测试某个类下的所有函数
D:\project\ut>python -m unittest -v  ut_module.ut1_test
# 测试类中某一个函数
D:\project\ut>python -m unittest -v ut_module.ut1_test.MyUnittest.test_add
# 执行模块下所有test结尾的文件
D:\project\ut>python -m unittest discover D:\project\ut\ut_module "*_test.py"
# 同时执行多个测试类下的测试函数
D:\project\ut>python -m unittest  ut_module.ut1_test.MyUnittest.test_add  ut_module.ut2_test.MyUnittest.test_subtract1

class MyTestCase(unittest.TestCase):

    @unittest.skip("demonstrating skipping")
    def test_nothing(self):
        self.fail("shouldn't happen")

    @unittest.skipIf(mylib.__version__ < (1, 3),
                     "not supported in this library version")
    def test_format(self):
        # Tests that work for only a certain version of the library.
        pass

    @unittest.skipUnless(sys.platform.startswith("win"), "requires Windows")
    def test_windows_support(self):
        # windows specific testing code
        pass

    def test_maybe_skipped(self):
        if not external_resource_available():
            self.skipTest("external resource not available")
        # test code that depends on the external resource
        pass
   @unittest.expectedFailure
    def test_fail(self):
        self.assertEqual(1, 0, "broken")

@unittest.skip("showing class skipping")
class MySkippedTestCase(unittest.TestCase):
    def test_not_run(self):
        pass

import unittest
from ddt import ddt, file_data
from ddt_demo.mycode import has_three_elements,is_a_greeting

@ddt
class FooTestCase(unittest.TestCase):

    @file_data('test_data_dict.json')
    def test_file_data_json_dict(self, value):
        self.assertTrue(has_three_elements(value))

    @file_data('test_data_list.json')
    def test_file_data_json_list(self, value):
        self.assertTrue(is_a_greeting(value))
        
if __name__=='__main__':
    unittest.main(verbosity=2

import unittest
from ddt import ddt, file_data
from ddt_demo.mycode import has_three_elements,is_a_greeting

@ddt
class FooTestCase(unittest.TestCase):

    @file_data('test_data_dict.yaml')
    def test_file_data_yaml_dict(self, value):
        self.assertTrue(has_three_elements(value))

    @file_data('test_data_list.yaml')
    def test_file_data_yaml_list(self, value):
        self.assertTrue(is_a_greeting(value))
        
if __name__=='__main__':
    unittest.main(verbosity=2)

import threading
import time
def one_thread(name,id):
    print("start....")
    print(name)
    print(id)
    time.sleep(5)
    print("end...")

print("start thread")
threading.Thread(target=one_thread, args=(), kwargs={"name": 111, "id": 222}).start()
# args是一个list
# kwargs是一个字典，需要对应函数的key
print("end thread")

import threading
import time


class myThread(threading.Thread):
    def __init__(self, threadID, name):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name

    def run(self):
        print_time(self.threadID, self.name)

num = 0
def print_time(threadID, name):
    global num
    # 每一个线程循环10次，最终总循环次数为30次
    for i in range(10):
        print("start run")
        time.sleep(2)
        print(i)
        num += 1
    print("thread_id=%s:name=%s" % (threadID, name))


if __name__ == '__main__':
    threads = []
    # 新增三个线程
    for i in range(3):
        name = "Thread-%d" % i
        t = myThread(i, name)
        t.start()
        threads.append(t)
    for t in threads:
        t.join()
    print("所有线程执行完毕")
    print("总循环次数为：%s" % num)

# -*- coding: utf-8 -*-
import time
import threading
# 创建锁对象
lock = threading.Lock()
num = 0

def run(n):
    global num
    for i in range(10):
        # 加锁  为了确保下面代码只能由一个线程从头到尾的执行
        # 会阻止多线程的并发执行，所以效率会大大降低
        """
        lock.acquire()
        try:
            num = num - n
            num = num + n
        finally:
            # 解锁
            lock.release()
        """
        with lock:
            time.sleep(2)
            print("start")
            num = num + 1
            print("==============")


if __name__ == '__main__':
    t1 = threading.Thread(target=run,args=(6,))
    t2 = threading.Thread(target=run,args=(9,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("num = %s"%(num))

start
==============
...
num = 20

import threading
import time
import queue

q = queue.Queue(10)

threadLock = threading.Lock()


class myThread(threading.Thread):
    def __init__(self, threadID, name):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.exitFlag = 0

    def run(self):
        while not self.exitFlag:
            threadLock.acquire()
            if not q.empty():
                id = q.get()
                print_time(self.name, id)
                threadLock.release()
            else:
                threadLock.release()


def print_time(threadName, id):
    print ("%s:%s:%s"%(threadName,time.ctime(time.time()),id))
    # pass


# 创建3个线程
threads = []
for i in range(3):
    name = "Thread-%d" % i
    t = myThread(i, name)
    t.start()
    threads.append(t)
print(threads)

# 新增队列数据
for i in range(10000):
    q_name = "Queue:%d" % i
    q.put(q_name)

# 等待队列清空
while not q.empty():
    pass

# 也可以join方法，与上同效
# q.join()

# 通知线程，处理完之后关闭
for t in threads:
    t.exitFlag = 1

# 等待所有线程结束之后才退出
for t in threads:
    t.join()

print("Exiting Main Thread")

with open('data1.txt', mode='w', encoding="utf-8") as f:
    data = '11111e'
    f.seek(10)   #光标移动到10的位置
    f.write(data)

  f1 = open(r'D:\project\operate_file\data.txt', "r", encoding="utf-8")
    print(f1.readline())
    f1.close()

f = open("file_test","w",encoding="utf-8")
f.write("hello\n")
f.write("hello\n")
f.write("hello\n")
f.flush()

import subprocess
t = subprocess.call('adb devices')
print(t) # 打印为0

output = os.popen('adb devices')
print(output.read()) # 得到List of devices attached

pipe=subprocess.Popen("adb devices",shell=True,stdout=subprocess.PIPE).stdout
print(pipe.read()) # 得到值b'List of devices attached\r\n\r\n'

import  sys
import xlrd
#https://pypi.python.org/pypi/xlrd/0.9.3
import xlsxwriter
#https://pypi.python.org/pypi/XlsxWriter/0.6.6#downloads
def read_excel(file='c:/test.xls'):
    data = xlrd.open_workbook(file)
    table = data.sheet_by_index(0)
    nrows = table.nrows
    ncols = table.ncols
    colnames = table.row_values(0) #one rows data
    list = []
    for rownum in range(1, nrows):
        row = table.row_values(rownum)
        if row:
            app = {}
            for i in range(len(colnames)):
                row[i] = colnames[i] + row[i]
                app[colnames[i]] = row[i]
            list.append(app)
    return list

def write_excel(file='d:/result.xlsx', httpurl=["",""], httpmethod=get, response_time=[0,10,14,0]):
    workbook = xlsxwriter.Workbook(file)
    worksheet = workbook.add_worksheet()
    #worksheet.set_column('A:A', 20)
    worksheet.write(0, 0, "接口URL")
    worksheet.write(0, 1, "请求方法")
    worksheet.write(0, 2, "响应时间")
    for i in range(len(response_time)):
        worksheet.write(i + 1, 0, httpurl[i])
        worksheet.write(i + 1, 1, httpmethod)
        if response_time[i] == 0:
            worksheet.write(i + 1, 2, "请求超时")
        else:
             worksheet.write(i + 1, 2, response_time[i])
    #worksheet.insert_image('B5', 'logo.png')
    workbook.close()