Refactored to handle sessions cleaner. Command logic now is in Commands.py. This should make adding new commands easier and keep logic the same in MC.

doc_update
Joey Hines 2018-07-22 18:56:07 -05:00
parent 77814ce42d
commit 93827f5f90
7 changed files with 473 additions and 215 deletions

View File

@ -27,5 +27,11 @@ class EntryNameNotUniqueError(DataBaseError):
class StringTooLong(DataBaseError):
'''Given string is too long.'''
class DatabaseValueError(DataBaseError):
''''String too long or number too large'''
class ItemNotFound(DataBaseError):
'''No item matches found in database'''

157
Commands.py 100644
View File

@ -0,0 +1,157 @@
from DatabaseInterface import *
class Commands:
def __init__(self, db_engine_arg):
self.interface = DatabaseInterface(db_engine_arg)
def get_player(self, session, discord_uuid=None, mc_uuid=None):
if discord_uuid is not None:
player = self.interface.find_player_by_discord_uuid(session, discord_uuid)
elif mc_uuid is not None:
player = self.interface.find_player_by_mc_uuid(session, mc_uuid)
else:
raise AttributeError
return player
def register(self, player_name, discord_uuid):
session = self.interface.database.Session()
try:
player = self.interface.add_player(session, player_name, discord_uuid)
player_name = player.name
finally:
session.close()
return player_name
def addbase(self, x_pos, z_pos, base_name=None, discord_uuid=None, mc_uuid=None):
session = self.interface.database.Session()
try:
player = self.get_player(session, discord_uuid, mc_uuid)
if len(self.interface.find_location_by_owner(session, player)) == 0:
if base_name is None:
base_name = "{}'s Base".format()
elif base_name is None:
raise EntryNameNotUniqueError
base = self.interface.add_base(session, player, base_name, x_pos, z_pos)
base_name = base.name
except EntryNameNotUniqueError:
session.rollback()
raise EntryNameNotUniqueError
finally:
session.close()
return base_name
def addshop(self, x_pos, z_pos, shop_name=None, discord_uuid=None, mc_uuid=None):
session = self.interface.database.Session()
try:
player = self.get_player(session, discord_uuid, mc_uuid)
if len(self.interface.find_shop_by_owner(session, player)) == 0:
if shop_name is None:
shop_name = "{}'s Shop".format()
elif shop_name is None:
raise EntryNameNotUniqueError
shop = self.interface.add_shop(session, player, shop_name, x_pos, z_pos)
shop_name = shop.name
finally:
session.close()
return shop_name
def tunnel(self, tunnel_color, tunnel_number, location_name, discord_uuid=None, mc_uuid=None):
session = self.interface.database.Session()
try:
player = self.get_player(session, discord_uuid, mc_uuid)
tunnel = self.interface.add_tunnel(session, player, tunnel_color, tunnel_number, location_name)
tunnel_info = tunnel.__str__()
except EntryNameNotUniqueError:
session.rollback()
raise EntryNameNotUniqueError
finally:
session.close()
return tunnel_info
def find(self, search):
session = self.interface.database.Session()
try:
result = self.interface.search_all_fields(session, search)
finally:
session.close()
return result
def delete(self, name, discord_uuid=None, mc_uuid=None):
session = self.interface.database.Session()
try:
player = self.get_player(session, discord_uuid, mc_uuid)
self.interface.delete_location(session, player, name)
finally:
session.close()
def findaround(self, x_pos, z_pos, radius=200, dimension='Overworld'):
session = self.interface.database.Session()
try:
loc_list = self.interface.find_location_around(session, x_pos, z_pos, radius, dimension)
finally:
session.close()
return loc_list
def additem(self, item_name, quantity, diamond_price, shop_name, discord_uuid=None, mc_uuid=None):
session = self.interface.database.Session()
try:
player = self.get_player(session, discord_uuid, mc_uuid)
shop_list = self.interface.find_shop_by_owner(session, player)
if shop_name is None:
if len(shop_list) == 1:
shop_name = shop_list[0].name
else:
raise LocationInitError
item_listing = self.interface.add_item(session, player, shop_name, item_name, diamond_price, quantity)
item_listing_str = item_listing.__str__()
finally:
session.close()
return item_listing_str
def selling(self, item_name: str):
session = self.interface.database.Session()
try:
shop_list = self.interface.find_shop_selling_item(session, item_name)
finally:
session.close()
return shop_list
def info(self, location_name):
session = self.interface.database.Session()
try:
loc = self.interface.find_location_by_name(session, location_name)[0].full_str()
finally:
session.close()
return loc

View File

@ -44,21 +44,18 @@ class DatabaseInterface:
return item
def add_player(self, session, player_name):
def add_player(self, session, player_name, discord_uuid=None):
try:
player = self.find_player(session, player_name)
except PlayerNotFound:
uuid = grab_UUID(player_name)
mc_uuid = grab_UUID(player_name)
try:
player = self.find_player_by_mc_uuid(session, uuid)
player = self.find_player_by_mc_uuid(session, mc_uuid)
player.name = player_name
except PlayerNotFound:
player = Player(player_name, uuid)
player = Player(player_name, discord_uuid)
self.database.add_object(session, player)
player.name = player_name
session.commit()
return player
def find_location_by_name(self, session, name):
@ -94,7 +91,7 @@ class DatabaseInterface:
expr = (Location.x < x_pos + radius + 1) & (Location.x > x_pos - radius - 1) & (Location.z < z_pos + radius + 1) \
& (Location.z > z_pos - radius - 1) & (Location.dimension == dimension_obj)
return self.database.query_by_filter(session, Location, expr)
return list_to_string(self.database.query_by_filter(session, Location, expr))
def find_tunnel_by_owner(self, session, owner):
expr = Tunnel.owner == owner
@ -111,12 +108,7 @@ class DatabaseInterface:
def find_shop_selling_item(self, session, item_name):
listings = self.find_item(session, item_name)
shops = []
for listing in listings:
shops.append(listing.selling_info())
return shops
return list_to_string(listings)
def find_player(self, session, player_name):
expr = Player.name.ilike(player_name)
@ -177,4 +169,13 @@ def check_similarity(a, b):
if (ratio > 0.6) or (a[0] == b[0]):
return True
else:
return False
return False
def list_to_string(loc_list, str_format='{}\n{}'):
loc_string = ''
for loc in loc_list:
loc_string = str_format.format(loc_string, loc)
return loc_string

View File

@ -43,17 +43,19 @@ class GeoffreyDatabase:
session.add(obj)
session.commit()
except IntegrityError:
session.rollback()
raise EntryNameNotUniqueError
except DataError:
raise DatabaseValueError
except:
session.rollback()
raise StringTooLong
def query_by_filter(self, session, obj_type, * args):
filter_value = self.combine_filter(args)
return session.query(obj_type).filter(filter_value).all()
def delete_entry(self, session, obj_type, * args):
filter_value = self.combine_filter(args)
entry = session.query(obj_type).filter(filter_value)
@ -63,7 +65,7 @@ class GeoffreyDatabase:
else:
raise DeleteEntryError
session.close()
session.close()
def print_database(self, session, obj_type):
obj_list = session.query(obj_type).all()
@ -192,7 +194,7 @@ class Location(SQL_Base):
raise LocationInitError
def dynmap_link(self):
return 'http://24carrotcraft.com:8123/?worldname=season3&mapname=surface&zoom=4&x={}&y=65&z={}'.\
return '<http://24carrotcraft.com:8123/?worldname=season3&mapname=surface&zoom=4&x={}&y=65&z={}>'.\
format(self.x, self.z)
def pos_to_str(self):
@ -223,11 +225,11 @@ class Shop(Location):
def inv_to_str(self):
if len(self.inventory.all()) != 0:
inv = '\n\t*Inventory:*'
str_format = '{}\n\t\t{}'
inv = '\n**Inventory**:'
str_format = '{}\n{}'
for item in self.inventory:
inv = str_format.format(inv, item)
inv = str_format.format(inv, item.listing_str())
return inv
else:
@ -260,10 +262,10 @@ class ItemListing(SQL_Base):
self.amount = amount
self.shop = shop
def selling_info(self):
return 'Shop: **{}**, {}'.format(self.shop.name, self.__str__())
def listing_str(self):
return 'Item: **{}**, Price: **{}** for **{}**D'.format(self.name, self.amount, self.price)
def __str__(self):
return "Item: **{}**, Price: **{}** for **{}**D".format(self.name, self.amount, self.price)
return 'Shop: **{}**, {}'.format(self.shop.name, self.listing_str())

View File

@ -1,11 +1,10 @@
from discord.ext import commands
from DiscordDatabaseInterface import *
from Commands import *
from BotErrors import *
from MinecraftAccountInfoGrabber import *
from itertools import zip_longest
from BotConfig import *
TOKEN = ''
command_prefix = '?'
description = '''
Geoffrey started his life as an inside joke none of you will understand.
@ -38,25 +37,18 @@ async def on_command_error(error, ctx):
elif isinstance(error, commands.UserInputError):
error_str = 'Invalid syntax for {} you ding dong, please read ?help {}.'\
.format(ctx.invoked_with, ctx.invoked_with)
database_interface.database.session.rollback()
elif isinstance(error.original, UsernameLookupFailed):
error_str = error.original.__doc__
elif isinstance(error.original, OverflowError):
error_str = 'Wow buddy, that\'s a big number. Please don\'t do that.'
database_interface.database.session.rollback()
elif isinstance(error.original, PlayerNotFound):
error_str = 'Make sure to use ?register first you ding dong.'
database_interface.database.session.rollback()
elif isinstance(error.original, EntryNameNotUniqueError):
error_str = 'An entry in the database already has that name ding dong.'
database_interface.database.session.rollback()
elif isinstance(error.original, StringTooLong):
error_str = 'Use a shorter name you, dong ding.'
database_interface.database.session.rollback()
elif isinstance(error.original, DatabaseValueError):
error_str = 'Use a shorter name or a smaller value, dong ding.'
else:
error_str = bad_error_message.format(ctx.invoked_with, error)
await bot.send_message(ctx.message.channel, error_str)
bot.send_message(ctx.message.channel, '**Error Running Command:** {}'.format(error_str))
@bot.command()
@ -74,17 +66,14 @@ async def register(ctx):
You must do this before adding entries to the database.
'''
session = database_interface.database.Session()
try:
player_name = get_nickname(ctx.message.author)
database_interface.add_player(session, player_name, ctx.message.author.id)
bot_commands.register(player_name, ctx.message.author.id)
await bot.say('{}, you have been added to the database.'.format(ctx.message.author.mention))
except AttributeError:
await bot.say('{}, run this command on 24CC whoever you are'.format(ctx.message.author.mention))
return
except LocationInitError:
raise commands.UserInputError
session.close()
await bot.say('{}, you have been added to the database.'.format(ctx.message.author.mention))
@bot.command(pass_context=True)
@ -94,24 +83,23 @@ async def addbase(ctx, x_pos: int, z_pos: int, * args):
The name is optional.
?addbase [X Coordinate] [Y Coordinate] [Z Coordinate] [Base Name]
'''
session = database_interface.database.Session()
if len(args) > 0:
name = ' '.join(args)
else:
name = '{}\'s_Base'.format(database_interface.find_player_by_discord_uuid(session, ctx.message.author.id).name)
name = None
try:
base = database_interface.add_location(session, ctx.message.author.id, name, x_pos, z_pos)
base = bot_commands.addbase(x_pos, z_pos, base_name=name, discord_uuid=ctx.message.author.id)
await bot.say('{}, your base has been added to the database: {}'.format(ctx.message.author.mention, base))
except LocationInitError:
raise commands.UserInputError
except EntryNameNotUniqueError:
await bot.say('{}, a based called {} already exists. You need to specify a different name.'.format(
ctx.message.author.mention, name))
return
await bot.say('{}, your base named **{}** located at {} has been added'
' to the database.'.format(ctx.message.author.mention, base.name, base.pos_to_str()))
session.close()
if name is None:
await bot.say('{}, you already have one base in the database, you need to specify a base'
' name'.format(ctx.message.author.mention))
else:
await bot.say('{}, a base called **{}** already exists. You need to specify a different name.'.format(
ctx.message.author.mention, name))
@bot.command(pass_context=True)
@ -121,26 +109,23 @@ async def addshop(ctx, x_pos: int, z_pos: int, *args):
The name is optional.
?addshop [X Coordinate] [Y Coordinate] [Z Coordinate] [Shop Name]
'''
session = database_interface.database.Session()
if len(args) > 0:
name = ' '.join(args)
else:
name = '{}\'s_Shop'.format(database_interface.find_player_by_discord_uuid(session, ctx.message.author.id).name)
name = None
try:
shop = database_interface.add_shop(session, ctx.message.author.id, name, x_pos, z_pos)
shop = bot_commands.addshop(x_pos, z_pos, shop_name=name, discord_uuid=ctx.message.author.id)
await bot.say('{}, your shop has been added to the database: {}'.format(ctx.message.author.mention, shop))
except LocationInitError:
raise commands.UserInputError
except EntryNameNotUniqueError:
await bot.say('{}, a shop called **{}** already exists. You need to specify a different name.'.format(
ctx.message.author.mention, name))
return
await bot.say('{}, your shop named **{}** located at {} has been added'
' to the database.'.format(ctx.message.author.mention, shop.name, shop.pos_to_str()))
session.close()
if name is None:
await bot.say('{}, you already have one shop in the database, you need to specify a shop name'.format(
ctx.message.author.mention))
else:
await bot.say('{}, a shop called **{}** already exists. You need to specify a different name.'.format(
ctx.message.author.mention, name))
@bot.command(pass_context=True)
async def tunnel(ctx, tunnel_color: str, tunnel_number: int, *args):
@ -149,14 +134,14 @@ async def tunnel(ctx, tunnel_color: str, tunnel_number: int, *args):
The location name is optional. If the location has a tunnel, it is updated.
?addtunnel [Tunnel Color] [Tunnel_Number] [Location Name]
'''
session = database_interface.database.Session()
try:
if len(args) == 0:
location_name = None
else:
if len(args) > 0:
location_name = ' '.join(args)
else:
location_name = None
database_interface.add_tunnel(session, ctx.message.author.id, tunnel_color, tunnel_number, location_name)
bot_commands.tunnel(tunnel_color, tunnel_number, discord_uuid=ctx.message.author.id, location_name=location_name)
await bot.say('{}, your tunnel has been added to the database'.format(ctx.message.author.mention))
except EntryNameNotUniqueError:
await bot.say('{}, you already have one tunnel in the database, please specify a location.'.format(
ctx.message.author.mention))
@ -164,15 +149,9 @@ async def tunnel(ctx, tunnel_color: str, tunnel_number: int, *args):
except LocationLookUpError:
await bot.say('{}, you do not have a location called **{}**.'.format(
ctx.message.author.mention, args[0]))
return
except ValueError:
raise commands.UserInputError
await bot.say('{}, your tunnel has been added to the database'.format(ctx.message.author.mention))
session.close()
@bot.command(pass_context=True)
async def find(ctx, search: str):
@ -180,33 +159,26 @@ async def find(ctx, search: str):
Finds all the locations and tunnels matching the search term
?find [Search]
'''
session = database_interface.database.Session()
try:
result = database_interface.search_all_fields(session, search)
result = bot_commands.find(search)
await bot.say('{}, The following entries match **{}**:\n{}'.format(ctx.message.author.mention, search, result))
except LocationLookUpError:
await bot.say('{}, no matches **{}** were found in the database'.format(ctx.message.author.mention, search))
session.close()
@bot.command(pass_context=True)
async def delete(ctx, * args):
'''
Deletes a location from the database.
?delete [Location name]
'''
session = database_interface.database.Session()
try:
name = ' '.join(args)
database_interface.delete_location(session, ctx.message.author.id, name)
commands.delete(name, discord_uuid=ctx.message.author.id)
await bot.say('{}, your location named **{}** has been deleted.'.format(ctx.message.author.mention, name))
except (DeleteEntryError, PlayerNotFound):
await bot.say('{}, you do not have a location named **{}**.'.format(ctx.message.author.mention, name))
session.close()
@bot.command(pass_context=True)
async def findaround(ctx, x_pos: int, z_pos: int, * args):
@ -220,11 +192,11 @@ async def findaround(ctx, x_pos: int, z_pos: int, * args):
Optional Flags:
-d [dimension]
'''
session = database_interface.database.Session()
radius = 200
dimension = 'Overworld'
try:
radius = 200
dimension = 'Overworld'
if len(args) > 0:
if args[0] == '-d':
dimension = args[1]
@ -235,21 +207,17 @@ async def findaround(ctx, x_pos: int, z_pos: int, * args):
if args[1] == '-d':
dimension = args[2]
base_list = database_interface.find_location_around(session, x_pos, z_pos, radius, dimension)
base_string = bot_commands.findaround(x_pos, z_pos, radius, dimension)
if len(base_list) != 0:
base_string = loc_list_to_string(base_list, '{} \n{}')
await bot.say('{}, there are **{}** locations(s) within **{}** blocks of that point: \n {}'.format(
ctx.message.author.mention, len(base_list), radius, base_string))
if len(base_string) != 0:
await bot.say('{}, the following locations(s) within **{}** blocks of that point: \n {}'.format(
ctx.message.author.mention, radius, base_string))
else:
await bot.say('{}, there are no locations within {} blocks of that point'
.format(ctx.message.author.mention, radius))
except ValueError:
raise commands.UserInputError
session.close()
@bot.command(pass_context=True)
async def additem(ctx, item_name: str, quantity: int, diamond_price: int, * args):
@ -259,19 +227,13 @@ async def additem(ctx, item_name: str, quantity: int, diamond_price: int, * args
?additem [Item Name] [Quantity] [Price] [Shop name]
'''
session = database_interface.database.Session()
try:
shop_list = database_interface.find_shop_by_owner_uuid(session, ctx.message.author.id)
if len(shop_list) == 1:
shop_name = shop_list[0].name
if len(args) > 0:
shop_name = ' '.join(args)
else:
if len(args) == 0:
raise LocationInitError
else:
shop_name = ' '.join(args)
shop_name = None
database_interface.add_item(session, ctx.message.author.id, shop_name, item_name, diamond_price, quantity)
bot_commands.additem(item_name, quantity, diamond_price, shop_name=shop_name)
await bot.say('{}, **{}** has been added to the inventory of **{}**.'.format(ctx.message.author.mention,
item_name, shop_name))
except PlayerNotFound:
@ -281,9 +243,7 @@ async def additem(ctx, item_name: str, quantity: int, diamond_price: int, * args
.format(ctx.message.author.mention))
except LocationLookUpError:
await bot.say('{}, you don\'t have any shops named **{}** in the database.'.format(ctx.message.author.mention,
shop_name))
session.close()
shop_name))
@bot.command(pass_context=True)
@ -293,14 +253,11 @@ async def selling(ctx, item_name: str):
?selling [item]
'''
session = database_interface.database.Session()
shop_list = database_interface.find_shop_selling_item(session, item_name)
shop_list_str = loc_list_to_string(shop_list)
await bot.say('{}, the following shops sell **{}**: \n{}'.format(ctx.message.author.mention, item_name,
shop_list_str))
session.close()
try:
result = bot_commands.selling(item_name)
await bot.say('{}, the following shops sell **{}**: \n{}'.format(ctx.message.author.mention, item_name, result))
except ItemNotFound:
await bot.say('{}, bo shops sell {}'.format(ctx.message.author.mention, item_name))
@bot.command(pass_context=True)
@ -312,20 +269,21 @@ async def info(ctx, * args):
?info [Location Name]
'''
session = database_interface.database.Session()
try:
name = ' '.join(args)
loc = database_interface.find_location_by_name(session, name)[0]
if len(args) > 0:
name = ' '.join(args)
else:
raise commands.UserInputError
info_str = bot_commands.info(name)
await bot.say(info_str)
except IndexError:
await bot.say('{}, no locations in the database match {}.'.format(ctx.message.author.mention, name))
return
await bot.say('{}'.format(loc.full_str()))
session.close()
# Helper Functions ************************************************************
def get_nickname(discord_user):
if discord_user.nick is None:
name = discord_user.display_name
@ -338,15 +296,6 @@ def get_nickname(discord_user):
return name
def loc_list_to_string(loc_list, str_format='{}\n{}'):
loc_string = ''
for loc in loc_list:
loc_string = str_format.format(loc_string, loc)
return loc_string
def get_args_dict(args):
if len(args) != 0:
return dict(zip_longest(*[iter(args)] * 2, fillvalue=""))
@ -362,7 +311,7 @@ TOKEN = config['Discord']['Token']
engine_arg = get_engine_arg(config)
database_interface = DiscordDatabaseInterface(engine_arg)
bot_commands = Commands(engine_arg)
bot.run(TOKEN)

143
test_commands.py 100644
View File

@ -0,0 +1,143 @@
from unittest import TestCase
from BotConfig import *
from Commands import *
from BotErrors import *
class TestCommands(TestCase):
def setUp(self):
config = read_config()
engine_arg = config['SQL']['test_args']
self.commands = Commands(engine_arg)
self.session = self.commands.interface.database.Session()
self.commands.interface.database.clear_all(self.session)
self.session.close()
def test_get_player(self):
session = self.commands.interface.database.Session()
self.commands.interface.add_player(session, 'ZeroHD', discord_uuid='143072699567177728')
player = self.commands.get_player(session, discord_uuid='143072699567177728')
self.assertEqual(player.name, 'ZeroHD')
self.session.close()
def test_register(self):
self.commands.register('ZeroHD', '143072699567177728')
player = self.commands.get_player(self.session, discord_uuid='143072699567177728')
self.assertEqual(player.name, 'ZeroHD')
def test_addbase(self):
player_name = self.commands.register('ZeroHD', '143072699567177728')
base = self.commands.addbase(0, 0, discord_uuid='143072699567177728')
if player_name not in base:
self.fail()
else:
pass
def test_addshop(self):
player_name = self.commands.register('ZeroHD', '143072699567177728')
shop = self.commands.addshop(0, 0, discord_uuid='143072699567177728')
if player_name not in shop:
self.fail()
else:
pass
def test_tunnel(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, shop_name='test shop', discord_uuid='143072699567177728')
tunnel1 = self.commands.tunnel('green', 50, None, discord_uuid='143072699567177728')
self.assertGreater(len(tunnel1), 0)
tunnel2 = self.commands.tunnel('Green', 50, location_name='test_shop', discord_uuid='143072699567177728')
if 'Green' not in tunnel2:
self.fail()
else:
pass
def test_find(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, shop_name='frick', discord_uuid='143072699567177728')
self.commands.addbase(0, 0, 'heck', discord_uuid='143072699567177728')
result = self.commands.find('zerohd')
if ('frick' in result) & ('heck' in result):
pass
else:
self.fail()
def test_delete(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, shop_name='frick', discord_uuid='143072699567177728')
self.commands.delete('frick', discord_uuid='143072699567177728')
self.assertRaises(LocationLookUpError, self.commands.find, 'zerohd')
def test_findaround(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, shop_name='frick', discord_uuid='143072699567177728')
result = self.commands.findaround(0, 0)
if 'frick' in result:
pass
else:
self.fail()
def test_additem(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, discord_uuid='143072699567177728')
result = self.commands.additem('dirt', 5, 5, None, discord_uuid='143072699567177728')
if 'dirt' in result:
pass
else:
self.fail()
self.commands.addshop(0, 0, shop_name='frick', discord_uuid='143072699567177728')
result = self.commands.additem('cool', 5, 5, shop_name='frick', discord_uuid='143072699567177728')
if 'cool' in result:
pass
else:
self.fail()
def test_selling(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, shop_name='frick', discord_uuid='143072699567177728')
self.commands.additem('cool', 5, 5, shop_name='frick', discord_uuid='143072699567177728')
result = self.commands.selling('cool')
if 'cool' in result:
pass
else:
self.fail()
def test_info(self):
self.commands.register('ZeroHD', '143072699567177728')
self.commands.addshop(0, 0, shop_name='frick', discord_uuid='143072699567177728')
self.commands.tunnel('Green', 50, location_name='frick', discord_uuid='143072699567177728')
result = self.commands.info('frick')
if 'Green' in result:
pass
else:
self.fail()

View File

@ -1,5 +1,5 @@
from unittest import TestCase
from DiscordDatabaseInterface import *
from Commands import *
from BotErrors import *
from MinecraftAccountInfoGrabber import *
from BotConfig import *
@ -10,212 +10,212 @@ class TestGeoffreyDatabase(TestCase):
config = read_config()
engine_arg = config['SQL']['test_args']
self.interface = DiscordDatabaseInterface(engine_arg)
self.session = self.interface.database.Session()
self.commands = Commands(engine_arg)
self.owner = Player('ZeroHD', '143072699567177728')
self.loc = Location('test', 1, 3, self.owner, dimension='Nether')
self.tunnel = Tunnel(self.owner, 'Green', 105, self.loc)
self.session = self.commands.interface.database.Session()
self.commands.interface.database.clear_all(self.session)
def tearDown(self):
self.interface.database.clear_all(self.session)
self.session.close()
def add_shop(self, player):
return self.commands.interface.add_shop(self.session, player, 'test', 1, 3, "nether")
def add_player(self):
return self.commands.interface.add_player(self.session, 'ZeroHD', discord_uuid='143072699567177728')
def add_loc(self, player):
return self.commands.interface.add_location(self.session, player, 'test', 0, 0)
def test_add_object(self):
self.interface.database.add_object(self.session, self.loc)
self.interface.database.add_object(self.session, self.owner)
self.interface.database.add_object(self.session, self.tunnel)
self.commands.interface.database.add_object(self.session, self.loc)
self.commands.interface.database.add_object(self.session, self.owner)
self.commands.interface.database.add_object(self.session, self.tunnel)
uuid = grab_UUID('ZeroHD')
expr = Player.mc_uuid == uuid
p = self.interface.database.query_by_filter(self.session, Player, expr)[0]
p = self.commands.interface.database.query_by_filter(self.session, Player, expr)[0]
expr = Location.owner == p
loc2 = self.interface.database.query_by_filter(self.session, Location, expr)[0]
loc2 = self.commands.interface.database.query_by_filter(self.session, Location, expr)[0]
self.assertEqual(self.loc.id, loc2.id)
def test_query_by_filter(self):
self.interface.database.add_object(self.session, self.loc)
self.interface.database.add_object(self.session, self.owner)
self.commands.interface.database.add_object(self.session, self.loc)
self.commands.interface.database.add_object(self.session, self.owner)
expr = (Location.owner == self.owner)
loc2 = self.interface.database.query_by_filter(self.session, Location, expr)[0]
loc2 = self.commands.interface.database.query_by_filter(self.session, Location, expr)[0]
self.assertEqual(loc2.id, self.loc.id)
def test_delete_entry(self):
self.interface.database.add_object(self.session, self.loc)
self.interface.database.add_object(self.session, self.owner)
self.commands.interface.database.add_object(self.session, self.loc)
self.commands.interface.database.add_object(self.session, self.owner)
expr = Location.owner == self.owner
self.interface.database.delete_entry(self.session, Location, expr)
self.commands.interface.database.delete_entry(self.session, Location, expr)
loc2 = self.interface.database.query_by_filter(self.session, Location, expr)
loc2 = self.commands.interface.database.query_by_filter(self.session, Location, expr)
self.assertEqual(len(loc2), 0)
self.assertRaises(DeleteEntryError, self.interface.database.delete_entry, self.session, Location, expr)
self.assertRaises(DeleteEntryError, self.commands.interface.database.delete_entry, self.session, Location, expr)
def test_add_shop(self):
owner = self.add_player()
shop = self.add_shop()
shop = self.add_shop(owner)
self.assertEqual(type(shop), Shop)
shop_list = self.interface.find_shop_by_name(self.session, 'test')
shop_list = self.commands.interface.find_shop_by_name(self.session, 'test')
self.assertEqual(shop_list[0].dimension, shop.dimension)
def add_shop(self):
return self.interface.add_shop(self.session, '143072699567177728', 'test', 1, 3, "nether")
def add_player(self):
return self.interface.add_player(self.session, 'ZeroHD', '143072699567177728')
def add_loc(self):
return self.interface.add_location(self.session, '143072699567177728', 'test', 0, 0)
def test_add_two_shops(self):
owner = self.add_player()
shop1 = self.add_shop()
shop2 = self.interface.add_shop(self.session, '143072699567177728', 'no u', 1, 3)
shop1 = self.add_shop(owner)
shop2 = self.commands.interface.add_shop(self.session, owner, 'no u', 1, 3)
loc_list = self.interface.find_location_by_owner_uuid(self.session, '143072699567177728')
loc_list = self.commands.interface.find_location_by_owner(self.session, owner)
self.assertEqual(loc_list[1].id, shop2.id)
def test_add_tunnel(self):
self.add_player()
tunnel1 = self.interface.add_tunnel(self.session, '143072699567177728', 'green', 155, None)
self.session = self.commands.interface.database.Session()
player = self.add_player()
tunnel1 = self.commands.interface.add_tunnel(self.session, player, 'green', 155, None)
tunnel2 = self.interface.find_tunnel_by_owner_name(self.session, 'ZeroHD')[0]
tunnel2 = self.commands.interface.find_tunnel_by_owner_name(self.session, 'ZeroHD')[0]
self.assertEqual(tunnel1, tunnel2)
def test_add_item(self):
owner = self.add_player()
self.add_shop()
self.interface.add_item(self.session, '143072699567177728', 'test', 'dirt', 1, 15)
self.add_shop(owner)
self.commands.interface.add_item(self.session, owner, 'test', 'dirt', 1, 15)
shops = self.interface.find_shop_selling_item(self.session, 'dirt')
shops = self.commands.interface.find_shop_selling_item(self.session, 'dirt')
self.assertGreater(len(shops), 0)
def test_find_location_by_owner(self):
owner = self.add_player()
shop = self.add_shop()
shop = self.add_shop(owner)
loc_list = self.interface.find_location_by_owner(self.session, owner)
loc_list = self.commands.interface.find_location_by_owner(self.session, owner)
self.assertEqual(loc_list[0].id, shop.id)
def test_find_location_by_name_and_owner(self):
owner = self.add_player()
shop = self.add_shop()
shop = self.add_shop(owner)
loc_list = self.interface.find_location_by_name_and_owner_uuid(self.session, '143072699567177728', 'test')
loc_list = self.commands.interface.find_location_by_name_and_owner(self.session, owner, 'test')
self.assertEqual(loc_list[0].id, shop.id)
def test_delete_base(self):
self.session = self.commands.interface.database.Session()
owner = self.add_player()
self.add_loc()
self.add_loc(owner)
self.interface.delete_location(self.session, '143072699567177728', 'test')
self.commands.interface.delete_location(self.session, owner, 'test')
loc_list = self.interface.find_location_by_name_and_owner_uuid(self.session, '143072699567177728', 'test')
loc_list = self.commands.interface.find_location_by_name_and_owner(self.session, owner, 'test')
self.assertEqual(len(loc_list), 0)
self.session.close()
def test_find_location_around(self):
owner = self.add_player()
loc = self.add_loc()
loc = self.add_loc(owner)
dim = "o"
loc_list = self.interface.find_location_around(self.session, 100, 100, 100, dim)
loc_list = self.commands.interface.find_location_around(self.session, 100, 100, 100, dim)
self.assertEqual(loc_list[0].name, loc.name)
self.assertGreater(len(loc_list), 0)
loc_list = self.interface.find_location_around(self.session, 200, 200, 100, dim)
loc_list = self.commands.interface.find_location_around(self.session, 200, 200, 100, dim)
self.assertEqual(len(loc_list), 0)
loc_list = self.interface.find_location_around(self.session, -100, -100, 100, dim)
loc_list = self.commands.interface.find_location_around(self.session, -100, -100, 100, dim)
self.assertEqual(loc_list[0].name, loc.name)
self.assertGreater(len(loc_list), 0)
loc_list = self.interface.find_location_around(self.session, 100, -100, 100, dim)
loc_list = self.commands.interface.find_location_around(self.session, 100, -100, 100, dim)
self.assertEqual(loc_list[0].name, loc.name)
self.assertGreater(len(loc_list), 0)
loc_list = self.interface.find_location_around(self.session, -100, 100, 100, dim)
loc_list = self.commands.interface.find_location_around(self.session, -100, 100, 100, dim)
self.assertEqual(loc_list[0].name, loc.name)
self.assertGreater(len(loc_list), 0)
loc_list = self.interface.find_location_around(self.session, 50, -50, 100, dim)
loc_list = self.commands.interface.find_location_around(self.session, 50, -50, 100, dim)
self.assertEqual(loc_list[0].name, loc.name)
self.assertGreater(len(loc_list), 0)
def test_find_location_by_name(self):
owner = self.add_player()
loc = self.add_loc()
loc = self.add_loc(owner)
loc_list = self.interface.find_location_by_name(self.session, 'test')
loc_list = self.commands.interface.find_location_by_name(self.session, 'test')
self.assertEqual(loc_list[0].name, loc.name)
def test_search_all(self):
owner = self.add_player()
loc = self.add_loc()
loc = self.add_loc(owner)
loc_list = self.interface.search_all_fields(self.session, 'ZeroHD')
loc_list = self.commands.interface.search_all_fields(self.session, 'ZeroHD')
self.assertEqual(type(loc_list), str)
self.assertGreater(len(loc_list), 0)
def test_wrong_case(self):
owner = self.add_player()
loc = self.add_loc()
loc = self.add_loc(owner)
loc_list = self.interface.find_location_by_owner_name(self.session, 'zerohd')
loc_list = self.commands.interface.find_location_by_owner_name(self.session, 'zerohd')
self.assertEqual(loc_list[0].id, loc.id)
self.interface.add_shop(self.session, '143072699567177728', 'testshop', 1, 3, 'neThEr')
self.commands.interface.add_shop(self.session, owner, 'testshop', 1, 3, 'neThEr')
self.interface.add_item(self.session, '143072699567177728', 'testshop', 'dirts', 1, 15)
self.commands.interface.add_item(self.session, owner, 'testshop', 'dirts', 1, 15)
shops = self.interface.find_shop_selling_item(self.session, 'diRt')
shops = self.commands.interface.find_shop_selling_item(self.session, 'diRt')
self.assertGreater(len(shops), 0)
#shops = self.database.find_shop_selling_item('sDirt')
#self.assertEqual(shops[0].name, 'testshop')
loc_list = self.interface.find_location_by_name(self.session, 'TEST')
loc_list = self.commands.interface.find_location_by_name(self.session, 'TEST')
self.assertEqual(loc_list[0].name, 'test')
def test_big_input(self):
owner = self.add_player()
self.assertRaises(StringTooLong, self.interface.add_location, self.session,'143072699567177728',
self.assertRaises(StringTooLong, self.commands.interface.add_location, self.session, owner,
'TTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTT'
'TTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTT'
'TTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTT', 0, 0,)
def test_duplicate_name(self):
self.add_player()
self.add_loc()
owner = self.add_player()
self.add_loc(owner)
self.assertRaises(EntryNameNotUniqueError, self.interface.add_location, self.session,
'143072699567177728', 'test', 0, 0, 0)
self.assertRaises(EntryNameNotUniqueError, self.commands.interface.add_location, self.session,
owner, 'test', 0, 0, 0)
def test_delete_parent(self):
owner = self.add_player()
loc = self.add_shop()
loc = self.add_shop(owner)
self.interface.add_item(self.session, '143072699567177728', 'test', 'dirt', 1, 15)
self.commands.interface.add_item(self.session, owner, 'test', 'dirt', 1, 15)
self.interface.delete_location(self.session, '143072699567177728', 'test')
self.commands.interface.delete_location(self.session, owner, 'test')
shops = self.interface.find_shop_selling_item(self.session, 'dirt')
shops = self.commands.interface.find_shop_selling_item(self.session, 'dirt')
self.assertEqual(len(shops), 0)