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

Compare commits

base: wutta:master
Branches Tags
wutta:master
wutta:v0.1.1
wutta:v0.1.0
..
compare: wutta:v0.1.0
Branches Tags
wutta:master
wutta:v0.1.1
wutta:v0.1.0

No commits in common. "master" and "v0.1.0" have entirely different histories.
master ... v0.1.0

18 changed files with 245 additions and 415 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.pylintrc
View file

@ -1,4 +0,0 @@
# -*- mode: conf; -*-
[MESSAGES CONTROL]
disable=fixme

6
CHANGELOG.md
View file

@ -1,9 +1,3 @@
## v0.1.1 (2025-02-20)
### Fix
- set vendor name, sku when refreshing neworder batch row
## v0.1.0 (2025-01-13)
### Feat

3
README.md
View file

@ -1,7 +1,6 @@
# Sideshow-COREPOS
This package adds [CORE-POS](https://www.core-pos.com/) integration
for [Sideshow](https://pypi.org/project/Sideshow/).
This package adds CORE-POS integration for Sideshow.
Full docs are at https://rattailproject.org/docs/sideshow-corepos/

29
docs/conf.py
View file

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

6
docs/index.rst
View file

@ -8,12 +8,6 @@ This is `Sideshow`_ with integration for `CORE-POS`_.
.. _CORE-POS: https://www.core-pos.com/
.. image:: https://img.shields.io/badge/linting-pylint-yellowgreen
:target: https://github.com/pylint-dev/pylint
.. image:: https://img.shields.io/badge/code%20style-black-000000.svg
:target: https://github.com/psf/black
.. toctree::
:maxdepth: 1

10
pyproject.toml
View file

@ -5,7 +5,7 @@ build-backend = "hatchling.build"
[project]
name = "Sideshow-COREPOS"
version = "0.1.1"
version = "0.1.0"
description = "Case/Special Order Tracker for CORE-POS"
readme = "README.md"
authors = [
@ -27,19 +27,17 @@ classifiers = [
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Topic :: Office/Business",
"Topic :: Software Development :: Libraries :: Python Modules",
]
license = {text = "GNU General Public License v3+"}
requires-python = ">= 3.8"
dependencies = [
"Sideshow>=0.7.1",
"Wutta-COREPOS[web]>=0.3.0",
"Sideshow>=0.3.0",
"Wutta-COREPOS[web]>=0.2.0",
]
[project.optional-dependencies]
docs = ["Sphinx", "furo"]
tests = ["pylint", "pytest", "pytest-cov", "tox"]
tests = ["pytest-cov", "tox"]
[project.entry-points."paste.app_factory"]

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

@ -43,35 +43,28 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
for more info.
"""
def autocomplete_customers_external( # pylint: disable=empty-docstring
self, session, term, user=None
):
def autocomplete_customers_external(self, session, term, user=None):
""" """
corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op()
op_session = corepos.make_session_office_op()
# base query
query = op_session.query(op_model.CustomerClassic).join(
op_model.MemberInfo,
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number,
)
query = op_session.query(op_model.CustomerClassic)\
.join(op_model.MemberInfo,
op_model.MemberInfo.card_number == op_model.CustomerClassic.card_number)
# filter query
criteria = []
for word in term.split():
criteria.append(
sa.or_(
op_model.CustomerClassic.first_name.ilike(f"%{word}%"),
op_model.CustomerClassic.last_name.ilike(f"%{word}%"),
)
)
criteria.append(sa.or_(
op_model.CustomerClassic.first_name.ilike(f'%{word}%'),
op_model.CustomerClassic.last_name.ilike(f'%{word}%')))
query = query.filter(sa.and_(*criteria))
# sort query
query = query.order_by(
op_model.CustomerClassic.first_name, op_model.CustomerClassic.last_name
)
query = query.order_by(op_model.CustomerClassic.first_name,
op_model.CustomerClassic.last_name)
# get data
# TODO: need max_results option
@ -79,41 +72,32 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
# get results
def result(customer):
return {"value": str(customer.card_number), "label": str(customer)}
return {'value': str(customer.card_number),
'label': str(customer)}
results = [result(c) for c in customers]
op_session.close()
return results
def refresh_batch_from_external_customer( # pylint: disable=empty-docstring
self, batch
):
def refresh_batch_from_external_customer(self, batch):
""" """
corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op()
op_session = corepos.make_session_office_op()
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:
customer = (
op_session.query(op_model.CustomerClassic)
.join(
op_model.MemberInfo,
op_model.MemberInfo.card_number
== 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()
)
except orm.exc.NoResultFound as e:
raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}") from e
customer = op_session.query(op_model.CustomerClassic)\
.join(op_model.MemberInfo,
op_model.MemberInfo.card_number == 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()
except orm.exc.NoResultFound:
raise ValueError(f"CORE-POS Customer not found: {batch.customer_id}")
batch.customer_name = str(customer)
batch.phone_number = customer.member_info.phone
@ -121,9 +105,7 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session.close()
def autocomplete_products_external( # pylint: disable=empty-docstring
self, session, term, user=None
):
def autocomplete_products_external(self, session, term, user=None):
""" """
corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op()
@ -135,16 +117,14 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
# filter query
criteria = []
for word in term.split():
criteria.append(
sa.or_(
op_model.Product.brand.ilike(f"%{word}%"),
op_model.Product.description.ilike(f"%{word}%"),
)
)
criteria.append(sa.or_(
op_model.Product.brand.ilike(f'%{word}%'),
op_model.Product.description.ilike(f'%{word}%')))
query = query.filter(sa.and_(*criteria))
# sort query
query = query.order_by(op_model.Product.brand, op_model.Product.description)
query = query.order_by(op_model.Product.brand,
op_model.Product.description)
# get data
# TODO: need max_results option
@ -152,50 +132,43 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
# get results
def result(product):
return {
"value": product.upc,
"label": self.app.make_full_name(
product.brand, product.description, product.size
),
}
return {'value': product.upc,
'label': self.app.make_full_name(product.brand,
product.description,
product.size)}
results = [result(c) for c in products]
op_session.close()
return results
def get_product_info_external( # pylint: disable=empty-docstring
self, session, product_id, user=None
):
def get_product_info_external(self, session, product_id, user=None):
""" """
corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op()
op_session = corepos.make_session_office_op()
try:
product = (
op_session.query(op_model.Product)
.filter(op_model.Product.upc == product_id)
.one()
)
except orm.exc.NoResultFound as e:
raise ValueError(f"CORE-POS Product not found: {product_id}") from e
product = op_session.query(op_model.Product)\
.filter(op_model.Product.upc == product_id)\
.one()
except orm.exc.NoResultFound:
raise ValueError(f"CORE-POS Product not found: {product_id}")
data = {
"product_id": product.upc,
"scancode": product.upc,
"brand_name": product.brand,
"description": product.description,
"size": product.size,
"full_description": self.app.make_full_name(
product.brand, product.description, product.size
),
"weighed": product.scale,
"special_order": False,
"department_id": product.department_number,
"department_name": product.department.name if product.department else None,
"case_size": self.get_case_size_for_external_product(product),
"unit_price_reg": self.get_unit_price_reg_for_external_product(product),
'product_id': product.upc,
'scancode': product.upc,
'brand_name': product.brand,
'description': product.description,
'size': product.size,
'full_description': self.app.make_full_name(product.brand,
product.description,
product.size),
'weighed': product.scale,
'special_order': False,
'department_id': product.department_number,
'department_name': product.department.name if product.department else None,
'case_size': self.get_case_size_for_external_product(product),
'unit_price_reg': self.get_unit_price_reg_for_external_product(product),
# TODO
# 'vendor_name': product.vendor_name,
# 'vendor_item_code': product.vendor_item_code,
@ -204,20 +177,18 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session.close()
return data
def refresh_row_from_external_product(self, row): # pylint: disable=empty-docstring
def refresh_row_from_external_product(self, row):
""" """
corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op()
op_session = corepos.make_session_office_op()
try:
product = (
op_session.query(op_model.Product)
.filter(op_model.Product.upc == row.product_id)
.one()
)
except orm.exc.NoResultFound as e:
raise ValueError(f"CORE-POS Product not found: {row.product_id}") from e
product = op_session.query(op_model.Product)\
.filter(op_model.Product.upc == row.product_id)\
.one()
except orm.exc.NoResultFound:
raise ValueError(f"CORE-POS Product not found: {row.product_id}")
row.product_scancode = product.upc
row.product_brand = product.brand
@ -227,34 +198,20 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
row.department_id = product.department_number
row.department_name = product.department.name if product.department else None
row.special_order = False
row.vendor_name = None
row.vendor_item_code = None
item = product.default_vendor_item
if item:
row.vendor_name = item.vendor.name if item.vendor else None
row.vendor_item_code = item.sku
row.case_size = self.get_case_size_for_external_product(product)
row.unit_cost = product.cost
row.unit_price_reg = self.get_unit_price_reg_for_external_product(product)
op_session.close()
def get_case_size_for_external_product( # pylint: disable=empty-docstring
self, product
):
def get_case_size_for_external_product(self, product):
""" """
if product.vendor_items:
item = product.vendor_items[0]
if item.units is not None:
return decimal.Decimal(f"{item.units:0.4f}")
return None
return decimal.Decimal(f'{item.units:0.4f}')
def get_unit_price_reg_for_external_product( # pylint: disable=empty-docstring
self, product
):
def get_unit_price_reg_for_external_product(self, product):
""" """
if product.normal_price is not None:
return decimal.Decimal(f"{product.normal_price:0.3f}")
return None
return decimal.Decimal(f'{product.normal_price:0.3f}')

17
src/sideshow_corepos/config.py
View file

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

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

@ -25,5 +25,5 @@ Sideshow-COREPOS - Case/Special Order Tracker for CORE-POS
"""
def includeme(config): # pylint: disable=missing-function-docstring
config.include("sideshow_corepos.web.views")
def includeme(config):
config.include('sideshow_corepos.web.views')

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

@ -29,43 +29,39 @@ from wuttaweb import app as base
from wutta_corepos.web.db import CoreOpSession
def main(global_config, **settings): # pylint: disable=unused-argument
def main(global_config, **settings):
"""
Make and return the WSGI app (Paste entry point).
"""
# prefer Sideshow templates over wuttaweb
settings.setdefault(
"mako.directories",
[
"sideshow.web:templates",
"wuttaweb:templates",
],
)
settings.setdefault('mako.directories', [
'sideshow.web:templates',
'wuttaweb:templates',
])
# make config objects
wutta_config = base.make_wutta_config(settings)
pyramid_config = base.make_pyramid_config(settings)
# configure DB sessions
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
pyramid_config.include("sideshow.web")
pyramid_config.include("sideshow_corepos.web")
pyramid_config.include('sideshow.web')
pyramid_config.include('sideshow_corepos.web')
return pyramid_config.make_wsgi_app()
def make_wsgi_app(config=None):
def make_wsgi_app():
"""
Make and return the WSGI app (generic entry point).
"""
return base.make_wsgi_app(main, config=config)
return base.make_wsgi_app(main)
def make_asgi_app(config=None):
def make_asgi_app():
"""
Make and return the ASGI app (generic entry point).
"""
return base.make_asgi_app(main, config=config)
return base.make_asgi_app(main)

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

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

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

@ -27,8 +27,8 @@ This adds config for readonly views for CORE-POS members and products.
"""
def includeme(config): # pylint: disable=missing-function-docstring
def includeme(config):
# CORE-POS views
config.include("wutta_corepos.web.views.corepos.members")
config.include("wutta_corepos.web.views.corepos.products")
config.include('wutta_corepos.web.views.corepos.members')
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
"""
if not skip_tests:
c.run("pytest")
c.run('pytest')
# rebuild pkg
if os.path.exists("dist"):
shutil.rmtree("dist")
if os.path.exists("Sideshow_COREPOS.egg-info"):
shutil.rmtree("Sideshow_COREPOS.egg-info")
c.run("python -m build --sdist")
if os.path.exists('dist'):
shutil.rmtree('dist')
if os.path.exists('Sideshow_COREPOS.egg-info'):
shutil.rmtree('Sideshow_COREPOS.egg-info')
c.run('python -m build --sdist')
# upload
c.run("twine upload dist/*")
c.run('twine upload dist/*')

207
tests/batch/test_neworder.py
View file

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

20
tests/test_config.py
View file

@ -1,20 +0,0 @@
# -*- coding: utf-8; -*-
from wuttjamaican.testing import ConfigTestCase
from sideshow_corepos import config as mod
class TestSideshowCoreposConfig(ConfigTestCase):
def test_basic(self):
self.assertIsNone(self.config.get("wutta.batch.neworder.handler.spec"))
ext = mod.SideshowCoreposConfig()
ext.configure(self.config)
self.assertEqual(
self.config.get("wutta.batch.neworder.handler.spec"),
"sideshow_corepos.batch.neworder:NewOrderBatchHandler",
)

28
tests/web/test_app.py
View file

@ -1,44 +1,32 @@
# -*- coding: utf-8; -*-
import sqlalchemy as sa
from wuttjamaican.testing import DataTestCase
from wuttjamaican.testing import FileTestCase, ConfigTestCase
from asgiref.wsgi import WsgiToAsgi
from pyramid.router import Router
from wutta_corepos.web.db import CoreOpSession
from sideshow_corepos.web import app as mod
class TestMain(DataTestCase):
class TestMain(FileTestCase):
def test_basic(self):
global_config = None
settings = {"wutta_config": self.config}
myconf = self.write_file('my.conf', '')
settings = {'wutta.config': myconf}
app = mod.main(global_config, **settings)
self.assertIsInstance(app, Router)
self.assertIsNone(CoreOpSession.session_factory.kw["bind"])
def test_corepos_engine(self):
engine = sa.create_engine("sqlite://")
self.config.core_office_op_engine = engine
settings = {"wutta_config": self.config}
app = mod.main(None, **settings)
self.assertIsInstance(app, Router)
self.assertIs(CoreOpSession.session_factory.kw["bind"], engine)
class TestMakeWsgiApp(DataTestCase):
class TestMakeWsgiApp(ConfigTestCase):
def test_basic(self):
wsgi = mod.make_wsgi_app(config=self.config)
wsgi = mod.make_wsgi_app()
self.assertIsInstance(wsgi, Router)
class TestMakeAsgiApp(DataTestCase):
class TestMakeAsgiApp(ConfigTestCase):
def test_basic(self):
asgi = mod.make_asgi_app(config=self.config)
asgi = mod.make_asgi_app()
self.assertIsInstance(asgi, WsgiToAsgi)

55
tests/web/test_menus.py
View file

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

4
tox.ini
View file

@ -6,10 +6,6 @@ envlist = py38, py39, py310, py311
extras = tests
commands = pytest {posargs}
[testenv:pylint]
basepython = python3.11
commands = pylint sideshow_corepos
[testenv:coverage]
basepython = python3.11
commands = pytest --cov=sideshow_corepos --cov-report=html --cov-fail-under=100