1
1
import asyncio
2
2
import aiohttp
3
3
import json
4
+ import pytest
4
5
5
6
6
- API_URL = "http://localhost:55555"
7
+ API_URL = "http://localhost:55555/api/v1/ "
7
8
8
9
9
10
async def get_proxies (session , request ):
@@ -12,7 +13,7 @@ async def get_proxies(session, request):
12
13
return json_data ['data' ]
13
14
14
15
15
- async def test_ordering (session, field_name):
16
+ async def check_ordering (session , field_name ):
16
17
request_data = {
17
18
'method' : 'get' ,
18
19
'model' : 'proxy' ,
@@ -31,7 +32,7 @@ async def test_ordering(session, field_name):
31
32
return True
32
33
33
34
34
- async def test_complex_ordering (session, *args):
35
+ async def check_complex_ordering (session , * args ):
35
36
fields = args
36
37
37
38
request_data = {
@@ -58,31 +59,29 @@ async def test_complex_ordering(session, *args):
58
59
return True
59
60
60
61
61
- async def run_tests(session):
62
+ @pytest .mark .asyncio
63
+ async def test_ordering ():
62
64
tests = [
63
- (test_ordering, 'response_time'),
64
- (test_ordering, 'uptime'),
65
- (test_ordering, 'number_of_bad_checks'),
66
- (test_ordering, 'last_check_time'),
67
- (test_complex_ordering, 'uptime', 'last_check_time'),
68
- (test_complex_ordering, 'number_of_bad_checks', 'uptime', 'response_time'),
65
+ (check_ordering , 'response_time' ),
66
+ (check_ordering , 'uptime' ),
67
+ (check_ordering , 'number_of_bad_checks' ),
68
+ (check_ordering , 'last_check_time' ),
69
69
]
70
70
71
71
for test in tests :
72
- try :
72
+ async with aiohttp . ClientSession () as session :
73
73
result = await test [0 ](session , * test [1 :])
74
- print("PASSED" if result else "FAILED", end='')
75
- except BaseException as ex:
76
- print("FAILED DUE TO EXCEPTION: {}".format(ex), end='')
77
- finally:
78
- print(" test {} ".format(test))
74
+ assert result
79
75
80
76
81
- async def main():
82
- async with aiohttp.ClientSession() as session:
83
- await run_tests(session)
84
-
77
+ @pytest .mark .asyncio
78
+ async def test_complex_ordering ():
79
+ tests = [
80
+ (check_complex_ordering , 'uptime' , 'last_check_time' ),
81
+ (check_complex_ordering , 'number_of_bad_checks' , 'uptime' , 'response_time' ),
82
+ ]
85
83
86
- if __name__ == '__main__':
87
- loop = asyncio.get_event_loop()
88
- loop.run_until_complete(main())
84
+ for test in tests :
85
+ async with aiohttp .ClientSession () as session :
86
+ result = await test [0 ](session , * test [1 :])
87
+ assert result
0 commit comments