arcanis/ffxivbis
1
0
Fork 0
mirror of https://github.com/arcan1s/ffxivbis.git synced 2025-04-25 09:47:18 +00:00
Code Issues Projects Releases Wiki Activity

use async database

Browse Source
This commit is contained in:
Evgenii Alekseev 2019-09-11 04:21:31 +03:00
parent 14c1e4842c
commit 98b20a073a
23 changed files with 199 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
setup.py
View File

@ -28,6 +28,7 @@ setup(
'aiohttp',
'aiohttp_jinja2',
'aiohttp_security',
'aiosqlite',
'Jinja2',
'passlib',
'requests',
@ -37,7 +38,7 @@ setup(
'pytest-runner'
],
tests_require=[
'pytest', 'pytest-aiohttp'
'pytest', 'pytest-aiohttp', 'pytest-asyncio'
],
include_package_data=True,

4
src/service/api/auth.py
View File

@ -19,11 +19,11 @@ class AuthorizationPolicy(AbstractAuthorizationPolicy):
self.database = database
async def authorized_userid(self, identity: str) -> Optional[str]:
user = self.database.get_user(identity)
user = await self.database.get_user(identity)
return identity if user is not None else None
async def permits(self, identity: str, permission: str, context: str = None) -> bool:
user = self.database.get_user(identity)
user = await self.database.get_user(identity)
if user is None:
return False
if user.username != identity:

4
src/service/api/views/api/bis.py
View File

@ -45,7 +45,7 @@ class BiSView(BiSBaseView):
try:
player_id = PlayerId(Job[data['job']], data['nick'])
piece: Piece = Piece.get(data) # type: ignore
self.bis_post(action, player_id, piece)
await self.bis_post(action, player_id, piece)
except Exception as e:
self.request.app.logger.exception('could not add bis')
@ -65,7 +65,7 @@ class BiSView(BiSBaseView):
try:
player_id = PlayerId(Job[data['job']], data['nick'])
link = self.bis_put(player_id, data['link'])
link = await self.bis_put(player_id, data['link'])
except Exception as e:
self.request.app.logger.exception('could not parse bis')

2
src/service/api/views/api/loot.py
View File

@ -45,7 +45,7 @@ class LootView(LootBaseView):
try:
player_id = PlayerId(Job[data['job']], data['nick'])
piece: Piece = Piece.get(data) # type: ignore
self.loot_post(action, player_id, piece)
await self.loot_post(action, player_id, piece)
except Exception as e:
self.request.app.logger.exception('could not add loot')

2
src/service/api/views/api/player.py
View File

@ -43,7 +43,7 @@ class PlayerView(PlayerBaseView):
return wrap_invalid_param(['action'], data)
try:
player_id = self.player_post(action, Job[data['job']], data['nick'], link, priority)
player_id = await self.player_post(action, Job[data['job']], data['nick'], link, priority)
except Exception as e:
self.request.app.logger.exception('could not add loot')

20
src/service/api/views/common/bis_base.py
View File

@ -16,8 +16,8 @@ from service.models.player import PlayerId
class BiSBaseView(View):
def bis_add(self, player_id: PlayerId, piece: Piece) -> Piece:
self.request.app['party'].set_item_bis(player_id, piece)
async def bis_add(self, player_id: PlayerId, piece: Piece) -> Piece:
await self.request.app['party'].set_item_bis(player_id, piece)
return piece
def bis_get(self, nick: Optional[str]) -> List[Piece]:
@ -28,21 +28,21 @@ class BiSBaseView(View):
]
return list(sum([player.bis.pieces for player in party], []))
def bis_post(self, action: str, player_id: PlayerId, piece: Piece) -> Optional[Piece]:
async def bis_post(self, action: str, player_id: PlayerId, piece: Piece) -> Optional[Piece]:
if action == 'add':
return self.bis_add(player_id, piece)
return await self.bis_add(player_id, piece)
elif action == 'remove':
return self.bis_remove(player_id, piece)
return await self.bis_remove(player_id, piece)
return None
def bis_put(self, player_id: PlayerId, link: str) -> str:
async def bis_put(self, player_id: PlayerId, link: str) -> str:
parser = AriyalaParser(self.request.app['config'])
items = parser.get(link, player_id.job.name)
for piece in items:
self.request.app['party'].set_item_bis(player_id, piece)
self.request.app['party'].set_bis_link(player_id, link)
await self.request.app['party'].set_item_bis(player_id, piece)
await self.request.app['party'].set_bis_link(player_id, link)
return link
def bis_remove(self, player_id: PlayerId, piece: Piece) -> Piece:
self.request.app['party'].remove_item_bis(player_id, piece)
async def bis_remove(self, player_id: PlayerId, piece: Piece) -> Piece:
await self.request.app['party'].remove_item_bis(player_id, piece)
return piece

6
src/service/api/views/common/login_base.py
View File

@ -16,13 +16,13 @@ from service.models.user import User
class LoginBaseView(View):
async def check_credentials(self, username: str, password: str) -> bool:
user = self.request.app['database'].get_user(username)
user = await self.request.app['database'].get_user(username)
if user is None:
return False
return md5_crypt.verify(password, user.password)
async def create_user(self, username: str, password: str, permission: str) -> None:
self.request.app['database'].insert_user(User(username, password, permission), False)
await self.request.app['database'].insert_user(User(username, password, permission), False)
async def login(self, username: str, password: str) -> None:
if await self.check_credentials(username, password):
@ -40,4 +40,4 @@ class LoginBaseView(View):
raise response
async def remove_user(self, username: str) -> None:
self.request.app['database'].delete_user(username)
await self.request.app['database'].delete_user(username)

14
src/service/api/views/common/loot_base.py
View File

@ -16,8 +16,8 @@ from service.models.upgrade import Upgrade
class LootBaseView(View):
def loot_add(self, player_id: PlayerId, piece: Piece) -> Piece:
self.request.app['party'].set_item(player_id, piece)
async def loot_add(self, player_id: PlayerId, piece: Piece) -> Piece:
await self.request.app['party'].set_item(player_id, piece)
return piece
def loot_get(self, nick: Optional[str]) -> List[Piece]:
@ -28,16 +28,16 @@ class LootBaseView(View):
]
return list(sum([player.loot for player in party], []))
def loot_post(self, action: str, player_id: PlayerId, piece: Piece) -> Optional[Piece]:
async def loot_post(self, action: str, player_id: PlayerId, piece: Piece) -> Optional[Piece]:
if action == 'add':
return self.loot_add(player_id, piece)
return await self.loot_add(player_id, piece)
elif action == 'remove':
return self.loot_remove(player_id, piece)
return await self.loot_remove(player_id, piece)
return None
def loot_put(self, piece: Union[Piece, Upgrade]) -> List[PlayerIdWithCounters]:
return self.request.app['loot'].suggest(piece)
def loot_remove(self, player_id: PlayerId, piece: Piece) -> Piece:
self.request.app['party'].remove_item(player_id, piece)
async def loot_remove(self, player_id: PlayerId, piece: Piece) -> Piece:
await self.request.app['party'].remove_item(player_id, piece)
return piece

16
src/service/api/views/common/player_base.py
View File

@ -17,16 +17,16 @@ from service.models.player import Player, PlayerId
class PlayerBaseView(View):
def player_add(self, job: Job, nick: str, link: Optional[str], priority: int) -> PlayerId:
async def player_add(self, job: Job, nick: str, link: Optional[str], priority: int) -> PlayerId:
player = Player(job, nick, BiS(), [], link, int(priority))
player_id = player.player_id
self.request.app['party'].set_player(player)
await self.request.app['party'].set_player(player)
if link:
parser = AriyalaParser(self.request.app['config'])
items = parser.get(link, job.name)
for piece in items:
self.request.app['party'].set_item_bis(player_id, piece)
await self.request.app['party'].set_item_bis(player_id, piece)
return player_id
@ -37,14 +37,14 @@ class PlayerBaseView(View):
if nick is None or player.nick == nick
]
def player_post(self, action: str, job: Job, nick: str, link: Optional[str], priority: int) -> Optional[PlayerId]:
async def player_post(self, action: str, job: Job, nick: str, link: Optional[str], priority: int) -> Optional[PlayerId]:
if action == 'add':
return self.player_add(job, nick, link, priority)
return await self.player_add(job, nick, link, priority)
elif action == 'remove':
return self.player_remove(job, nick)
return await self.player_remove(job, nick)
return None
def player_remove(self, job: Job, nick: str) -> PlayerId:
async def player_remove(self, job: Job, nick: str) -> PlayerId:
player_id = PlayerId(job, nick)
self.request.app['party'].remove_player(player_id)
await self.request.app['party'].remove_player(player_id)
return player_id

4
src/service/api/views/html/bis.py
View File

@ -65,7 +65,7 @@ class BiSHtmlView(BiSBaseView, PlayerBaseView):
if any(param not in data for param in required):
return wrap_invalid_param(required, data)
is_tome = (data.getone('is_tome', None) == 'on')
self.bis_post(data.getone('action'), player_id, # type: ignore
await self.bis_post(data.getone('action'), player_id, # type: ignore
Piece.get({'piece': data.getone('piece'), 'is_tome': is_tome})) # type: ignore
elif method == 'put':
@ -73,7 +73,7 @@ class BiSHtmlView(BiSBaseView, PlayerBaseView):
if any(param not in data for param in required):
return wrap_invalid_param(required, data)
self.bis_put(player_id, data.getone('bis')) # type: ignore
await self.bis_put(player_id, data.getone('bis')) # type: ignore
except Exception as e:
self.request.app.logger.exception('could not manage bis')

2
src/service/api/views/html/loot.py
View File

@ -60,7 +60,7 @@ class LootHtmlView(LootBaseView, PlayerBaseView):
try:
player_id = PlayerId.from_pretty_name(data.getone('player')) # type: ignore
is_tome = (data.getone('is_tome', None) == 'on')
self.loot_post(data.getone('action'), player_id, # type: ignore
await self.loot_post(data.getone('action'), player_id, # type: ignore
Piece.get({'piece': data.getone('piece'), 'is_tome': is_tome})) # type: ignore
except Exception as e:

2
src/service/api/views/html/player.py
View File

@ -58,7 +58,7 @@ class PlayerHtmlView(PlayerBaseView):
action = data.getone('action')
priority = data.getone('priority', 0)
link = data.getone('bis', None)
self.player_post(action, Job[data['job'].upper()], data['nick'], link, priority) # type: ignore
await self.player_post(action, Job[data['job'].upper()], data['nick'], link, priority) # type: ignore
except Exception as e:
self.request.app.logger.exception('could not manage players')

2
src/service/api/views/html/users.py
View File

@ -24,7 +24,7 @@ class UsersHtmlView(LoginBaseView):
users: List[User] = []
try:
users = self.request.app['database'].get_users()
users = await self.request.app['database'].get_users()
except Exception as e:
self.request.app.logger.exception('could not get users')
error = repr(e)

8
src/service/application/core.py
View File

@ -6,6 +6,7 @@
#
# License: 3-clause BSD, see https://opensource.org/licenses/BSD-3-Clause
#
import asyncio
import logging
from service.api.web import run_server, setup_service
@ -23,12 +24,15 @@ class Application:
self.logger = logging.getLogger('application')
def run(self) -> None:
loop = asyncio.get_event_loop()
database = Database.get(self.config)
database.migration()
party = Party.get(database)
party = loop.run_until_complete(Party.get(database))
admin = User(self.config.get('auth', 'root_username'), self.config.get('auth', 'root_password'), 'admin')
database.insert_user(admin, True)
loop.run_until_complete(database.insert_user(admin, True))
priority = self.config.get('settings', 'priority').split()
loot_selector = LootSelector(party, priority)

24
src/service/core/database.py
View File

@ -51,40 +51,40 @@ class Database:
def connection(self) -> str:
raise NotImplementedError
def delete_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
async def delete_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
raise NotImplementedError
def delete_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
async def delete_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
raise NotImplementedError
def delete_player(self, player_id: PlayerId) -> None:
async def delete_player(self, player_id: PlayerId) -> None:
raise NotImplementedError
def delete_user(self, username: str) -> None:
async def delete_user(self, username: str) -> None:
raise NotImplementedError
def get_party(self) -> List[Player]:
async def get_party(self) -> List[Player]:
raise NotImplementedError
def get_player(self, player_id: PlayerId) -> Optional[int]:
async def get_player(self, player_id: PlayerId) -> Optional[int]:
raise NotImplementedError
def get_user(self, username: str) -> Optional[User]:
async def get_user(self, username: str) -> Optional[User]:
raise NotImplementedError
def get_users(self) -> List[User]:
async def get_users(self) -> List[User]:
raise NotImplementedError
def insert_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
async def insert_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
raise NotImplementedError
def insert_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
async def insert_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
raise NotImplementedError
def insert_player(self, player: Player) -> None:
async def insert_player(self, player: Player) -> None:
raise NotImplementedError
def insert_user(self, user: User, hashed_password: bool) -> None:
async def insert_user(self, user: User, hashed_password: bool) -> None:
raise NotImplementedError
def migration(self) -> None:

33
src/service/core/party.py
View File

@ -31,50 +31,51 @@ class Party:
return list(self.players.values())
@classmethod
def get(cls: Type[Party], database: Database) -> Party:
async def get(cls: Type[Party], database: Database) -> Party:
obj = Party(database)
for player in database.get_party():
players = await database.get_party()
for player in players:
obj.players[player.player_id] = player
return obj
def set_bis_link(self, player_id: PlayerId, link: str) -> None:
async def set_bis_link(self, player_id: PlayerId, link: str) -> None:
with self.lock:
player = self.players[player_id]
player.link = link
self.database.insert_player(player)
await self.database.insert_player(player)
def remove_player(self, player_id: PlayerId) -> Optional[Player]:
self.database.delete_player(player_id)
async def remove_player(self, player_id: PlayerId) -> Optional[Player]:
await self.database.delete_player(player_id)
with self.lock:
player = self.players.pop(player_id, None)
return player
def set_player(self, player: Player) -> PlayerId:
async def set_player(self, player: Player) -> PlayerId:
player_id = player.player_id
self.database.insert_player(player)
await self.database.insert_player(player)
with self.lock:
self.players[player_id] = player
return player_id
def set_item(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
self.database.insert_piece(player_id, piece)
async def set_item(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
await self.database.insert_piece(player_id, piece)
with self.lock:
self.players[player_id].loot.append(piece)
def remove_item(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
self.database.delete_piece(player_id, piece)
async def remove_item(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
await self.database.delete_piece(player_id, piece)
with self.lock:
try:
self.players[player_id].loot.remove(piece)
except ValueError:
pass
def set_item_bis(self, player_id: PlayerId, piece: Piece) -> None:
self.database.insert_piece_bis(player_id, piece)
async def set_item_bis(self, player_id: PlayerId, piece: Piece) -> None:
await self.database.insert_piece_bis(player_id, piece)
with self.lock:
self.players[player_id].bis.set_item(piece)
def remove_item_bis(self, player_id: PlayerId, piece: Piece) -> None:
self.database.delete_piece_bis(player_id, piece)
async def remove_item_bis(self, player_id: PlayerId, piece: Piece) -> None:
await self.database.delete_piece_bis(player_id, piece)
with self.lock:
self.players[player_id].bis.remove_item(piece)

103
src/service/core/sqlite.py
View File

@ -31,13 +31,13 @@ class SQLiteDatabase(Database):
def connection(self) -> str:
return 'sqlite:///{path}'.format(path=self.database_path)
def delete_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
player = self.get_player(player_id)
async def delete_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
player = await self.get_player(player_id)
if player is None:
return
with SQLiteHelper(self.database_path) as cursor:
cursor.execute(
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute(
'''delete from loot
where loot_id in (
select loot_id from loot
@ -45,63 +45,70 @@ class SQLiteDatabase(Database):
)''',
(player, piece.name, getattr(piece, 'is_tome', True)))
def delete_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
player = self.get_player(player_id)
async def delete_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
player = await self.get_player(player_id)
if player is None:
return
with SQLiteHelper(self.database_path) as cursor:
cursor.execute(
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute(
'''delete from bis where player_id = ? and piece = ?''',
(player, piece.name))
def delete_player(self, player_id: PlayerId) -> None:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute('''delete from players where nick = ? and job = ?''',
async def delete_player(self, player_id: PlayerId) -> None:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute('''delete from players where nick = ? and job = ?''',
(player_id.nick, player_id.job.name))
def delete_user(self, username: str) -> None:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute('''delete from users where username = ?''', (username,))
async def delete_user(self, username: str) -> None:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute('''delete from users where username = ?''', (username,))
def get_party(self) -> List[Player]:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute('''select * from bis''')
bis_pieces = [Loot(row['player_id'], Piece.get(row)) for row in cursor.fetchall()]
cursor.execute('''select * from loot''')
loot_pieces = [Loot(row['player_id'], Piece.get(row)) for row in cursor.fetchall()]
cursor.execute('''select * from players''')
async def get_party(self) -> List[Player]:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute('''select * from bis''')
rows = await cursor.fetchall()
bis_pieces = [Loot(row['player_id'], Piece.get(row)) for row in rows]
await cursor.execute('''select * from loot''')
rows = await cursor.fetchall()
loot_pieces = [Loot(row['player_id'], Piece.get(row)) for row in rows]
await cursor.execute('''select * from players''')
rows = await cursor.fetchall()
party = {
row['player_id']: Player(Job[row['job']], row['nick'], BiS(), [], row['bis_link'], row['priority'])
for row in cursor.fetchall()
for row in rows
}
return self.set_loot(party, bis_pieces, loot_pieces)
def get_player(self, player_id: PlayerId) -> Optional[int]:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute('''select player_id from players where nick = ? and job = ?''',
async def get_player(self, player_id: PlayerId) -> Optional[int]:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute('''select player_id from players where nick = ? and job = ?''',
(player_id.nick, player_id.job.name))
player = cursor.fetchone()
player = await cursor.fetchone()
return player['player_id'] if player is not None else None
def get_user(self, username: str) -> Optional[User]:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute('''select * from users where username = ?''', (username,))
user = cursor.fetchone()
async def get_user(self, username: str) -> Optional[User]:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute('''select * from users where username = ?''', (username,))
user = await cursor.fetchone()
return User(user['username'], user['password'], user['permission']) if user is not None else None
def get_users(self) -> List[User]:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute('''select * from users''')
return [User(user['username'], user['password'], user['permission']) for user in cursor.fetchall()]
async def get_users(self) -> List[User]:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute('''select * from users''')
users = await cursor.fetchall()
return [User(user['username'], user['password'], user['permission']) for user in users]
def insert_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
player = self.get_player(player_id)
async def insert_piece(self, player_id: PlayerId, piece: Union[Piece, Upgrade]) -> None:
player = await self.get_player(player_id)
if player is None:
return
with SQLiteHelper(self.database_path) as cursor:
cursor.execute(
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute(
'''insert into loot
(created, piece, is_tome, player_id)
values
@ -109,13 +116,13 @@ class SQLiteDatabase(Database):
(Database.now(), piece.name, getattr(piece, 'is_tome', True), player)
)
def insert_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
player = self.get_player(player_id)
async def insert_piece_bis(self, player_id: PlayerId, piece: Piece) -> None:
player = await self.get_player(player_id)
if player is None:
return
with SQLiteHelper(self.database_path) as cursor:
cursor.execute(
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute(
'''replace into bis
(created, piece, is_tome, player_id)
values
@ -123,9 +130,9 @@ class SQLiteDatabase(Database):
(Database.now(), piece.name, piece.is_tome, player)
)
def insert_player(self, player: Player) -> None:
with SQLiteHelper(self.database_path) as cursor:
cursor.execute(
async def insert_player(self, player: Player) -> None:
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute(
'''replace into players
(created, nick, job, bis_link, priority)
values
@ -133,10 +140,10 @@ class SQLiteDatabase(Database):
(Database.now(), player.nick, player.job.name, player.link, player.priority)
)
def insert_user(self, user: User, hashed_password: bool) -> None:
async def insert_user(self, user: User, hashed_password: bool) -> None:
password = user.password if hashed_password else md5_crypt.hash(user.password)
with SQLiteHelper(self.database_path) as cursor:
cursor.execute(
async with SQLiteHelper(self.database_path) as cursor:
await cursor.execute(
'''replace into users
(username, password, permission)
values

18
src/service/core/sqlite_helper.py
View File

@ -7,13 +7,13 @@
# License: 3-clause BSD, see https://opensource.org/licenses/BSD-3-Clause
#
# because sqlite3 does not support context management
import sqlite3
import aiosqlite
from types import TracebackType
from typing import Any, Dict, Optional, Type
def dict_factory(cursor: sqlite3.Cursor, row: sqlite3.Row) -> Dict[str, Any]:
def dict_factory(cursor: aiosqlite.Cursor, row: aiosqlite.Row) -> Dict[str, Any]:
return {
key: value
for key, value in zip([column[0] for column in cursor.description], row)
@ -24,13 +24,13 @@ class SQLiteHelper():
def __init__(self, database_path: str) -> None:
self.database_path = database_path
def __enter__(self) -> sqlite3.Cursor:
self.conn = sqlite3.connect(self.database_path)
async def __aenter__(self) -> aiosqlite.Cursor:
self.conn = await aiosqlite.connect(self.database_path)
self.conn.row_factory = dict_factory
self.conn.execute('''pragma foreign_keys = on''')
return self.conn.cursor()
await self.conn.execute('''pragma foreign_keys = on''')
return await self.conn.cursor()
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException],
async def __aexit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[BaseException],
traceback: Optional[TracebackType]) -> None:
self.conn.commit()
self.conn.close()
await self.conn.commit()
await self.conn.close()

6
test/conftest.py
View File

@ -105,14 +105,14 @@ def player2(bis2: BiS) -> Player:
@pytest.fixture
def selector(party: Party, player: Player, player2: Player,
async def selector(party: Party, player: Player, player2: Player,
head_with_upgrade: Piece, weapon: Piece) -> LootSelector:
obj = LootSelector(party)
obj.party.set_player(player)
await obj.party.set_player(player)
player.bis.set_item(weapon)
obj.party.set_player(player2)
await obj.party.set_player(player2)
player2.bis.set_item(head_with_upgrade)
player2.bis.set_item(weapon)

72
test/test_party.py
View File

@ -4,92 +4,92 @@ from service.models.piece import Piece
from service.models.player import Player
def test_set_player(party: Party, player: Player) -> None:
async def test_set_player(party: Party, player: Player) -> None:
assert len(party.players) == 0
party.set_player(player)
await party.set_player(player)
assert len(party.players) == 1
def test_remove_player(party: Party, player: Player) -> None:
party.remove_player(player.player_id)
async def test_remove_player(party: Party, player: Player) -> None:
await party.remove_player(player.player_id)
assert len(party.players) == 0
party.set_player(player)
party.remove_player(player.player_id)
await party.set_player(player)
await party.remove_player(player.player_id)
assert len(party.players) == 0
def test_set_bis_link(party: Party, player: Player, bis_link: str) -> None:
party.set_player(player)
async def test_set_bis_link(party: Party, player: Player, bis_link: str) -> None:
await party.set_player(player)
party.set_bis_link(player.player_id, bis_link)
await party.set_bis_link(player.player_id, bis_link)
assert player.link == bis_link
def test_set_item(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
party.set_player(player)
async def test_set_item(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
await party.set_player(player)
party.set_item(player.player_id, weapon)
await party.set_item(player.player_id, weapon)
assert abs(player.loot_count(weapon)) == 1
assert abs(player.loot_count(head_with_upgrade)) == 0
def test_remove_item(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
party.set_player(player)
async def test_remove_item(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
await party.set_player(player)
party.remove_item(player.player_id, weapon)
await party.remove_item(player.player_id, weapon)
assert abs(player.loot_count(weapon)) == 0
assert abs(player.loot_count(head_with_upgrade)) == 0
party.set_item(player.player_id, weapon)
await party.set_item(player.player_id, weapon)
assert abs(player.loot_count(weapon)) == 1
assert abs(player.loot_count(head_with_upgrade)) == 0
party.remove_item(player.player_id, weapon)
await party.remove_item(player.player_id, weapon)
assert abs(player.loot_count(weapon)) == 0
assert abs(player.loot_count(head_with_upgrade)) == 0
def test_set_item_bis(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
party.set_player(player)
async def test_set_item_bis(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
await party.set_player(player)
party.set_item_bis(player.player_id, head_with_upgrade)
await party.set_item_bis(player.player_id, head_with_upgrade)
assert player.bis.has_piece(head_with_upgrade)
assert not player.bis.has_piece(weapon)
def test_remove_item_bis(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
party.set_player(player)
async def test_remove_item_bis(party: Party, player: Player, head_with_upgrade: Piece, weapon: Piece) -> None:
await party.set_player(player)
party.remove_item_bis(player.player_id, head_with_upgrade)
await party.remove_item_bis(player.player_id, head_with_upgrade)
assert not player.bis.has_piece(head_with_upgrade)
assert not player.bis.has_piece(weapon)
party.set_item_bis(player.player_id, head_with_upgrade)
await party.set_item_bis(player.player_id, head_with_upgrade)
assert player.bis.has_piece(head_with_upgrade)
assert not player.bis.has_piece(weapon)
party.set_item_bis(player.player_id, weapon)
await party.set_item_bis(player.player_id, weapon)
assert player.bis.has_piece(head_with_upgrade)
assert player.bis.has_piece(weapon)
party.remove_item_bis(player.player_id, head_with_upgrade)
await party.remove_item_bis(player.player_id, head_with_upgrade)
assert not player.bis.has_piece(head_with_upgrade)
assert player.bis.has_piece(weapon)
def test_get(party: Party, database: Database, player: Player, head_with_upgrade: Piece, weapon: Piece,
bis_link: str) -> None:
party.set_player(player)
party.set_bis_link(player.player_id, bis_link)
party.set_item_bis(player.player_id, head_with_upgrade)
party.set_item_bis(player.player_id, weapon)
party.set_item(player.player_id, weapon)
async def test_get(party: Party, database: Database, player: Player, head_with_upgrade: Piece,
weapon: Piece, bis_link: str) -> None:
await party.set_player(player)
await party.set_bis_link(player.player_id, bis_link)
await party.set_item_bis(player.player_id, head_with_upgrade)
await party.set_item_bis(player.player_id, weapon)
await party.set_item(player.player_id, weapon)
new_party = Party.get(database)
new_party = await Party.get(database)
assert party.party == new_party.party
party.remove_player(player.player_id)
new_party = Party.get(database)
await party.remove_player(player.player_id)
new_party = await Party.get(database)
assert party.party == new_party.party

12
test/test_view_bis.py
View File

@ -8,9 +8,9 @@ from service.models.player import Player
async def test_bis_get(server: Any, party: Party, player: Player, player2: Player,
head_with_upgrade: Piece, weapon: Piece) -> None:
party.set_item_bis(player.player_id, weapon)
party.set_item_bis(player2.player_id, weapon)
party.set_item_bis(player2.player_id, head_with_upgrade)
await party.set_item_bis(player.player_id, weapon)
await party.set_item_bis(player2.player_id, weapon)
await party.set_item_bis(player2.player_id, head_with_upgrade)
response = await server.get('/api/v1/party/bis')
assert response.status == 200
@ -19,9 +19,9 @@ async def test_bis_get(server: Any, party: Party, player: Player, player2: Playe
async def test_bis_get_with_filter(server: Any, party: Party, player: Player, player2: Player,
head_with_upgrade: Piece, weapon: Piece) -> None:
party.set_item_bis(player.player_id, weapon)
party.set_item_bis(player2.player_id, weapon)
party.set_item_bis(player2.player_id, head_with_upgrade)
await party.set_item_bis(player.player_id, weapon)
await party.set_item_bis(player2.player_id, weapon)
await party.set_item_bis(player2.player_id, head_with_upgrade)
response = await server.get('/api/v1/party/bis', params={'nick': player.nick})
assert response.status == 200

8
test/test_view_loot.py
View File

@ -7,8 +7,8 @@ from service.models.player import Player
async def test_loot_get(server: Any, party: Party, player: Player, player2: Player, weapon: Piece) -> None:
party.set_item(player.player_id, weapon)
party.set_item(player2.player_id, weapon)
await party.set_item(player.player_id, weapon)
await party.set_item(player2.player_id, weapon)
response = await server.get('/api/v1/party/loot')
assert response.status == 200
@ -16,8 +16,8 @@ async def test_loot_get(server: Any, party: Party, player: Player, player2: Play
async def test_loot_get_with_filter(server: Any, party: Party, player: Player, player2: Player, weapon: Piece) -> None:
party.set_item(player.player_id, weapon)
party.set_item(player2.player_id, weapon)
await party.set_item(player.player_id, weapon)
await party.set_item(player2.player_id, weapon)
response = await server.get('/api/v1/party/loot', params={'nick': player.nick})
assert response.status == 200

8
test/test_view_player.py
View File

@ -7,7 +7,7 @@ from service.models.player import Player
async def test_players_get(server: Any, party: Party, player: Player) -> None:
party.set_player(player)
await party.set_player(player)
response = await server.get('/api/v1/party')
assert response.status == 200
@ -15,7 +15,7 @@ async def test_players_get(server: Any, party: Party, player: Player) -> None:
async def test_players_get_with_filter(server: Any, party: Party, player: Player, player2: Player) -> None:
party.set_player(player)
await party.set_player(player)
response = await server.get('/api/v1/party', params={'nick': player.nick})
assert response.status == 200
@ -27,7 +27,7 @@ async def test_players_get_with_filter(server: Any, party: Party, player: Player
async def test_players_post_add(server: Any, party: Party, player: Player) -> None:
party.remove_player(player.player_id)
await party.remove_player(player.player_id)
response = await server.get('/api/v1/party', params={'nick': player.nick})
assert response.status == 200
@ -64,7 +64,7 @@ async def test_players_post_remove(server: Any, party: Party, player: Player) ->
async def test_players_post_add_with_link(server: Any, party: Party, player: Player,
bis_link: str, bis_set: List[Piece]) -> None:
party.remove_player(player.player_id)
await party.remove_player(player.player_id)
response = await server.get('/api/v1/party', params={'nick': player.nick})
assert response.status == 200