wutta/sideshow-corepos
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: format all code with black

Browse source 
and from now on should not deviate from that...
This commit is contained in:
Lance Edgar 2025-08-31 13:02:02 -05:00
parent bad6ded72d
commit a0b36a6384
12 changed files with 313 additions and 220 deletions
Download patch file Download diff file Expand all files Collapse all files

28
docs/conf.py
View file

@ -8,32 +8,32 @@
from importlib.metadata import version as get_version from importlib.metadata import version as get_version
project = 'Sideshow-COREPOS' project = "Sideshow-COREPOS"
copyright = '2025, Lance Edgar' copyright = "2025, Lance Edgar"
author = 'Lance Edgar' author = "Lance Edgar"
release = get_version('Sideshow-COREPOS') release = get_version("Sideshow-COREPOS")
# -- General configuration --------------------------------------------------- # -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
extensions = [ extensions = [
'sphinx.ext.autodoc', "sphinx.ext.autodoc",
'sphinx.ext.intersphinx', "sphinx.ext.intersphinx",
'sphinx.ext.viewcode', "sphinx.ext.viewcode",
'sphinx.ext.todo', "sphinx.ext.todo",
] ]
templates_path = ['_templates'] templates_path = ["_templates"]
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store'] exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
intersphinx_mapping = { intersphinx_mapping = {
'sideshow': ('https://docs.wuttaproject.org/sideshow/', None), "sideshow": ("https://docs.wuttaproject.org/sideshow/", None),
'wuttjamaican': ('https://docs.wuttaproject.org/wuttjamaican/', None), "wuttjamaican": ("https://docs.wuttaproject.org/wuttjamaican/", None),
} }
# -- Options for HTML output ------------------------------------------------- # -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
html_theme = 'furo' html_theme = "furo"
html_static_path = ['_static'] html_static_path = ["_static"]

113
src/sideshow_corepos/batch/neworder.py
View file

@ -50,21 +50,26 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session = corepos.make_session_office_op() op_session = corepos.make_session_office_op()
# base query # base query
query = op_session.query(op_model.CustomerClassic)\ query = op_session.query(op_model.CustomerClassic).join(
.join(op_model.MemberInfo, op_model.MemberInfo,
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number) op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number,
)
# filter query # filter query
criteria = [] criteria = []
for word in term.split(): for word in term.split():
criteria.append(sa.or_( criteria.append(
op_model.CustomerClassic.first_name.ilike(f'%{word}%'), sa.or_(
op_model.CustomerClassic.last_name.ilike(f'%{word}%'))) op_model.CustomerClassic.first_name.ilike(f"%{word}%"),
op_model.CustomerClassic.last_name.ilike(f"%{word}%"),
)
)
query = query.filter(sa.and_(*criteria)) query = query.filter(sa.and_(*criteria))
# sort query # sort query
query = query.order_by(op_model.CustomerClassic.first_name, query = query.order_by(
op_model.CustomerClassic.last_name) op_model.CustomerClassic.first_name, op_model.CustomerClassic.last_name
)
# get data # get data
# TODO: need max_results option # TODO: need max_results option
@ -72,8 +77,8 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
# get results # get results
def result(customer): def result(customer):
return {'value': str(customer.card_number), return {"value": str(customer.card_number), "label": str(customer)}
'label': str(customer)}
results = [result(c) for c in customers] results = [result(c) for c in customers]
op_session.close() op_session.close()
@ -86,16 +91,23 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session = corepos.make_session_office_op() op_session = corepos.make_session_office_op()
if not batch.customer_id.isdigit(): if not batch.customer_id.isdigit():
raise ValueError(f"invalid CORE-POS customer card number: {batch.customer_id}") raise ValueError(
f"invalid CORE-POS customer card number: {batch.customer_id}"
)
try: try:
customer = op_session.query(op_model.CustomerClassic)\ customer = (
.join(op_model.MemberInfo, op_session.query(op_model.CustomerClassic)
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number)\ .join(
.filter(op_model.CustomerClassic.card_number == int(batch.customer_id))\ op_model.MemberInfo,
.filter(op_model.CustomerClassic.person_number == 1)\ op_model.MemberInfo.card_number
.options(orm.joinedload(op_model.CustomerClassic.member_info))\ == op_model.CustomerClassic.card_number,
)
.filter(op_model.CustomerClassic.card_number == int(batch.customer_id))
.filter(op_model.CustomerClassic.person_number == 1)
.options(orm.joinedload(op_model.CustomerClassic.member_info))
.one() .one()
)
except orm.exc.NoResultFound: except orm.exc.NoResultFound:
raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}") raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}")
@ -117,14 +129,16 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
# filter query # filter query
criteria = [] criteria = []
for word in term.split(): for word in term.split():
criteria.append(sa.or_( criteria.append(
op_model.Product.brand.ilike(f'%{word}%'), sa.or_(
op_model.Product.description.ilike(f'%{word}%'))) op_model.Product.brand.ilike(f"%{word}%"),
op_model.Product.description.ilike(f"%{word}%"),
)
)
query = query.filter(sa.and_(*criteria)) query = query.filter(sa.and_(*criteria))
# sort query # sort query
query = query.order_by(op_model.Product.brand, query = query.order_by(op_model.Product.brand, op_model.Product.description)
op_model.Product.description)
# get data # get data
# TODO: need max_results option # TODO: need max_results option
@ -132,10 +146,13 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
# get results # get results
def result(product): def result(product):
return {'value': product.upc, return {
'label': self.app.make_full_name(product.brand, "value": product.upc,
product.description, "label": self.app.make_full_name(
product.size)} product.brand, product.description, product.size
),
}
results = [result(c) for c in products] results = [result(c) for c in products]
op_session.close() op_session.close()
@ -148,27 +165,29 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session = corepos.make_session_office_op() op_session = corepos.make_session_office_op()
try: try:
product = op_session.query(op_model.Product)\ product = (
.filter(op_model.Product.upc == product_id)\ op_session.query(op_model.Product)
.filter(op_model.Product.upc == product_id)
.one() .one()
)
except orm.exc.NoResultFound: except orm.exc.NoResultFound:
raise ValueError(f"CORE-POS Product not found: {product_id}") raise ValueError(f"CORE-POS Product not found: {product_id}")
data = { data = {
'product_id': product.upc, "product_id": product.upc,
'scancode': product.upc, "scancode": product.upc,
'brand_name': product.brand, "brand_name": product.brand,
'description': product.description, "description": product.description,
'size': product.size, "size": product.size,
'full_description': self.app.make_full_name(product.brand, "full_description": self.app.make_full_name(
product.description, product.brand, product.description, product.size
product.size), ),
'weighed': product.scale, "weighed": product.scale,
'special_order': False, "special_order": False,
'department_id': product.department_number, "department_id": product.department_number,
'department_name': product.department.name if product.department else None, "department_name": product.department.name if product.department else None,
'case_size': self.get_case_size_for_external_product(product), "case_size": self.get_case_size_for_external_product(product),
'unit_price_reg': self.get_unit_price_reg_for_external_product(product), "unit_price_reg": self.get_unit_price_reg_for_external_product(product),
# TODO # TODO
# 'vendor_name': product.vendor_name, # 'vendor_name': product.vendor_name,
# 'vendor_item_code': product.vendor_item_code, # 'vendor_item_code': product.vendor_item_code,
@ -184,9 +203,11 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session = corepos.make_session_office_op() op_session = corepos.make_session_office_op()
try: try:
product = op_session.query(op_model.Product)\ product = (
.filter(op_model.Product.upc == row.product_id)\ op_session.query(op_model.Product)
.filter(op_model.Product.upc == row.product_id)
.one() .one()
)
except orm.exc.NoResultFound: except orm.exc.NoResultFound:
raise ValueError(f"CORE-POS Product not found: {row.product_id}") raise ValueError(f"CORE-POS Product not found: {row.product_id}")
@ -217,9 +238,9 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
if product.vendor_items: if product.vendor_items:
item = product.vendor_items[0] item = product.vendor_items[0]
if item.units is not None: if item.units is not None:
return decimal.Decimal(f'{item.units:0.4f}') return decimal.Decimal(f"{item.units:0.4f}")
def get_unit_price_reg_for_external_product(self, product): def get_unit_price_reg_for_external_product(self, product):
""" """ """ """
if product.normal_price is not None: if product.normal_price is not None:
return decimal.Decimal(f'{product.normal_price:0.3f}') return decimal.Decimal(f"{product.normal_price:0.3f}")

15
src/sideshow_corepos/config.py
View file

@ -33,15 +33,20 @@ class SideshowCoreposConfig(WuttaConfigExtension):
This establishes some config defaults specific to Sideshow-COREPOS. This establishes some config defaults specific to Sideshow-COREPOS.
""" """
key = 'sideshow_corepos'
key = "sideshow_corepos"
def configure(self, config): def configure(self, config):
""" """ """ """
# batch handlers # batch handlers
config.setdefault(f'{config.appname}.batch.neworder.handler.spec', config.setdefault(
'sideshow_corepos.batch.neworder:NewOrderBatchHandler') f"{config.appname}.batch.neworder.handler.spec",
"sideshow_corepos.batch.neworder:NewOrderBatchHandler",
)
# web app menu # web app menu
config.setdefault(f'{config.appname}.web.menus.handler.spec', config.setdefault(
'sideshow_corepos.web.menus:SideshowMenuHandler') f"{config.appname}.web.menus.handler.spec",
"sideshow_corepos.web.menus:SideshowMenuHandler",
)

2
src/sideshow_corepos/web/__init__.py
View file

@ -26,4 +26,4 @@ Sideshow-COREPOS - Case/Special Order Tracker for CORE-POS
def includeme(config): def includeme(config):
config.include('sideshow_corepos.web.views') config.include("sideshow_corepos.web.views")

17
src/sideshow_corepos/web/app.py
View file

@ -34,22 +34,25 @@ def main(global_config, **settings):
Make and return the WSGI app (Paste entry point). Make and return the WSGI app (Paste entry point).
""" """
# prefer Sideshow templates over wuttaweb # prefer Sideshow templates over wuttaweb
settings.setdefault('mako.directories', [ settings.setdefault(
'sideshow.web:templates', "mako.directories",
'wuttaweb:templates', [
]) "sideshow.web:templates",
"wuttaweb:templates",
],
)
# make config objects # make config objects
wutta_config = base.make_wutta_config(settings) wutta_config = base.make_wutta_config(settings)
pyramid_config = base.make_pyramid_config(settings) pyramid_config = base.make_pyramid_config(settings)
# configure DB sessions # configure DB sessions
if hasattr(wutta_config, 'core_office_op_engine'): if hasattr(wutta_config, "core_office_op_engine"):
CoreOpSession.configure(bind=wutta_config.core_office_op_engine) CoreOpSession.configure(bind=wutta_config.core_office_op_engine)
# bring in the rest of Sideshow # bring in the rest of Sideshow
pyramid_config.include('sideshow.web') pyramid_config.include("sideshow.web")
pyramid_config.include('sideshow_corepos.web') pyramid_config.include("sideshow_corepos.web")
return pyramid_config.make_wsgi_app() return pyramid_config.make_wsgi_app()

40
src/sideshow_corepos/web/menus.py
View file

@ -38,14 +38,16 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
""" """
menu = super().make_customers_menu(request, **kwargs) menu = super().make_customers_menu(request, **kwargs)
menu['items'].extend([ menu["items"].extend(
{'type': 'sep'}, [
{"type": "sep"},
{ {
'title': "CORE-POS Members", "title": "CORE-POS Members",
'route': 'corepos_members', "route": "corepos_members",
'perm': 'corepos_members.list', "perm": "corepos_members.list",
}, },
]) ]
)
return menu return menu
@ -55,14 +57,16 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
""" """
menu = super().make_products_menu(request, **kwargs) menu = super().make_products_menu(request, **kwargs)
menu['items'].extend([ menu["items"].extend(
{'type': 'sep'}, [
{"type": "sep"},
{ {
'title': "CORE-POS Products", "title": "CORE-POS Products",
'route': 'corepos_products', "route": "corepos_products",
'perm': 'corepos_products.list', "perm": "corepos_products.list",
}, },
]) ]
)
return menu return menu
@ -75,12 +79,14 @@ class SideshowMenuHandler(base.SideshowMenuHandler):
corepos = self.app.get_corepos_handler() corepos = self.app.get_corepos_handler()
url = corepos.get_office_url() url = corepos.get_office_url()
if url: if url:
menu['items'].extend([ menu["items"].extend(
[
{ {
'title': "CORE Office", "title": "CORE Office",
'url': url, "url": url,
'target': '_blank', "target": "_blank",
}, },
]) ]
)
return menu return menu

4
src/sideshow_corepos/web/views.py
View file

@ -30,5 +30,5 @@ This adds config for readonly views for CORE-POS members and products.
def includeme(config): def includeme(config):
# CORE-POS views # CORE-POS views
config.include('wutta_corepos.web.views.corepos.members') config.include("wutta_corepos.web.views.corepos.members")
config.include('wutta_corepos.web.views.corepos.products') config.include("wutta_corepos.web.views.corepos.products")

14
tasks.py
View file

@ -15,14 +15,14 @@ def release(c, skip_tests=False):
Release a new version of Sideshow-COREPOS Release a new version of Sideshow-COREPOS
""" """
if not skip_tests: if not skip_tests:
c.run('pytest') c.run("pytest")
# rebuild pkg # rebuild pkg
if os.path.exists('dist'): if os.path.exists("dist"):
shutil.rmtree('dist') shutil.rmtree("dist")
if os.path.exists('Sideshow_COREPOS.egg-info'): if os.path.exists("Sideshow_COREPOS.egg-info"):
shutil.rmtree('Sideshow_COREPOS.egg-info') shutil.rmtree("Sideshow_COREPOS.egg-info")
c.run('python -m build --sdist') c.run("python -m build --sdist")
# upload # upload
c.run('twine upload dist/*') c.run("twine upload dist/*")

209
tests/batch/test_neworder.py
View file

@ -17,8 +17,8 @@ class TestNewOrderBatchHandler(DataTestCase):
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.op_engine = sa.create_engine('sqlite://') self.op_engine = sa.create_engine("sqlite://")
self.config.core_office_op_engines = {'default': self.op_engine} self.config.core_office_op_engines = {"default": self.op_engine}
self.config.core_office_op_engine = self.op_engine self.config.core_office_op_engine = self.op_engine
op_model.Base.metadata.create_all(bind=self.op_engine) op_model.Base.metadata.create_all(bind=self.op_engine)
@ -31,7 +31,7 @@ class TestNewOrderBatchHandler(DataTestCase):
def make_config(self, **kwargs): def make_config(self, **kwargs):
config = super().make_config(**kwargs) config = super().make_config(**kwargs)
config.setdefault('wutta.enum_spec', 'sideshow.enum') config.setdefault("wutta.enum_spec", "sideshow.enum")
return config return config
def make_handler(self): def make_handler(self):
@ -41,33 +41,41 @@ class TestNewOrderBatchHandler(DataTestCase):
handler = self.make_handler() handler = self.make_handler()
# empty results by default # empty results by default
self.assertEqual(handler.autocomplete_customers_external(self.session, 'foo'), []) self.assertEqual(
handler.autocomplete_customers_external(self.session, "foo"), []
)
# add a member # add a member
member = op_model.MemberInfo(card_number=42) member = op_model.MemberInfo(card_number=42)
self.op_session.add(member) self.op_session.add(member)
customer = op_model.CustomerClassic(first_name="Chuck", last_name="Norris", customer = op_model.CustomerClassic(
last_change=datetime.datetime.now()) first_name="Chuck", last_name="Norris", last_change=datetime.datetime.now()
)
member.customers.append(customer) member.customers.append(customer)
self.op_session.add(customer) self.op_session.add(customer)
self.op_session.flush() self.op_session.flush()
# search for chuck finds chuck # search for chuck finds chuck
results = handler.autocomplete_customers_external(self.session, 'chuck') results = handler.autocomplete_customers_external(self.session, "chuck")
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertEqual(results[0], { self.assertEqual(
'value': '42', results[0],
'label': "Chuck Norris", {
}) "value": "42",
"label": "Chuck Norris",
},
)
# search for sally finds nothing # search for sally finds nothing
self.assertEqual(handler.autocomplete_customers_external(self.session, 'sally'), []) self.assertEqual(
handler.autocomplete_customers_external(self.session, "sally"), []
)
def test_refresh_batch_from_external_customer(self): def test_refresh_batch_from_external_customer(self):
model = self.app.model model = self.app.model
handler = self.make_handler() handler = self.make_handler()
user = model.User(username='barney') user = model.User(username="barney")
self.session.add(user) self.session.add(user)
self.session.flush() self.session.flush()
@ -76,91 +84,114 @@ class TestNewOrderBatchHandler(DataTestCase):
self.session.flush() self.session.flush()
# add a member # add a member
member = op_model.MemberInfo(card_number=42, phone='555-1234', email='chuck@example.com') member = op_model.MemberInfo(
card_number=42, phone="555-1234", email="chuck@example.com"
)
self.op_session.add(member) self.op_session.add(member)
customer = op_model.CustomerClassic(first_name="Chuck", last_name="Norris", customer = op_model.CustomerClassic(
last_change=datetime.datetime.now()) first_name="Chuck", last_name="Norris", last_change=datetime.datetime.now()
)
member.customers.append(customer) member.customers.append(customer)
self.op_session.add(customer) self.op_session.add(customer)
self.op_session.flush() self.op_session.flush()
# error if invalid customer_id # error if invalid customer_id
batch.customer_id = 'BreakThings!' batch.customer_id = "BreakThings!"
self.assertRaises(ValueError, handler.refresh_batch_from_external_customer, batch) self.assertRaises(
ValueError, handler.refresh_batch_from_external_customer, batch
)
# error if customer not found # error if customer not found
batch.customer_id = '9999' batch.customer_id = "9999"
self.assertRaises(ValueError, handler.refresh_batch_from_external_customer, batch) self.assertRaises(
ValueError, handler.refresh_batch_from_external_customer, batch
)
# batch should reflect customer info # batch should reflect customer info
batch.customer_id = '42' batch.customer_id = "42"
self.assertIsNone(batch.customer_name) self.assertIsNone(batch.customer_name)
self.assertIsNone(batch.phone_number) self.assertIsNone(batch.phone_number)
self.assertIsNone(batch.email_address) self.assertIsNone(batch.email_address)
handler.refresh_batch_from_external_customer(batch) handler.refresh_batch_from_external_customer(batch)
self.assertEqual(batch.customer_name, "Chuck Norris") self.assertEqual(batch.customer_name, "Chuck Norris")
self.assertEqual(batch.phone_number, '555-1234') self.assertEqual(batch.phone_number, "555-1234")
self.assertEqual(batch.email_address, 'chuck@example.com') self.assertEqual(batch.email_address, "chuck@example.com")
def test_autocomplete_products_local(self): def test_autocomplete_products_local(self):
handler = self.make_handler() handler = self.make_handler()
# empty results by default # empty results by default
self.assertEqual(handler.autocomplete_products_external(self.session, 'foo'), []) self.assertEqual(
handler.autocomplete_products_external(self.session, "foo"), []
)
# add a product # add a product
product = op_model.Product(upc='07430500132', brand="Bragg's", product = op_model.Product(
description="Vinegar", size='32oz') upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
)
self.op_session.add(product) self.op_session.add(product)
self.op_session.commit() self.op_session.commit()
# search for vinegar finds product # search for vinegar finds product
results = handler.autocomplete_products_external(self.session, 'vinegar') results = handler.autocomplete_products_external(self.session, "vinegar")
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertEqual(results[0], { self.assertEqual(
'value': '07430500132', results[0],
'label': "Bragg's Vinegar 32oz", {
}) "value": "07430500132",
"label": "Bragg's Vinegar 32oz",
},
)
# search for brag finds product # search for brag finds product
results = handler.autocomplete_products_external(self.session, 'brag') results = handler.autocomplete_products_external(self.session, "brag")
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertEqual(results[0], { self.assertEqual(
'value': '07430500132', results[0],
'label': "Bragg's Vinegar 32oz", {
}) "value": "07430500132",
"label": "Bragg's Vinegar 32oz",
},
)
# search for juice finds nothing # search for juice finds nothing
self.assertEqual(handler.autocomplete_products_external(self.session, 'juice'), []) self.assertEqual(
handler.autocomplete_products_external(self.session, "juice"), []
)
def test_get_case_size_for_external_product(self): def test_get_case_size_for_external_product(self):
handler = self.make_handler() handler = self.make_handler()
# null # null
product = op_model.Product(upc='07430500132', brand="Bragg's", product = op_model.Product(
description="Vinegar", size='32oz') upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
)
self.op_session.add(product) self.op_session.add(product)
self.op_session.commit() self.op_session.commit()
self.op_session.refresh(product) self.op_session.refresh(product)
self.assertIsNone(handler.get_case_size_for_external_product(product)) self.assertIsNone(handler.get_case_size_for_external_product(product))
# typical # typical
vendor = op_model.Vendor(id=42, name='Acme Distributors') vendor = op_model.Vendor(id=42, name="Acme Distributors")
self.op_session.add(vendor) self.op_session.add(vendor)
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34, item = op_model.VendorItem(
vendor_item_id=1) vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
)
product.vendor_items.append(item) product.vendor_items.append(item)
self.op_session.commit() self.op_session.commit()
self.op_session.refresh(product) self.op_session.refresh(product)
self.assertEqual(handler.get_case_size_for_external_product(product), self.assertEqual(
decimal.Decimal('12.3400')) handler.get_case_size_for_external_product(product),
decimal.Decimal("12.3400"),
)
def test_get_unit_price_reg_for_external_product(self): def test_get_unit_price_reg_for_external_product(self):
handler = self.make_handler() handler = self.make_handler()
# null # null
product = op_model.Product(upc='07430500132', brand="Bragg's", product = op_model.Product(
description="Vinegar", size='32oz') upc="07430500132", brand="Bragg's", description="Vinegar", size="32oz"
)
self.op_session.add(product) self.op_session.add(product)
self.op_session.commit() self.op_session.commit()
self.op_session.refresh(product) self.op_session.refresh(product)
@ -170,53 +201,64 @@ class TestNewOrderBatchHandler(DataTestCase):
product.normal_price = 4.19 product.normal_price = 4.19
self.op_session.commit() self.op_session.commit()
self.op_session.refresh(product) self.op_session.refresh(product)
self.assertEqual(handler.get_unit_price_reg_for_external_product(product), self.assertEqual(
decimal.Decimal('4.19')) handler.get_unit_price_reg_for_external_product(product),
decimal.Decimal("4.19"),
)
def test_get_product_info_external(self): def test_get_product_info_external(self):
model = self.app.model model = self.app.model
handler = self.make_handler() handler = self.make_handler()
user = model.User(username='barney') user = model.User(username="barney")
self.session.add(user) self.session.add(user)
batch = handler.make_batch(self.session, created_by=user) batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch) self.session.add(batch)
self.session.flush() self.session.flush()
vendor = op_model.Vendor(id=42, name='Acme Distributors') vendor = op_model.Vendor(id=42, name="Acme Distributors")
self.op_session.add(vendor) self.op_session.add(vendor)
product = op_model.Product(upc='07430500132', brand="Bragg", product = op_model.Product(
description="Vinegar", size='32oz', upc="07430500132",
normal_price=4.19) brand="Bragg",
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34, description="Vinegar",
vendor_item_id=1) size="32oz",
normal_price=4.19,
)
item = op_model.VendorItem(
vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
)
product.vendor_items.append(item) product.vendor_items.append(item)
self.op_session.add(product) self.op_session.add(product)
self.op_session.commit() self.op_session.commit()
# typical # typical
info = handler.get_product_info_external(self.session, '07430500132') info = handler.get_product_info_external(self.session, "07430500132")
self.assertEqual(info['product_id'], '07430500132') self.assertEqual(info["product_id"], "07430500132")
self.assertEqual(info['scancode'], '07430500132') self.assertEqual(info["scancode"], "07430500132")
self.assertEqual(info['brand_name'], 'Bragg') self.assertEqual(info["brand_name"], "Bragg")
self.assertEqual(info['description'], 'Vinegar') self.assertEqual(info["description"], "Vinegar")
self.assertEqual(info['size'], '32oz') self.assertEqual(info["size"], "32oz")
self.assertEqual(info['full_description'], 'Bragg Vinegar 32oz') self.assertEqual(info["full_description"], "Bragg Vinegar 32oz")
self.assertEqual(info['case_size'], decimal.Decimal('12.3400')) self.assertEqual(info["case_size"], decimal.Decimal("12.3400"))
self.assertEqual(info['unit_price_reg'], decimal.Decimal('4.19')) self.assertEqual(info["unit_price_reg"], decimal.Decimal("4.19"))
# error if no product_id # error if no product_id
self.assertRaises(ValueError, handler.get_product_info_external, self.session, None) self.assertRaises(
ValueError, handler.get_product_info_external, self.session, None
)
# error if product not found # error if product not found
self.assertRaises(ValueError, handler.get_product_info_external, self.session, 'BADUPC') self.assertRaises(
ValueError, handler.get_product_info_external, self.session, "BADUPC"
)
def test_refresh_row_from_external_product(self): def test_refresh_row_from_external_product(self):
model = self.app.model model = self.app.model
enum = self.app.enum enum = self.app.enum
handler = self.make_handler() handler = self.make_handler()
user = model.User(username='barney') user = model.User(username="barney")
self.session.add(user) self.session.add(user)
batch = handler.make_batch(self.session, created_by=user) batch = handler.make_batch(self.session, created_by=user)
self.session.add(batch) self.session.add(batch)
@ -225,27 +267,32 @@ class TestNewOrderBatchHandler(DataTestCase):
self.session.add(row) self.session.add(row)
self.session.flush() self.session.flush()
vendor = op_model.Vendor(id=42, name='Acme Distributors') vendor = op_model.Vendor(id=42, name="Acme Distributors")
self.op_session.add(vendor) self.op_session.add(vendor)
product = op_model.Product(upc='07430500132', brand="Bragg", product = op_model.Product(
description="Vinegar", size='32oz', upc="07430500132",
normal_price=4.19) brand="Bragg",
item = op_model.VendorItem(vendor=vendor, sku='1234', units=12.34, description="Vinegar",
vendor_item_id=1) size="32oz",
normal_price=4.19,
)
item = op_model.VendorItem(
vendor=vendor, sku="1234", units=12.34, vendor_item_id=1
)
product.vendor_items.append(item) product.vendor_items.append(item)
self.op_session.add(product) self.op_session.add(product)
self.op_session.commit() self.op_session.commit()
# error if invalid product_id # error if invalid product_id
row.product_id = 'BreakThings!' row.product_id = "BreakThings!"
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row) self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
# error if product not found # error if product not found
row.product_id = '9999' row.product_id = "9999"
self.assertRaises(ValueError, handler.refresh_row_from_external_product, row) self.assertRaises(ValueError, handler.refresh_row_from_external_product, row)
# row should reflect product info # row should reflect product info
row.product_id = '07430500132' row.product_id = "07430500132"
self.assertIsNone(row.product_scancode) self.assertIsNone(row.product_scancode)
self.assertIsNone(row.product_brand) self.assertIsNone(row.product_brand)
self.assertIsNone(row.product_description) self.assertIsNone(row.product_description)
@ -253,11 +300,11 @@ class TestNewOrderBatchHandler(DataTestCase):
self.assertIsNone(row.case_size) self.assertIsNone(row.case_size)
self.assertIsNone(row.unit_price_reg) self.assertIsNone(row.unit_price_reg)
handler.refresh_row_from_external_product(row) handler.refresh_row_from_external_product(row)
self.assertEqual(row.product_scancode, '07430500132') self.assertEqual(row.product_scancode, "07430500132")
self.assertEqual(row.product_brand, "Bragg") self.assertEqual(row.product_brand, "Bragg")
self.assertEqual(row.product_description, "Vinegar") self.assertEqual(row.product_description, "Vinegar")
self.assertEqual(row.product_size, "32oz") self.assertEqual(row.product_size, "32oz")
self.assertEqual(row.case_size, decimal.Decimal('12.3400')) self.assertEqual(row.case_size, decimal.Decimal("12.3400"))
self.assertEqual(row.unit_price_reg, decimal.Decimal('4.19')) self.assertEqual(row.unit_price_reg, decimal.Decimal("4.19"))
self.assertEqual(row.vendor_name, 'Acme Distributors') self.assertEqual(row.vendor_name, "Acme Distributors")
self.assertEqual(row.vendor_item_code, '1234') self.assertEqual(row.vendor_item_code, "1234")

8
tests/test_config.py
View file

@ -9,10 +9,12 @@ class TestSideshowCoreposConfig(ConfigTestCase):
def test_basic(self): def test_basic(self):
self.assertIsNone(self.config.get('wutta.batch.neworder.handler.spec')) self.assertIsNone(self.config.get("wutta.batch.neworder.handler.spec"))
ext = mod.SideshowCoreposConfig() ext = mod.SideshowCoreposConfig()
ext.configure(self.config) ext.configure(self.config)
self.assertEqual(self.config.get('wutta.batch.neworder.handler.spec'), self.assertEqual(
'sideshow_corepos.batch.neworder:NewOrderBatchHandler') self.config.get("wutta.batch.neworder.handler.spec"),
"sideshow_corepos.batch.neworder:NewOrderBatchHandler",
)

10
tests/web/test_app.py
View file

@ -16,18 +16,18 @@ class TestMain(DataTestCase):
def test_basic(self): def test_basic(self):
global_config = None global_config = None
settings = {'wutta_config': self.config} settings = {"wutta_config": self.config}
app = mod.main(global_config, **settings) app = mod.main(global_config, **settings)
self.assertIsInstance(app, Router) self.assertIsInstance(app, Router)
self.assertIsNone(CoreOpSession.session_factory.kw['bind']) self.assertIsNone(CoreOpSession.session_factory.kw["bind"])
def test_corepos_engine(self): def test_corepos_engine(self):
engine = sa.create_engine('sqlite://') engine = sa.create_engine("sqlite://")
self.config.core_office_op_engine = engine self.config.core_office_op_engine = engine
settings = {'wutta_config': self.config} settings = {"wutta_config": self.config}
app = mod.main(None, **settings) app = mod.main(None, **settings)
self.assertIsInstance(app, Router) self.assertIsInstance(app, Router)
self.assertIs(CoreOpSession.session_factory.kw['bind'], engine) self.assertIs(CoreOpSession.session_factory.kw["bind"], engine)
class TestMakeWsgiApp(DataTestCase): class TestMakeWsgiApp(DataTestCase):

53
tests/web/test_menus.py
View file

@ -13,39 +13,48 @@ class TestSideshowMenuHandler(WebTestCase):
def test_make_customers_menu(self): def test_make_customers_menu(self):
handler = self.make_handler() handler = self.make_handler()
menu = handler.make_customers_menu(self.request) menu = handler.make_customers_menu(self.request)
item = menu['items'][-1] item = menu["items"][-1]
self.assertEqual(item, { self.assertEqual(
'title': "CORE-POS Members", item,
'route': 'corepos_members', {
'perm': 'corepos_members.list', "title": "CORE-POS Members",
}) "route": "corepos_members",
"perm": "corepos_members.list",
},
)
def test_make_products_menu(self): def test_make_products_menu(self):
handler = self.make_handler() handler = self.make_handler()
menu = handler.make_products_menu(self.request) menu = handler.make_products_menu(self.request)
item = menu['items'][-1] item = menu["items"][-1]
self.assertEqual(item, { self.assertEqual(
'title': "CORE-POS Products", item,
'route': 'corepos_products', {
'perm': 'corepos_products.list', "title": "CORE-POS Products",
}) "route": "corepos_products",
"perm": "corepos_products.list",
},
)
def test_make_other_menu(self): def test_make_other_menu(self):
handler = self.make_handler() handler = self.make_handler()
# no url configured by default # no url configured by default
menu = handler.make_other_menu(self.request) menu = handler.make_other_menu(self.request)
if menu['items']: if menu["items"]:
item = menu['items'][-1] item = menu["items"][-1]
self.assertNotEqual(item['title'], "CORE Office") self.assertNotEqual(item["title"], "CORE Office")
# entry added if url configured # entry added if url configured
self.config.setdefault('corepos.office.url', 'http://localhost/fannie/') self.config.setdefault("corepos.office.url", "http://localhost/fannie/")
menu = handler.make_other_menu(self.request) menu = handler.make_other_menu(self.request)
item = menu['items'][-1] item = menu["items"][-1]
self.assertEqual(item, { self.assertEqual(
'title': "CORE Office", item,
{
"title": "CORE Office",
# nb. trailing slash gets stripped # nb. trailing slash gets stripped
'url': 'http://localhost/fannie', "url": "http://localhost/fannie",
'target': '_blank', "target": "_blank",
}) },
)