forked from Boris-code/feapder
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_guest_user_pool.py
More file actions
76 lines (62 loc) · 2.07 KB
/
Copy pathtest_guest_user_pool.py
File metadata and controls
76 lines (62 loc) · 2.07 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# -*- coding: utf-8 -*-
"""
Created on 2021/9/13 2:33 下午
---------
@summary:
---------
@author: Boris
@email: boris_liu@foxmail.com
"""
import time
import unittest
from typing import Optional
from feapder.network.user_pool import GuestUser
from feapder.network.user_pool import GuestUserPool
class TestUserPool(unittest.TestCase):
def setUp(self) -> None:
# 默认的用户池，使用webdriver访问page_url生产cookie
self.user_pool = GuestUserPool(
"test:user_pool", page_url="https://www.baidu.com"
)
# 自定义生产cookie的方法
class CustomGuestUserPool(GuestUserPool):
def login(self) -> Optional[GuestUser]:
# 此处为假数据，正常需通过网站获取cookie
user = GuestUser(
user_agent="xxx",
proxies="yyy",
cookies={"some_key": "some_value{}".format(time.time())},
)
return user
self.custom_user_pool = CustomGuestUserPool(
"test:custom_user_pool", min_users=10, keep_alive=True
)
def test_get_user(self):
"""
测试直接获取游客用户
Returns:
"""
user = self.custom_user_pool.get_user(block=True)
print("取到user：", user)
print("cookie：", user.cookies)
print("user_agent：", user.user_agent)
print("proxies：", user.proxies)
def test_del_user(self):
user = GuestUser(
**{
"user_id": "9f1654ba654e12adfea548eae89a8f6f",
"user_agent": "xxx",
"proxies": "yyy",
"cookies": {"some_key": "some_value1640006728.908013"},
}
)
print(user.user_id)
self.custom_user_pool.del_user(user.user_id)
def test_keep_alive(self):
"""
测试生产游客用户，面对需要大量cookie，需要单独起个进程维护cookie的场景
Returns:
"""
self.custom_user_pool.run()
if __name__ == "__main__":
unittest.main()