Compare commits

..

9 commits

Author SHA1 Message Date
b9a34003c1 fix: address all warnings from pylint 2025-09-01 19:56:16 -05:00
ff573cc17d docs: add badge for black code style 2025-08-31 13:29:01 -05:00
a0b36a6384 fix: format all code with black
and from now on should not deviate from that...
2025-08-31 13:02:02 -05:00
bad6ded72d fix: allow config injection for sake of tests
per changes in wuttaweb, sideshow
2025-07-06 14:09:11 -05:00
198c3f77bb bump: version 0.1.0 → 0.1.1 2025-02-20 09:34:41 -06:00
c9d093d343 fix: set vendor name, sku when refreshing neworder batch row 2025-02-20 09:02:26 -06:00
5d3820a8f2 docs: update intersphinx doc links per server migration 2025-02-18 12:15:27 -06:00
Lance Edgar
7721ddf11e docs: remove bad version string 2025-01-13 20:37:53 -06:00
Lance Edgar
ebb9211f2d docs: tweak readme, project trove classifiers 2025-01-13 17:03:32 -06:00
18 changed files with 415 additions and 245 deletions

4
.pylintrc Normal file
View file

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

View file

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

View file

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

View file

@ -8,33 +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 = '0.1' 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://rattailproject.org/docs/sideshow/', None), "sideshow": ("https://docs.wuttaproject.org/sideshow/", None),
'wuttjamaican': ('https://rattailproject.org/docs/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"]

View file

@ -8,6 +8,12 @@ This is `Sideshow`_ with integration for `CORE-POS`_.
.. _CORE-POS: https://www.core-pos.com/ .. _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:: .. toctree::
:maxdepth: 1 :maxdepth: 1

View file

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

View file

@ -43,28 +43,35 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
for more info. for more info.
""" """
def autocomplete_customers_external(self, session, term, user=None): def autocomplete_customers_external( # pylint: disable=empty-docstring
self, session, term, user=None
):
""" """ """ """
corepos = self.app.get_corepos_handler() corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op() op_model = corepos.get_model_office_op()
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,32 +79,41 @@ 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()
return results return results
def refresh_batch_from_external_customer(self, batch): def refresh_batch_from_external_customer( # pylint: disable=empty-docstring
self, batch
):
""" """ """ """
corepos = self.app.get_corepos_handler() corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op() op_model = corepos.get_model_office_op()
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,
.one() )
except orm.exc.NoResultFound: .filter(op_model.CustomerClassic.card_number == int(batch.customer_id))
raise ValueError(f"CORE-POS Customer not found: {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
batch.customer_name = str(customer) batch.customer_name = str(customer)
batch.phone_number = customer.member_info.phone batch.phone_number = customer.member_info.phone
@ -105,7 +121,9 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session.close() op_session.close()
def autocomplete_products_external(self, session, term, user=None): def autocomplete_products_external( # pylint: disable=empty-docstring
self, session, term, user=None
):
""" """ """ """
corepos = self.app.get_corepos_handler() corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op() op_model = corepos.get_model_office_op()
@ -117,14 +135,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,43 +152,50 @@ 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()
return results return results
def get_product_info_external(self, session, product_id, user=None): def get_product_info_external( # pylint: disable=empty-docstring
self, session, product_id, user=None
):
""" """ """ """
corepos = self.app.get_corepos_handler() corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op() op_model = corepos.get_model_office_op()
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)
.one() .filter(op_model.Product.upc == product_id)
except orm.exc.NoResultFound: .one()
raise ValueError(f"CORE-POS Product not found: {product_id}") )
except orm.exc.NoResultFound as e:
raise ValueError(f"CORE-POS Product not found: {product_id}") from e
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,
@ -177,18 +204,20 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
op_session.close() op_session.close()
return data return data
def refresh_row_from_external_product(self, row): def refresh_row_from_external_product(self, row): # pylint: disable=empty-docstring
""" """ """ """
corepos = self.app.get_corepos_handler() corepos = self.app.get_corepos_handler()
op_model = corepos.get_model_office_op() op_model = corepos.get_model_office_op()
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)
.one() .filter(op_model.Product.upc == row.product_id)
except orm.exc.NoResultFound: .one()
raise ValueError(f"CORE-POS Product not found: {row.product_id}") )
except orm.exc.NoResultFound as e:
raise ValueError(f"CORE-POS Product not found: {row.product_id}") from e
row.product_scancode = product.upc row.product_scancode = product.upc
row.product_brand = product.brand row.product_brand = product.brand
@ -198,20 +227,34 @@ class NewOrderBatchHandler(base.NewOrderBatchHandler):
row.department_id = product.department_number row.department_id = product.department_number
row.department_name = product.department.name if product.department else None row.department_name = product.department.name if product.department else None
row.special_order = False 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.case_size = self.get_case_size_for_external_product(product)
row.unit_cost = product.cost row.unit_cost = product.cost
row.unit_price_reg = self.get_unit_price_reg_for_external_product(product) row.unit_price_reg = self.get_unit_price_reg_for_external_product(product)
op_session.close() op_session.close()
def get_case_size_for_external_product(self, product): def get_case_size_for_external_product( # pylint: disable=empty-docstring
self, product
):
""" """ """ """
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}")
return None
def get_unit_price_reg_for_external_product(self, product): def get_unit_price_reg_for_external_product( # pylint: disable=empty-docstring
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}")
return None

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'
def configure(self, config): key = "sideshow_corepos"
def configure(self, config): # pylint: disable=empty-docstring
""" """ """ """
# 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",
)

View file

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

View file

@ -29,39 +29,43 @@ from wuttaweb import app as base
from wutta_corepos.web.db import CoreOpSession from wutta_corepos.web.db import CoreOpSession
def main(global_config, **settings): def main(global_config, **settings): # pylint: disable=unused-argument
""" """
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
CoreOpSession.configure(bind=wutta_config.core_office_op_engine) if hasattr(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()
def make_wsgi_app(): def make_wsgi_app(config=None):
""" """
Make and return the WSGI app (generic entry point). Make and return the WSGI app (generic entry point).
""" """
return base.make_wsgi_app(main) return base.make_wsgi_app(main, config=config)
def make_asgi_app(): def make_asgi_app(config=None):
""" """
Make and return the ASGI app (generic entry point). Make and return the ASGI app (generic entry point).
""" """
return base.make_asgi_app(main) return base.make_asgi_app(main, config=config)

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", {
'route': 'corepos_members', "title": "CORE-POS Members",
'perm': 'corepos_members.list', "route": "corepos_members",
}, "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", {
'route': 'corepos_products', "title": "CORE-POS Products",
'perm': 'corepos_products.list', "route": "corepos_products",
}, "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", {
'url': url, "title": "CORE Office",
'target': '_blank', "url": url,
}, "target": "_blank",
]) },
]
)
return menu return menu

View file

@ -27,8 +27,8 @@ This adds config for readonly views for CORE-POS members and products.
""" """
def includeme(config): def includeme(config): # pylint: disable=missing-function-docstring
# 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")

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/*")

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,9 +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_item_code, "1234")

20
tests/test_config.py Normal file
View file

@ -0,0 +1,20 @@
# -*- 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",
)

View file

@ -1,32 +1,44 @@
# -*- coding: utf-8; -*- # -*- coding: utf-8; -*-
from wuttjamaican.testing import FileTestCase, ConfigTestCase import sqlalchemy as sa
from wuttjamaican.testing import DataTestCase
from asgiref.wsgi import WsgiToAsgi from asgiref.wsgi import WsgiToAsgi
from pyramid.router import Router from pyramid.router import Router
from wutta_corepos.web.db import CoreOpSession
from sideshow_corepos.web import app as mod from sideshow_corepos.web import app as mod
class TestMain(FileTestCase): class TestMain(DataTestCase):
def test_basic(self): def test_basic(self):
global_config = None global_config = None
myconf = self.write_file('my.conf', '') settings = {"wutta_config": self.config}
settings = {'wutta.config': myconf}
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"])
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(ConfigTestCase): class TestMakeWsgiApp(DataTestCase):
def test_basic(self): def test_basic(self):
wsgi = mod.make_wsgi_app() wsgi = mod.make_wsgi_app(config=self.config)
self.assertIsInstance(wsgi, Router) self.assertIsInstance(wsgi, Router)
class TestMakeAsgiApp(ConfigTestCase): class TestMakeAsgiApp(DataTestCase):
def test_basic(self): def test_basic(self):
asgi = mod.make_asgi_app() asgi = mod.make_asgi_app(config=self.config)
self.assertIsInstance(asgi, WsgiToAsgi) self.assertIsInstance(asgi, WsgiToAsgi)

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,
# nb. trailing slash gets stripped {
'url': 'http://localhost/fannie', "title": "CORE Office",
'target': '_blank', # nb. trailing slash gets stripped
}) "url": "http://localhost/fannie",
"target": "_blank",
},
)

View file

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