Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit d692982

Browse files
committed
feat: add test cases for python code
1 parent 72d0e1c commit d692982

11 files changed

+293
-0
lines changed

‎src/python/main/distributed_lock.py

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
import random
2+
import string
3+
import time
4+
5+
from redis import Redis
6+
7+
8+
def generate_random_value():
9+
return ''.join(random.sample(string.ascii_letters + string.digits, 32))
10+
11+
12+
# 设定默认过期时长为 10s
13+
DEFAULT_TIMEOUT = 10
14+
15+
16+
class DistributedLock:
17+
def __init__(self, client: Redis, key: str):
18+
self.client = client
19+
self.key = key
20+
21+
def acquire(self, timeout=DEFAULT_TIMEOUT) -> bool:
22+
"""尝试获取锁"""
23+
res = self.client.set(self.key, generate_random_value(), ex=timeout, nx=True) # ex 表示过期时长秒数,nx 表示 key 不存在时才设置
24+
return res is True
25+
26+
def release(self) -> bool:
27+
"""尝试释放锁"""
28+
return self.client.delete(self.key) == 1

‎src/python/main/paginate.py

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
from redis import Redis
2+
3+
4+
class Paginate:
5+
def __init__(self, client: Redis, key: str):
6+
self.client = client
7+
self.key = key
8+
9+
def add(self, item):
10+
"""添加元素到分页列表中"""
11+
self.client.lpush(self.key, item)
12+
13+
def get_page(self, page: int, per_page: int):
14+
"""根据页码取出指定数量的元素"""
15+
start = (page - 1) * per_page
16+
end = page * per_page - 1
17+
return self.client.lrange(self.key, start, end)
18+
19+
def get_size(self):
20+
"""获取列表的元素数量"""
21+
return self.client.llen(self.key)

‎src/python/main/ranking_list.py

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
import datetime
2+
import time
3+
4+
from redis import Redis
5+
6+
7+
def get_day_rank_key():
8+
return 'rank:day:{}'.format(datetime.date.today().strftime('%Y%m%d'))
9+
10+
11+
def get_week_rank_key():
12+
return 'rank:week:{}{}'.format(datetime.date.today().strftime('%Y'), time.strftime("%W"))
13+
14+
15+
def get_month_rank_key():
16+
today = datetime.date.today()
17+
return 'rank:month:{}{}'.format(today.strftime('%Y'), today.strftime('%m'))
18+
19+
20+
class Ranking:
21+
def __init__(self, client: Redis):
22+
self.client = client
23+
24+
def incr(self, user, score=1):
25+
self.client.zincrby(get_day_rank_key(), score, user)
26+
self.client.zincrby(get_week_rank_key(), score, user)
27+
self.client.zincrby(get_month_rank_key(), score, user)
28+
29+
def get_today_top_n(self, n, with_scores=False):
30+
"""获取日榜单topN"""
31+
return self.client.zrevrange(get_day_rank_key(), 0, n - 1, withscores=with_scores)
32+
33+
def get_week_top_n(self, n, with_scores=False):
34+
"""获取周榜单topN"""
35+
return self.client.zrevrange(get_week_rank_key(), 0, n - 1, withscores=with_scores)
36+
37+
def get_month_top_n(self, n, with_scores=False):
38+
"""获取月榜单topN"""
39+
return self.client.zrevrange(get_month_rank_key(), 0, n - 1, withscores=with_scores)

‎src/python/main/session_token.py

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
import os
2+
from _sha256 import sha256
3+
from time import time
4+
5+
from redis import Redis
6+
7+
# 会话默认过期时间,一个月
8+
DEFAULT_TIMEOUT = 30 * 24 * 3600
9+
10+
# 会话 token 及过期时间的 key
11+
SESSION_TOKEN_KEY = 'SESSION:TOKEN'
12+
SESSION_EXPIRE_TS_KEY = 'SESSION:EXPIRE'
13+
14+
# 会话状态
15+
SESSION_NOT_LOGIN = 'SESSION_NOT_LOGIN'
16+
SESSION_EXPIRE = 'SESSION_EXPIRE'
17+
SESSION_TOKEN_CORRECT = 'SESSION_TOKEN_CORRECT'
18+
SESSION_TOKEN_INCORRECT = 'SESSION_TOKEN_INCORRECT'
19+
20+
21+
def generate_token():
22+
"""生成一个随机的会话令牌"""
23+
return sha256(os.urandom(256)).hexdigest()
24+
25+
26+
class LoginSession:
27+
def __init__(self, client: Redis, user_id: str):
28+
self.client = client
29+
self.user_id = user_id
30+
31+
def create(self, timeout=DEFAULT_TIMEOUT) -> str:
32+
"""创建新的会话,并返回会话token"""
33+
session_token = generate_token()
34+
35+
# 设置过期时间
36+
expire_time = time() + timeout
37+
self.client.hset(SESSION_TOKEN_KEY, self.user_id, session_token)
38+
self.client.hset(SESSION_EXPIRE_TS_KEY, self.user_id, expire_time)
39+
return session_token
40+
41+
def validate(self, token) -> str:
42+
"""校验token"""
43+
session_token = self.client.hget(SESSION_TOKEN_KEY, self.user_id)
44+
expire_time = self.client.hget(SESSION_EXPIRE_TS_KEY, self.user_id)
45+
46+
if (session_token is None) or (expire_time is None):
47+
return SESSION_NOT_LOGIN
48+
49+
# 将字符串类型的时间转换为浮点数类型
50+
if time() > float(expire_time):
51+
return SESSION_EXPIRE
52+
53+
if session_token == token:
54+
return SESSION_TOKEN_CORRECT
55+
56+
return SESSION_TOKEN_INCORRECT
57+
58+
def destroy(self):
59+
"""销毁会话"""
60+
self.client.hdel(SESSION_TOKEN_KEY, self.user_id)
61+
self.client.hdel(SESSION_EXPIRE_TS_KEY, self.user_id)

‎src/python/main/shorten_url.py

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
from redis import Redis
2+
3+
4+
# 10进制数转换为36进制字符串
5+
def base10_to_base36(number: int) -> str:
6+
alphabets = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ'
7+
result = ''
8+
while number != 0:
9+
number, i = divmod(number, 36)
10+
result = alphabets[i] + result
11+
12+
return result or alphabets[0]
13+
14+
15+
URL_HASH_SHORT_SOURCE_KEY = 'url_hash:short_source'
16+
ID_COUNTER = 'short_url:id_counter'
17+
18+
19+
class URLShorten:
20+
21+
def __init__(self, client: Redis):
22+
self.client = client
23+
# 设置初始ID值,保留1-5位的短码,从6位的短码开始生成
24+
self.client.set(ID_COUNTER, 36 ** 5 - 1)
25+
26+
def shorten(self, source_url: str) -> str:
27+
"""对源网址进行缩短,返回短网址ID"""
28+
new_id = self.client.incr(ID_COUNTER)
29+
short_id = base10_to_base36(new_id)
30+
self.client.hset(URL_HASH_SHORT_SOURCE_KEY, short_id, source_url)
31+
return short_id
32+
33+
def restore(self, short_id: str) -> str:
34+
"""根据短网址ID,返回对应的源网址"""
35+
return self.client.hget(URL_HASH_SHORT_SOURCE_KEY, short_id)
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
import time
2+
3+
from redis import Redis
4+
5+
from main.distributed_lock import DistributedLock
6+
7+
8+
def test_distributed_lock():
9+
redis = Redis(decode_responses=True)
10+
redis.flushall()
11+
lock = DistributedLock(redis, 'bingo_distributed_lock')
12+
13+
assert lock.acquire(10) is True
14+
assert lock.acquire(10) is False
15+
16+
time.sleep(10)
17+
18+
assert lock.acquire() is True
19+
assert lock.release() is True
20+
21+
assert lock.acquire() is True

‎src/python/test/test_paginate.py

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
from unittest import TestCase
2+
3+
from redis import Redis
4+
5+
from main.paginate import Paginate
6+
7+
8+
def test_paginate():
9+
test_case = TestCase()
10+
redis = Redis(decode_responses=True)
11+
redis.flushall()
12+
topics = Paginate(redis, 'user-topics')
13+
14+
for i in range(24):
15+
topics.add(i)
16+
17+
# 取总数
18+
assert topics.get_size() == 24
19+
20+
# 以每页5条数据的方式取出第1页数据
21+
test_case.assertListEqual(['23', '22', '21', '20', '19'], topics.get_page(1, 5))
22+
23+
# 以每页10条数据的方式取出第1页数据
24+
test_case.assertListEqual(['23', '22', '21', '20', '19', '18', '17', '16', '15', '14'], topics.get_page(1, 10))
25+
26+
# 以每页5条数据的方式取出第5页数据
27+
test_case.assertListEqual(['3', '2', '1', '0'], topics.get_page(5, 5))

‎src/python/test/test_ranking_list.py

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
from unittest import TestCase
2+
3+
from redis import Redis
4+
5+
from main.ranking_list import Ranking
6+
7+
8+
def test_ranking():
9+
test_case = TestCase()
10+
redis = Redis(decode_responses=True)
11+
redis.flushall()
12+
ranking = Ranking(redis)
13+
14+
ranking.incr('bingo', 5)
15+
ranking.incr('iris', 3)
16+
ranking.incr('lily', 4)
17+
ranking.incr('helen', 6)
18+
19+
test_case.assertListEqual([('helen', 6.0), ('bingo', 5.0)], ranking.get_today_top_n(n=2, with_scores=True))
20+
test_case.assertListEqual([('helen', 6.0), ('bingo', 5.0)], ranking.get_week_top_n(n=2, with_scores=True))
21+
test_case.assertListEqual([('helen', 6.0), ('bingo', 5.0), ('lily', 4.0)], ranking.get_month_top_n(n=3, with_scores=True))

‎src/python/test/test_session_token.py

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
from redis import Redis
2+
3+
from main.session_token import LoginSession
4+
5+
6+
def test_session_token():
7+
redis = Redis(decode_responses=True)
8+
redis.flushall()
9+
session = LoginSession(redis, 'bingo')
10+
11+
user_token = session.create()
12+
13+
res = session.validate('this is a wrong token')
14+
assert res == 'SESSION_TOKEN_INCORRECT'
15+
16+
res = session.validate(user_token)
17+
assert res == 'SESSION_TOKEN_CORRECT'
18+
19+
session.destroy()
20+
res = session.validate(user_token)
21+
assert res == 'SESSION_NOT_LOGIN'

‎src/python/test/test_shorten_url.py

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
from redis import Redis
2+
3+
from main.shorten_url import URLShorten
4+
5+
6+
def test_shorten_url():
7+
redis = Redis(decode_responses=True)
8+
redis.flushall()
9+
url_shorten = URLShorten(redis)
10+
11+
short_id = url_shorten.shorten('https://github.com/yanglbme')
12+
assert short_id == '100000'
13+
14+
source_url = url_shorten.restore(short_id)
15+
assert source_url == 'https://github.com/yanglbme'
16+
17+
assert url_shorten.shorten('https://doocs.github.io') == '100001'

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /